omni-box for AI agents¶

One page holding everything a coding assistant needs to wire and drive omni-box correctly, plus a map of where the rest of the documentation keeps the details it leaves out. Give an agent this page rather than the whole site.

How to read this page¶

Every page of this site is also served as raw Markdown at its own URL with .md in place of the trailing slash — this page is /agents.md, the quick start is /guide/quickstart.md — so anything the map below points at can be fetched as plain text rather than scraped out of HTML. The Copy page control at the top of a page does the same thing for a human with a chat window open. The one exception is the generated API reference: its Markdown is a directive to a docstring renderer rather than the API, so it carries neither the control nor a .md twin — read it as HTML, or read the docstrings in the source.

Top to bottom before writing code. Rules that hold or break the code is the section correctness lives in — those are the things the library will not save you from. Every name used below is in the public API; if you need something not listed here, fetch the page the documentation map points at rather than guessing a method that sounds plausible.

Scope¶

It does give you the two halves of transactional messaging as primitives. The outbox: a validated event entity you insert in the same transaction as your business state, and a relay that locks a batch of pending rows, hands each to a broker, and writes the outcome back. The inbox: a row keyed by (message_id, consumer_group) that a redelivered message collides with, a consumer runner that lands one message per transaction with configurable commit semantics, and a batch processor for rows already landed. Around both: a composable pipeline (metrics, OpenTelemetry, DLQ, circuit breaker, sibling deduplication), a PostgreSQL repository that speaks FOR UPDATE SKIP LOCKED, and an aiokafka adapter.

It does not own a transaction. There is no Unit of Work, no session factory, no commit() anywhere in the library except the broker offset commit — the transactional boundary that makes an outbox an outbox is yours, and it is the one thing you must get right. It creates no tables and ships no migrations: the ORM bases are abstract and you bind them to your own DeclarativeBase. It runs no scheduler and no daemon; the publisher loop is a while in your worker. It does not deliver exactly once — nothing does — and it does not make your handler idempotent.

Mental model¶

Outbox. OmniBoxDomainService.create_outbox_event(...) builds an OutboxEvent — a frozen Pydantic model with status=PENDING. You insert it through an OutboxEventRepository in the same transaction as the business change; either both land or neither does. Later, in a different transaction, OutboxPublisher.publish_batch(...) runs one cycle: fetch and lock pending rows, publish each through an EventPublisher, commit the outcomes back to the rows. All three are one database transaction, and it is yours to commit.

Inbox. InboxConsumerRunner pulls one message from an EventConsumer, opens a transaction through your InboxTransactionProviderProtocol, inserts an InboxEvent, runs your handler inside that same transaction, and commits the broker offset according to the AckStrategy. A message already in the table does not insert a second row: the repository returns the existing one, and if it is completed the runner reports a duplicate and moves on. Alternatively, land rows fast with no handler and drain them later with create_inbox_processor — the same fetch/lock/process/commit cycle the outbox uses.

The cycle underneath both batch paths is EventBatchProcessor.process_batch:

• a fetch strategy returns a locked batch — DistributedLockingFetchStrategy (SELECT … FOR UPDATE SKIP LOCKED then UPDATE … SET locked_by) when the repository reports supports_distributed_locking, OptimisticLockingFetchStrategy otherwise;
• a pipeline of ProcessingSteps runs per event against a ProcessingContext that accumulates completed, failed-counted, failed-non-counted and skipped ids;
• a commit strategy writes those back — BulkCommitStrategy when the repository reports supports_bulk, SingleCommitStrategy otherwise;
• a BatchProcessingResult comes out. EventProcessorBuilder.build() picks both strategies from repo.capabilities; you only call with_* to override.

State is three values. pending → completed on success; pending → pending with attempts_made + 1 on a counted failure; pending → failed when attempts_made reaches max_attempts. failed is terminal — nothing fetches it again. A non-counted failure leaves the row pending with attempts_made where it was: that is what a broker or a downstream that is not there produces, and an outage therefore costs no budget however long it lasts.

