feat(OMN-1740): Add operational semantics to EventBusSubcontractWiring

[jonahgabriel]

Implement idempotency, error classification, and offset commit policies for contract-driven Kafka consumers per ARCH-002.

Changes:

• Add config models: ModelIdempotencyConfig, ModelOffsetPolicyConfig, ModelConsumerRetryConfig, ModelDlqConfig
• Add dispatch_with_transaction() to ProtocolDispatchEngine
• Implement idempotency gate with envelope_id deduplication
• Add error classification: content errors → DLQ, infra errors → fail-fast
• Add offset commit policies (commit_after_handler default)
• Add retry tracking with configurable max attempts

Test coverage:

• 20 new unit tests for idempotency, error classification, offset policy
• 6 new integration tests for AC#7 (deduplication) and AC#8 (offset commit)

Related: OMN-1740

Summary by CodeRabbit

• New Features

  • Configurable retry presets and fine-grained retry controls for message handlers.
  • Message deduplication (idempotency) to skip duplicates.
  • DLQ support with configurable failure semantics and consumer-group alignment.
  • Offset commit policy control for when offsets are committed.
  • Transactional dispatch API for transaction-scoped dispatch.

• Tests

  • Expanded integration and unit tests covering idempotency, retry tracking, DLQ behavior, offset commit semantics, and error classification.

• Chores

  • Dependency version bump.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

📝 Walkthrough

Walkthrough

Adds four event-bus Pydantic models and re-exports them, adds transaction-scoped dispatch API, and significantly extends EventBusSubcontractWiring with idempotency, retry/DLQ classification and handling, offset-commit controls, and associated tests; also bumps omnibase-core and passes node_name into runtime wiring.

Changes

Cohort / File(s)	Summary
Event Bus Configuration Models src/omnibase_infra/models/event_bus/model_consumer_retry_config.py, src/omnibase_infra/models/event_bus/model_dlq_config.py, src/omnibase_infra/models/event_bus/model_idempotency_config.py, src/omnibase_infra/models/event_bus/model_offset_policy_config.py	Added four validated Pydantic models: consumer retry (delay calc, presets), DLQ config, idempotency config (store type, retention), and offset commit policy.
Model Exports src/omnibase_infra/models/event_bus/__init__.py, src/omnibase_infra/models/__init__.py	New package initializer and parent module import and re-export the new event-bus models, expanding public surface.
Dispatch API src/omnibase_infra/protocols/protocol_dispatch_engine.py, src/omnibase_infra/runtime/service_message_dispatch_engine.py	Added dispatch_with_transaction(..., tx) to ProtocolDispatchEngine and service dispatch engine (service implementation contains a duplicated insertion in the patch).
Runtime Wiring src/omnibase_infra/runtime/event_bus_subcontract_wiring.py	Major wiring update: constructor/signatures extended to accept node_name and idempotency/dlq/retry/offset configs; added idempotency gate, retry counting and exhaustion, DLQ publishing, content vs infra error classification, offset commit helpers, correlation_id handling, and related state/utility methods.
Runtime Host Usage src/omnibase_infra/runtime/service_runtime_host_process.py	Now passes node_name="runtime-host" into EventBusSubcontractWiring when wiring subscriptions.
Tests — Integration tests/integration/runtime/test_event_bus_contract_wiring.py	Added integration tests for idempotency and offset commit semantics; updated tests to supply node_name and DLQ/idempotency configs; added in-memory idempotency store scaffolding.
Tests — Unit tests/unit/runtime/test_event_bus_subcontract_wiring.py	Expanded unit tests covering idempotency gate, retry tracking, DLQ routing, error classification (content vs infra), offset commit policy, cleanup, and related behaviors.
Project Config pyproject.toml	Bumped omnibase-core to ^0.10.1.

Sequence Diagram(s)

sequenceDiagram
    participant Consumer
    participant Wiring as EventBusSubcontractWiring
    participant Idempotency as IdempotencyStore
    participant Dispatcher as ProtocolDispatchEngine
    participant DLQ as DLQPublisher
    participant Offset as OffsetManager

    Consumer->>Wiring: deliver_message(message)
    Wiring->>Idempotency: is_processed(envelope_id)?
    alt duplicate
        Idempotency-->>Wiring: processed
        Wiring->>Offset: commit_offset(correlation_id)
    else new_message
        Idempotency-->>Wiring: not_processed
        Wiring->>Dispatcher: dispatch_with_transaction(envelope, tx?)
        alt success
            Dispatcher-->>Wiring: success
            Wiring->>Idempotency: record_processed(envelope_id)
            Wiring->>Offset: commit_offset(correlation_id)
        else content_error
            Dispatcher-->>Wiring: content_error
            alt dlq_and_commit
                Wiring->>DLQ: publish_to_dlq(message, error, consumer_group)
                Wiring->>Offset: commit_offset(correlation_id)
            else fail_fast
                Wiring-->>Consumer: raise ProtocolConfigurationError
            end
        else infra_error
            Dispatcher-->>Wiring: infra_error
            alt retry_available
                Wiring->>Wiring: increment_retry_count(correlation_id)
            else retries_exhausted
                Wiring->>DLQ: publish_to_dlq(message, error, consumer_group)
                opt dlq_and_commit
                    Wiring->>Offset: commit_offset(correlation_id)
                end
            end
        end
    end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• feat(OMN-1602): Implement derived Kafka consumer group IDs #197 — Overlaps on event-bus consumer-group/node-identity surface and wiring (DLQ/consumer_group propagation).
