fix(postgres): stop a naming convention from qualifying the check constraints twice

docs/agents.md

@@ -366,7 +366,7 @@ handler it installs, for when you assemble the pipeline yourself.
 Abstract ORM bases — bind them to your own `DeclarativeBase`: `OutboxEventDBBase`,
 `InboxEventDBBase`, `OutboxEventPartitionedDBBase`, `InboxEventPartitionedDBBase`, the
 mixin `EventMixin`, and the helpers `get_event_constraints(table_name,
-include_created_at_in_unique=False)` and `UnConstrainedEnum`.
+include_created_at_in_unique=False, *, metadata=None)` and `UnConstrainedEnum`.
 
 `PostgresOutboxRepository(session, *, model_class, conflict_index_id=None,
 conflict_index_idempotency=None, batch_size=1000, error_max_length=2000,
@@ -488,7 +488,11 @@ protocols, not a third implementation. `EventBatchProcessor` sets
     Give the sink an idempotent key.
 17. **The tables are yours.** No `Base`, no migrations, no DDL. Bind the abstract bases to
     your `DeclarativeBase` and generate the migration yourself; the repositories depend on the
-    column names, so keep them.
+    column names, so keep them. The check constraints are named `ck_<table>_<rule>` unless
+    your `MetaData` has a `ck` naming convention with `%(constraint_name)s` in it — then the
+    bases hand it the bare rule (`attempts_valid`, …) and the convention names them, so nothing
+    runs past PostgreSQL's 63 bytes. Composing `__table_args__` yourself, pass
+    `metadata=Base.metadata` to `get_event_constraints` for the same.
 18. **`shutdown_requested_func` stops a batch, it does not abort an event.** It is polled
     before the fetch — returning `True` there locks nothing at all — and again before each
     event. What was already processed is committed; the events left untouched come back in

docs/api_reference.md

@@ -197,7 +197,7 @@ All inherit from `OmniBoxError`.
 
 - ORM bases: `OutboxEventDBBase`, `InboxEventDBBase`, `OutboxEventPartitionedDBBase`, `InboxEventPartitionedDBBase`, plus the underlying `EventMixin`, `OutboxColumnsMixin`, `InboxColumnsMixin`.
 - Repositories: `PostgresOutboxRepository`, `PostgresInboxRepository`, `PostgresEventRepository` (shared base). Both expose `session` — the `AsyncSession` they were built on, read-only — which is how a handler passed to `InboxConsumerRunner` writes its side effects in the transaction that inserts the inbox row.
-- Helpers: `UnConstrainedEnum`, `get_event_constraints(table_name, include_created_at_in_unique=False)`.
+- Helpers: `UnConstrainedEnum`, `get_event_constraints(table_name, include_created_at_in_unique=False, *, metadata=None)`.
 
 ### Kafka (extra: `kafka`)
 

docs/migrations.md

@@ -205,6 +205,28 @@ class InboxEventDB(Base, InboxEventDBBase):
 
 You can override `__tablename__`, add service-specific columns, or change `__inbox_dedup_index_columns__` if you partition.
 
+## Naming conventions
+
+The check constraints are named `ck_<table>_<rule>` — `ck_outbox_events_attempts_valid`, as in the DDL above — unless the `MetaData` of your `DeclarativeBase` carries a `ck` naming convention that interpolates `%(constraint_name)s`. Then the bases hand the convention the bare rule (`attempts_valid`, `completed_status_consistency`, `lock_consistency`) and it builds the name: `sqlalchemy-foundation-kit`'s `"%(table_name)s_%(constraint_name)s_check"` gives `outbox_events_attempts_valid_check`, SQLAlchemy's documented `"ck_%(table_name)s_%(constraint_name)s"` gives `ck_outbox_events_attempts_valid` again. Every name fits PostgreSQL's 63-byte limit either way; the longest, `outbox_events_partitioned_completed_status_consistency_check`, is 60. A finished name handed to such a convention would be qualified a second time, run past 63 bytes and be truncated with a hash. The indexes keep their `idx_<table>_…` names under any convention — SQLAlchemy leaves an explicitly named `Index` alone unless the `ix` rule interpolates `%(constraint_name)s`, which neither of the conventions above does.
+
+If you compose `__table_args__` yourself, pass the metadata so the helper can see the convention:
+
+```python
+class OrdersOutbox(Base, OutboxEventDBBase):
+    __tablename__ = "orders_outbox"
+    __table_args__ = get_event_constraints("orders_outbox", metadata=Base.metadata)
+```
+
+Up to 0.2.1 the bases always declared the finished name, so a database created under such a convention holds the doubly qualified names, six of the twelve truncated with a hash — `outbox_events_ck_outbox_events_completed_status_consist_a784`. Alembic's autogenerate does not compare check constraints unless `alembic.ext.checkconstraint_byname` (Alembic 1.19+) is enabled, so rename them in a hand-written migration. They are the ones with `_ck_` in the middle:
+
+```sql
+SELECT conrelid::regclass, conname FROM pg_constraint WHERE contype = 'c' AND conname LIKE '%\_ck\_%';
+ALTER TABLE outbox_events RENAME CONSTRAINT outbox_events_ck_outbox_events_completed_status_consist_a784
+    TO outbox_events_completed_status_consistency_check;
+```
+
+The four bases also build `__table_args__` in a `declared_attr` now, so a workaround that read the tuple off the abstract class — `OutboxEventPartitionedDBBase.__table_args__`, to wrap the names in `conv()` by hand — raises `AttributeError: type object 'OutboxEventPartitionedDBBase' has no attribute 'metadata'` at import. Delete it; the bases produce the convention's names on their own.
+
 ## Alembic example
 
 ```python

omni_box/infra/storage/postgres/orm.py

@@ -4,18 +4,20 @@
 
 import datetime
 from functools import partial
+from typing import ClassVar
 from uuid import UUID
 
 from sqlalchemy import (
     CheckConstraint,
     DateTime,
     Enum,
     Index,
+    MetaData,
     String,
     text,
 )
 from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, declared_attr, mapped_column
 from sqlalchemy.sql import func
 
 from ....core.constants import DEFAULT_MAX_ATTEMPTS
@@ -36,6 +38,10 @@ class EventMixin:
 
     __abstract__ = True
 
+    # Supplied by the DeclarativeBase a concrete model is bound to; declared so that
+    # __table_args__ can read its naming convention.
+    metadata: ClassVar[MetaData]
+
     # Identifiers
     id: Mapped[UUID] = mapped_column(primary_key=True)
     event_type: Mapped[str] = mapped_column(String(100), nullable=False)
@@ -100,25 +106,45 @@ class InboxColumnsMixin:
     source: Mapped[str] = mapped_column(String(255), nullable=False)
 
 
-def get_event_constraints(table_name: str, include_created_at_in_unique: bool = False) -> tuple:
-    """Generate generic constraints and indexes for event table."""
+def get_event_constraints(
+    table_name: str, include_created_at_in_unique: bool = False, *, metadata: MetaData | None = None
+) -> tuple:
+    """Generate generic constraints and indexes for event table.
+
+    Args:
+        table_name: Prefix of the constraint and index names.
+        include_created_at_in_unique: Add ``created_at`` to the unique index on ``idempotency_key``
+            (a partitioned table requires the partition key in every unique index).
+        metadata: The ``MetaData`` the table is registered in. When its ``ck`` naming convention
+            contains ``%(constraint_name)s`` the check constraints are named by their rule alone
+            (``attempts_valid``, ``completed_status_consistency``, ``lock_consistency``) and the
+            convention qualifies them; otherwise they are named ``ck_<table_name>_<rule>``.
+    """
     idempotency_key_cols = ["idempotency_key"]
     if include_created_at_in_unique:
         idempotency_key_cols.append("created_at")
 
+    # SQLAlchemy applies a convention to an explicitly named constraint only when the rule
+    # interpolates that name (sqlalchemy/sql/naming.py); a finished name would be qualified twice.
+    ck_rule = metadata.naming_convention.get("ck") if metadata is not None else None
+    convention_qualifies = isinstance(ck_rule, str) and "constraint_name" in ck_rule
+
+    def check_name(rule: str) -> str:
+        return rule if convention_qualifies else f"ck_{table_name}_{rule}"
+
     return (
         # Ensure attempts_made never exceeds max_attempts at database level
-        CheckConstraint("attempts_made <= max_attempts", name=f"ck_{table_name}_attempts_valid"),
+        CheckConstraint("attempts_made <= max_attempts", name=check_name("attempts_valid")),
         # Status consistency constraints
         CheckConstraint(
             f"(status = '{EventStatus.COMPLETED.value}' AND completed_at IS NOT NULL) OR "
             f"(status != '{EventStatus.COMPLETED.value}' AND completed_at IS NULL)",
-            name=f"ck_{table_name}_completed_status_consistency",
+            name=check_name("completed_status_consistency"),
         ),
         # Locking consistency: locked_at and locked_by must be both NULL or both NOT NULL
         CheckConstraint(
             "(locked_at IS NULL AND locked_by IS NULL) OR (locked_at IS NOT NULL AND locked_by IS NOT NULL)",
-            name=f"ck_{table_name}_lock_consistency",
+            name=check_name("lock_consistency"),
         ),
         Index(
             f"idx_{table_name}_pending_fetch",
@@ -153,7 +179,10 @@ class OutboxEventDBBase(EventMixin, OutboxColumnsMixin):
 
     __abstract__ = True
     __tablename__ = "outbox_events"
-    __table_args__ = get_event_constraints("outbox_events")
+
+    @declared_attr.directive
+    def __table_args__(self) -> tuple:
+        return get_event_constraints("outbox_events", metadata=self.metadata)
 
 
 class InboxEventDBBase(EventMixin, InboxColumnsMixin):
@@ -165,16 +194,18 @@ class InboxEventDBBase(EventMixin, InboxColumnsMixin):
     # Columns for INSERT ... ON CONFLICT DO NOTHING (PostgresInboxRepository)
     __inbox_dedup_index_columns__: tuple[str, ...] = ("message_id", "consumer_group")
 
-    __table_args__ = (
-        *get_event_constraints("inbox_events"),
-        # Inbox specific unique constraint for deduplication
-        Index(
-            "idx_inbox_deduplication",
-            "message_id",
-            "consumer_group",
-            unique=True,
-        ),
-    )
+    @declared_attr.directive
+    def __table_args__(self) -> tuple:
+        return (
+            *get_event_constraints("inbox_events", metadata=self.metadata),
+            # Inbox specific unique constraint for deduplication
+            Index(
+                "idx_inbox_deduplication",
+                "message_id",
+                "consumer_group",
+                unique=True,
+            ),
+        )
 
 
 class OutboxEventPartitionedDBBase(EventMixin, OutboxColumnsMixin):
@@ -193,10 +224,12 @@ class OutboxEventPartitionedDBBase(EventMixin, OutboxColumnsMixin):
     __outbox_conflict_index_id__ = ("id", "created_at")
     __outbox_conflict_index_idempotency__ = ("idempotency_key", "created_at")
 
-    __table_args__ = (
-        *get_event_constraints("outbox_events_p", include_created_at_in_unique=True),
-        {"postgresql_partition_by": "RANGE (created_at)"},
-    )
+    @declared_attr.directive
+    def __table_args__(self) -> tuple:
+        return (
+            *get_event_constraints("outbox_events_p", include_created_at_in_unique=True, metadata=self.metadata),
+            {"postgresql_partition_by": "RANGE (created_at)"},
+        )
 
 
 class InboxEventPartitionedDBBase(EventMixin, InboxColumnsMixin):
@@ -218,23 +251,25 @@ class InboxEventPartitionedDBBase(EventMixin, InboxColumnsMixin):
 
     __inbox_dedup_index_columns__: tuple[str, ...] = ("message_id", "consumer_group", "created_at")
 
-    __table_args__ = (
-        *get_event_constraints("inbox_events_p", include_created_at_in_unique=True),
-        Index(
-            "idx_inbox_events_p_deduplication",
-            "message_id",
-            "consumer_group",
-            "created_at",
-            unique=True,
-        ),
-        Index(
-            "idx_inbox_events_p_message_consumer",
-            "message_id",
-            "consumer_group",
-            unique=False,
-        ),
-        {"postgresql_partition_by": "RANGE (created_at)"},
-    )
+    @declared_attr.directive
+    def __table_args__(self) -> tuple:
+        return (
+            *get_event_constraints("inbox_events_p", include_created_at_in_unique=True, metadata=self.metadata),
+            Index(
+                "idx_inbox_events_p_deduplication",
+                "message_id",
+                "consumer_group",
+                "created_at",
+                unique=True,
+            ),
+            Index(
+                "idx_inbox_events_p_message_consumer",
+                "message_id",
+                "consumer_group",
+                unique=False,
+            ),
+            {"postgresql_partition_by": "RANGE (created_at)"},
+        )
 
 
 # Type aliases

tests/integration/postgres/test_orm.py

@@ -1,11 +1,30 @@
 import pytest
-from sqlalchemy import String, Table
+from sqlalchemy import Connection, MetaData, String, Table, inspect, text
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.orm import DeclarativeBase
 
 from omni_box.core.models.enums import EventStatus
+from omni_box.infra.storage.postgres.orm import (
+    InboxEventDBBase,
+    InboxEventPartitionedDBBase,
+    OutboxEventDBBase,
+    OutboxEventPartitionedDBBase,
+)
 from tests.models import ConcreteOutboxEvent
 
 pytestmark = pytest.mark.integration
 
+CHECK_RULES = ("attempts_valid", "completed_status_consistency", "lock_consistency")
+
+# sqlalchemy-foundation-kit's DB_NAMING_CONVENTION
+FOUNDATION_KIT_CONVENTION = {
+    "ix": "%(column_0_label)s_idx",
+    "uq": "%(table_name)s_%(column_0_name)s_key",
+    "ck": "%(table_name)s_%(constraint_name)s_check",
+    "fk": "%(table_name)s_%(column_0_name)s_fkey",
+    "pk": "%(table_name)s_pkey",
+}
+
 
 def test__outbox_event_db_model__concrete_model__has_expected_columns_and_defaults() -> None:
     # Arrange
@@ -92,3 +111,45 @@ def test__outbox_event_db_model__concrete_model__has_expected_indexes() -> None:
     assert "scheduled_at" in [c.name for c in pending_fetch_idx.columns]
     assert pending_fetch_idx.dialect_options["postgresql"]["where"] is not None
     assert EventStatus.PENDING.value in str(pending_fetch_idx.dialect_options["postgresql"]["where"])
+
+
+async def test__event_bases__foundation_kit_convention__postgres_stores_the_qualified_names(
+    db_engine: AsyncEngine,
+) -> None:
+    # Arrange
+    schema = "naming_convention"
+
+    class Base(DeclarativeBase):
+        metadata = MetaData(naming_convention=FOUNDATION_KIT_CONVENTION, schema=schema)
+
+    class Outbox(Base, OutboxEventDBBase):
+        pass
+
+    class Inbox(Base, InboxEventDBBase):
+        pass
+
+    class OutboxPartitioned(Base, OutboxEventPartitionedDBBase):
+        pass
+
+    class InboxPartitioned(Base, InboxEventPartitionedDBBase):
+        pass
+
+    def reflect_check_names(conn: Connection) -> dict[str, set[str]]:
+        inspector = inspect(conn)
+        return {
+            table.name: {c["name"] for c in inspector.get_check_constraints(table.name, schema=schema)}
+            for table in Base.metadata.sorted_tables
+        }
+
+    # Act
+    async with db_engine.begin() as conn:
+        await conn.execute(text(f"CREATE SCHEMA {schema}"))
+        try:
+            await conn.run_sync(Base.metadata.create_all)
+            stored = await conn.run_sync(reflect_check_names)
+        finally:
+            await conn.execute(text(f"DROP SCHEMA {schema} CASCADE"))
+
+    # Assert
+    for table in Base.metadata.sorted_tables:
+        assert stored[table.name] == {f"{table.name}_{rule}_check" for rule in CHECK_RULES}