Locking is a (locked_at, locked_by) pair on the row plus, on PostgreSQL, the row lock the fetching transaction holds. Two workers do not collide because of SKIP LOCKED; the column pair is what survives a crash, and OmniBoxMaintenanceService.release_stale_locks is what clears it afterwards.

Wiring¶

The outbox, end to end¶

import asyncio

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from omni_box import OmniBoxDomainService, OutboxPublisher
from omni_box.core.converters import EnvelopeEventConverter
from omni_box.infra.brokers.kafka import KafkaEventPublisher
from omni_box.infra.storage.postgres import OutboxEventDBBase, PostgresOutboxRepository


class Base(DeclarativeBase):
    pass


class OutboxEventDB(Base, OutboxEventDBBase):      # your table, your migration
    pass


engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")
session_factory = async_sessionmaker(engine, expire_on_commit=False)
domain = OmniBoxDomainService()

# 1. Write the event with the business state, in one transaction.
async with session_factory() as session, session.begin():
    session.add(User(id=user_id, email=email))
    await PostgresOutboxRepository(session, model_class=OutboxEventDB).create(
        domain.create_outbox_event(
            aggregate_type="user",
            aggregate_id=user_id,
            event_type="user.created",
            topic="users.events",
            partition_key=str(user_id),
            payload={"email": email},          # non-empty, JSON-serializable
            idempotency_key=f"user.created:{user_id}",
        )
    )

# 2. Relay the rows, in a different transaction, in a different process if you like.
broker = KafkaEventPublisher(producer=producer, converter=EnvelopeEventConverter())

while not shutdown:
    async with session_factory() as session, session.begin():   # the commit is yours
        repo = PostgresOutboxRepository(session, model_class=OutboxEventDB)
        result = await OutboxPublisher(repo, broker).publish_batch(
            worker_id="publisher-1",           # [A-Za-z0-9.-/:] only
            batch_size=100,
        )
    if not result.processed_event_ids:
        await asyncio.sleep(1.0)

The inbox, end to end¶

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from omni_box import AckStrategy, InboxConsumerRunner, InboxEvent, InboxEventRepository
from omni_box.core.protocols.transaction import InboxTransactionProviderProtocol
from omni_box.infra.storage.postgres import InboxEventDBBase, PostgresInboxRepository


class InboxTxProvider(InboxTransactionProviderProtocol):
    def __init__(self, session_factory) -> None:
        self._session_factory = session_factory

    @asynccontextmanager
    async def transaction(self) -> AsyncIterator[InboxEventRepository]:
        async with self._session_factory() as session, session.begin():
            yield PostgresInboxRepository(session, model_class=InboxEventDB)


async def handle(event: InboxEvent, repo: InboxEventRepository) -> None:
    await repo.session.execute(             # the transaction the inbox row is in
        invoices.insert().values(order_id=event.payload["order_id"])
    )


runner = InboxConsumerRunner(
    consumer=kafka_consumer,                # any EventConsumer
    transaction_provider=InboxTxProvider(session_factory),
    handler=handle,                         # optional; omit to land rows and drain later
    worker_id="worker-1",
    consumer_group="identity-service",      # part of the deduplication key
    ack_strategy=AckStrategy.EXACTLY_ONCE_INBOX,
)

await runner.start()
try:
    await runner.run_forever()
finally:
    await runner.stop()

The API¶

Entities¶

BaseEvent is frozen (model_config = ConfigDict(frozen=True)); every transition returns a new object.

OutboxEvent adds aggregate_type (≤50), aggregate_id (UUID), topic (≤255), partition_key (≤255) — all required. InboxEvent adds message_id (≤255), consumer_group (≤255), source (≤255), plus processed_at (alias for completed_at), get_context_value(key) and get_payload_as(schema_cls=None).

Properties on both: is_locked, can_retry, attempts_left, failure_count. There is no updated_at on the entity — that column exists only in the ORM row.