• refactor: enforce file/class naming conventions [OMN-1305] #151 — Related changes to event-bus models and exports, touching the same model/export surface and idempotency wiring.

Poem

🐰 I nibble logs and stitch the flows,

Envelopes numbered where the wild wind goes;
Retries that jitter, DLQs that keep,
Offsets committed so the stream can sleep;
A transactional hop — dedupe and leap!

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately and specifically describes the main feature: adding operational semantics (idempotency, error classification, offset policies, retry tracking) to EventBusSubcontractWiring, which is the primary change across the PR.
Docstring Coverage	✅ Passed	Docstring coverage is 93.33% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch jonah/omn-1740-infra-upstream-operational-semantics-idempotency-offset

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 5

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

src/omnibase_infra/runtime/event_bus_subcontract_wiring.py (1)

332-352: ⚠️ Potential issue | 🟡 Minor

Keep DLQ consumer_group aligned with the node_name used for subscriptions.

wire_subscriptions() can accept a different node_name than the constructor, but _publish_to_dlq() uses self._node_name. This can produce mismatched consumer_group metadata in DLQ entries.

🔧 Suggested fix

         for topic_suffix in subcontract.subscribe_topics:
+            if node_name != self._node_name:
+                self._logger.warning(
+                    "node_name override in wire_subscriptions: %s -> %s",
+                    self._node_name,
+                    node_name,
+                )
+                self._node_name = node_name
             full_topic = self.resolve_topic(topic_suffix)
             group_id = f"{self._environment}.{node_name}"

🤖 Fix all issues with AI agents

In `@src/omnibase_infra/models/event_bus/model_idempotency_config.py`:
- Around line 6-19: Update the module/class docstrings to consistently refer to
envelope_id instead of event_id: find the top-level module docstring and the
"Idempotency Overview" section plus any docstrings in the IdempotencyConfig
model (or any docstring mentioning the idempotency store) and replace the
incorrect `event_id` mentions with `envelope_id` (specifically update the lines
referenced in the review where `event_id` appears). Ensure any examples or
bullets describing the deduplication behavior mention `envelope.envelope_id` as
the key used for INSERT ... ON CONFLICT DO NOTHING and update the
retention/pruning description to reference `envelope_id` as well.

In `@src/omnibase_infra/protocols/protocol_dispatch_engine.py`:
- Around line 168-215: The docstring example for dispatch_with_transaction
currently shows using isinstance(tx, asyncpg.Connection) and raising TypeError;
update it to demonstrate duck-typing and raising the project-specific OnexError
instead: change the example in the dispatch_with_transaction docstring to
perform a minimal, protocol-style duck-typing check (e.g., verify required
method/attribute on tx) and raise OnexError (not TypeError) when the check
fails, and briefly mention that implementations should prefer duck-typing and
raise OnexError for transaction-type mismatches.

In `@src/omnibase_infra/runtime/event_bus_subcontract_wiring.py`:
- Around line 562-577: The duplicate branch currently returns before committing,
causing infinite redelivery; update the block after
_idempotency_store.check_and_record (when not is_new) to still
acknowledge/commit the message before exiting—e.g., invoke the same commit logic
used by commit_after_handler (or call the consumer commit/ack method your wiring
uses) and then log via self._logger.info and return; ensure you reference
envelope_id, topic, correlation_id, self._node_name and reuse the existing
commit routine so duplicates are acknowledged and not redelivered.

In `@src/omnibase_infra/runtime/service_message_dispatch_engine.py`:
- Around line 1494-1511: dispatch_with_transaction currently dereferences
envelope (e.g., envelope.correlation_id, envelope.trace_id) and assumes topic is
truthy before calling dispatch, which can raise AttributeError instead of the
expected ModelOnexError; update dispatch_with_transaction to validate inputs
up-front: check that envelope is not None and that topic is non-empty (or meets
whatever contract) and if invalid raise/return the appropriate ModelOnexError,
only then access envelope.correlation_id/trace_id, and finally call
self.dispatch(topic=topic, envelope=envelope); keep tx unused for now but
preserve the existing tx acknowledgement.