EventStatus is a StrEnum: PENDING = "pending", COMPLETED = "completed", FAILED = "failed". The DDL uses the lowercase strings.

OmniBoxDomainService¶

Constructor keywords, all optional: max_attempts=6, scheduled_at_skew_seconds=60, scheduled_at_max_future_seconds=31536000, payload_max_bytes=1_000_000, headers_max_count=100, header_key_max_length=64, header_value_max_length=512, last_error_max_length=2000, truncation_suffix="... [TRUNCATED]".

create_inbox_event takes no idempotency_key and no max_attempts: the inbox key is (message_id, consumer_group) and the budget comes from the service.

Repository protocols (omni_box.core.protocols)¶

FetchFilters is a TypedDict: source, topic, aggregate_type, aggregate_id, each a value or a list. Capability protocols in omni_box.core.protocols.features: SupportsBulkOperations (bulk_create, bulk_mark_completed, bulk_mark_failed, bulk_release_locks), SupportsDistributedLocking (fetch_and_lock_pending, refresh_lock, release_lock, force_unlock), SupportsRetentionPolicies (delete_old_completed, release_stale_locks). The builder accepts either the structural match or the capabilities flags.

High-level services¶

InboxConsumeResult is (message_id, event_id, committed, processed, duplicate). BatchProcessingResult is (processed_event_ids, failed_counted, failed_noncounted, remaining_event_ids, commit_failed); the two failure lists hold EventFailureUpdate(event_id, error, next_retry_at) tuples.

Commit semantics, exactly as the runner implements them:

Factories (omni_box.application.factories)¶

All three return an EventBatchProcessor and share metrics=None, dlq_storage=None, enable_otel=False, enable_circuit_breaker=False, circuit_breaker_failure_threshold=5, circuit_breaker_recovery_timeout=60, job_name, additional_steps_before=None, additional_steps_after=None.

• create_outbox_processor(repo, publisher, *, publish_timeout=30.0, …)
• create_inbox_processor(repo, handler, *, skip_duplicate_siblings=True, filter_sources=None, process_timeout=30.0, …)
• create_dispatching_processor(repo, router, *, filter_sources=None, skip_duplicate_siblings=True, process_timeout=30.0, dependencies=None, …)

Pipeline (omni_box.core.pipeline)¶

EventProcessorBuilder(repo) — add_step(step), with_fetch_strategy(s), with_commit_strategy(s), with_metrics(m), with_lease_ttl(seconds) (default 300), with_job_name(name), build(). Every with_* returns self.

A step implements async def execute(event, context) -> StepResult and may implement on_batch_start / on_batch_end / on_event_start / on_event_end; subclass BaseProcessingStep to get no-op hooks. The ProcessingContext it is handed carries repo, worker_id, metrics, the result lists, an extra dict of your own, and shutdown_requested — the predicate process_batch was called with. StepResult.next() continues, StepResult.skip() drops the rest of the pipeline for this event, StepResult.stop() ends the whole batch.

Fetch strategies: DistributedLockingFetchStrategy(ttl=300), OptimisticLockingFetchStrategy(), FilteredFetchStrategy(sources=None, ttl=300). Commit strategies: BulkCommitStrategy(), SingleCommitStrategy().

Handler results (omni_box.core.services.results)¶

A handler returns None (treated as success) or an EventHandlerResult(success, processed=True, status=None, error_message=None, count_as_attempt=True, next_retry_at=None). Three constructors are re-exported at the top level:

EventHandlerStatus is COMPLETED, STALE, SKIPPED, FAILED, RETRY.

Routing (omni_box.core.dispatch)¶

EventRouter(normalize_topic=None) keys handlers by (topic, event_type, schema_version). register_handler(event_type, topic, handler, schema_version=None, handler_name=None) registers a function; register_class(cls, topic=None) / register_instance(obj, topic=None) sweep a BaseEventHandler subclass for methods marked with @event_handler(event_type, topic=None, schema_version=None). dispatch(event, topic, repo, **dependencies) tries the exact schema_version, then a registered migration (BaseEventSchema.register_migration), then the version-agnostic entry, and returns a failed EventHandlerResult when nothing matches. create_dispatching_processor passes event.source as the topic; create_dispatching_handler(router, **dependencies) is the handler it installs, for when you assemble the pipeline yourself.

PostgreSQL (omni_box.infra.storage.postgres, extra postgres)¶

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.

PostgresOutboxRepository(session, *, model_class, conflict_index_id=None, conflict_index_idempotency=None, batch_size=1000, error_max_length=2000, truncation_suffix=…, scheduled_at_skew_seconds=60) and PostgresInboxRepository(...) with the same keywords. Both implement every capability protocol, and both expose two things from the shared base: session, read-only, the AsyncSession the repository was built on — in a handler passed to InboxConsumerRunner it is the transaction the inbox row is in — and requeue_failed(event_id) -> bool, the only way to move a failed row back to pending. A handler receives repo typed as InboxEventRepository, which has no session; narrow it with isinstance(repo, PostgresInboxRepository) where a type checker needs to see it.

Kafka (omni_box.infra.brokers.kafka, extra kafka)¶

KafkaEventPublisher(producer, converter, *, max_infra_retries=3) — you own the AIOKafkaProducer lifecycle; set enable_idempotence=True and acks="all" on it. A broker that does not answer — a connection or node error, a request or client timeout, or a topic it cannot fetch metadata for while it ignores a metadata request as well — is retried max_infra_retries times and then raised as TransientError, which costs the row no attempt. Everything else, a record the broker rejects or a topic it says it does not have included, is raised as it is and counts. KafkaEventConsumer(consumer, *, payload_loader=None, message_id_getter=None, event_type_getter=None, source_getter=None, envelope_parser=None) — you own the AIOKafkaConsumer and should set enable_auto_commit=False. Without a message_id_getter the message id is the message_id or event_id header, falling back to "{topic}:{partition}:{offset}".

Converters (omni_box.core.converters): RawEventConverter (the payload as-is), SchemaVersionedConverter (schema_version + payload), EnvelopeEventConverter( default_schema_version="1.0.0") (adds event_type, aggregate identity, timestamp, tracing ids).

Elsewhere¶

omni_box.infra.metrics (extra metrics) — PrometheusOutboxMetrics(prefix=None), PrometheusInboxMetrics(prefix=None); ProcessingMetrics is the shared base of the two protocols, not a third implementation. EventBatchProcessor sets set_locked_batch_size(len(batch)) after every fetch, zero included. omni_box.contrib.settings (extra settings) — BaseOutboxSettings / BaseInboxSettings, reading OMNI_OUTBOX_ / OMNI_INBOX_ with __ as the nesting delimiter. omni_box.contrib.dishka (extra dishka) — EventDispatcherProvider, DIAwareEventRouter, create_di_router. omni_box.testing — assert_outbox_event_created. omni_box.utils — utc_now, calculate_backoff_with_jitter, ErrorClassifier.

Rules that hold or break the code¶

1. The library never opens or commits a database transaction. Every repository call runs in the session you handed it. publish_batch and process_batch must be wrapped in a transaction you commit — the fetch, the lock, the publish and the status update are one unit. Forget the commit and nothing happened: the rows are still pending and the next cycle publishes them again.
2. Delivery is at-least-once in both directions. The broker send happens before the commit that marks the row completed, so a crash in between republishes. That is the point of the inbox on the other side; it is not a defect to work around.
3. The inbox deduplication key is (message_id, consumer_group), and the window is the lifetime of the row. It is a unique index, not a time window: it holds for as long as the row is in the table and not one moment longer. cleanup_old_events(retention_days=N) or dropping a partition removes the row, and a message redelivered after that is new again. Pick retention_days from your broker's retention, not from disk pressure. Change consumer_group and every message in flight is new too.
4. A handler passed to InboxConsumerRunner runs inside the transaction that inserts the inbox row. If it raises, the insert rolls back with it — there is no pending row left behind to retry from, and the retry has to come from the broker. That is the exactly-once effect the pattern gives you: the side effect and the record of it commit together — provided the side effect goes through repo.session, the AsyncSession the row was inserted on. A session the handler opens itself is a second transaction: its writes survive a rollback of the row, and the redelivery does them again. The runner takes the row's lock first and only runs the handler if it got it; when someone else holds it you get processed=False, duplicate=False, committed=False and the broker redelivers.
5. The runner never records a failure. It calls neither mark_failed nor anything that increments attempts_made, so max_attempts does nothing on that path. The retry budget only exists for the batch processors (create_*_processor, EventBatchProcessor).
6. worker_id is validated by the PostgreSQL fetch-and-lock path: [A-Za-z0-9.\-/:] only, at most 255 characters. An underscore or a space raises ValueError at fetch time, not at construction. worker-1 is fine, worker_1 is not.
7. payload must be a non-empty JSON object of at most 1 000 000 bytes, with no NaN and no Infinity. payload={} raises. Everything is normalised through orjson, so what you read back is what will go on the wire.
8. Entities are frozen. Every OmniBoxDomainService method returns a new event; none of them touch the database. They are for services that keep events in memory. The repository methods (mark_completed, mark_failed, …) are what changes a row, and they take an id, not an entity.
9. attempts_made == max_attempts means failed, and failed is terminal. The default budget is 6. fetch_pending and fetch_and_lock_pending both filter on attempts_made < max_attempts, so a failed row is never picked up again; PostgresEventRepository.requeue_failed(event_id) resets it.
10. create() returns the row that is in the table. On a duplicate — the same (message_id, consumer_group), or the same idempotency_key — you get the existing row back, with its own id and status, and no exception. Compare returned.id == event.id if you need to know whether you were the writer.
11. SiblingDeduplicationStep is a no-op on a non-partitioned table. PostgresInboxRepository.has_completed_sibling_for_inbox_key returns False unless the dedup key carries created_at, because on a plain table the unique index already makes a sibling impossible. It earns its keep only on the partitioned bases.
12. On a partitioned table the unique index cannot enforce the logical key — PostgreSQL requires the partition key in it, and two retries never share a created_at. Both repositories therefore serialize the identity with pg_advisory_xact_lock and look it up before inserting: one advisory lock and one SELECT per insert, held until your transaction ends. One more reason to keep those transactions short.
13. Nothing classifies errors for you; you say so, by raising or returning. In the pipeline every exception out of a handler or a publisher is a counted failure, with one named exception: TransientError, which says the failure belongs to the environment and not to the event. It is recorded without spending an attempt and the row comes back a second later. Raise it from your handler or your broker adapter when the thing you were talking to is not there; KafkaEventPublisher raises it once max_infra_retries are spent on a broker that does not answer. The equivalent for a handler that returns rather than raises is handler_retry(msg, count_as_attempt=False, next_retry_at=…) — and next_retry_at is mandatory when count_as_attempt=False. ErrorClassifier still classifies nothing on your behalf: it lives in omni_box.utils and is used only inside KafkaEventPublisher's own infrastructure retry.
14. scheduled_at is validated against created_at: no more than 60 seconds before it, no more than 365 days after. A retry scheduled beyond that is rejected, not clamped.
15. release_stale_locks(stale_timeout_seconds) must use a timeout comfortably larger than with_lease_ttl(...) and than the worst-case handler runtime. Set it too low and maintenance unlocks rows a healthy worker is still holding.
16. CircuitBreakerStep state is per process — replicas trip independently and a restart forgets everything. DLQStep is best-effort: it runs outside the commit, a failure in move_to_dlq is logged and swallowed, and the event goes to failed anyway. 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.
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 remaining_event_ids, still locked, for release_stale_locks or your own release_lock to clear. The event in flight always runs to completion.
19. There is no scheduler. The relay loop, its sleep, its shutdown and the maintenance cadence are yours to write.

Common mistakes¶

# WRONG — a session with no transaction: the lock and the completion never land
session = session_factory()
repo = PostgresOutboxRepository(session, model_class=OutboxEventDB)
await OutboxPublisher(repo, broker).publish_batch(worker_id="pub-1", batch_size=100)

# RIGHT — one transaction per cycle, committed by the caller
async with session_factory() as session, session.begin():
    repo = PostgresOutboxRepository(session, model_class=OutboxEventDB)
    result = await OutboxPublisher(repo, broker).publish_batch(worker_id="pub-1", batch_size=100)

# WRONG — underscores are not allowed in worker_id; this raises at fetch time
await publisher.publish_batch(worker_id="outbox_worker_1", batch_size=50)

# RIGHT
await publisher.publish_batch(worker_id="outbox-worker-1", batch_size=50)

# WRONG — an empty payload, and a mutation of a frozen model
event = domain.create_outbox_event(..., payload={})     # ValidationError: payload cannot be empty
event.status = EventStatus.COMPLETED                    # ValidationError: instance is frozen

# RIGHT
event = domain.create_outbox_event(..., payload={"user_id": str(user_id)})
event = domain.mark_event_completed(event, completed_at=utc_now(), worker_id="worker-1")

# WRONG — the decorator alone registers nothing, and it has no `source` argument
@event_handler(event_type="user.created", source="users")
async def handle_user_created(event, repo): ...

# RIGHT — either a plain function registered explicitly …
router.register_handler(event_type="user.created", topic="users", handler=handle_user_created)

# … or a class the router sweeps
class UserHandlers(BaseEventHandler):
    topic = "users"

    @event_handler("user.created")
    async def on_created(self, event: InboxEvent, repo: InboxEventRepository) -> None: ...

router.register_instance(UserHandlers())

# WRONG — assuming the new row is yours, and that a duplicate raises
stored = await repo.create(event)
await do_side_effect(event.id)          # `event.id` may be nowhere in the table

# RIGHT — the returned row is the one that exists
stored = await repo.create(event)
if stored.status is EventStatus.COMPLETED:
    return                              # a redelivery of work already done
await do_side_effect(stored.id)

# WRONG — expecting the runner to retry a failing handler out of the table
runner = InboxConsumerRunner(..., handler=flaky_handler)   # max_attempts does nothing here

# RIGHT — land the message, drain it with a processor that owns the retry budget
runner = InboxConsumerRunner(..., handler=None, ack_strategy=AckStrategy.AT_LEAST_ONCE)
processor = create_inbox_processor(repo=inbox_repo, handler=flaky_handler)
async with session_factory() as session, session.begin():
    await processor.process_batch(worker_id="inbox-1", batch_size=50)

# WRONG — the handler opens its own session: the invoice commits even when the inbox row rolls back
async def handle(event: InboxEvent, repo: InboxEventRepository) -> None:
    async with session_factory() as session, session.begin():
        await session.execute(invoices.insert().values(order_id=event.payload["order_id"]))

# RIGHT — write through the session the inbox row is on; both commit, or neither does
async def handle(event: InboxEvent, repo: InboxEventRepository) -> None:
    await repo.session.execute(invoices.insert().values(order_id=event.payload["order_id"]))

Errors¶

Everything derives from OmniBoxError.

Outside the top-level namespace: SchemaResolutionError (omni_box.core.models.schemas, also a ValueError) when BaseEventSchema.resolve finds no schema; HandlerAlreadyRegisteredError and its base DispatcherError (omni_box.core.dispatch.exceptions); PipelineStoppedError and its base PipelineError (omni_box.core.pipeline.exceptions), raised internally when a step returns StepResult.stop() and caught by the pipeline.

Documentation map¶

Fetch a page when the task is the one named beside it.