fix(OMN-15232): fail-close the deserialization-path DLQ publish in event_bus_kafka

[jonahgabriel]

Defect

src/omnibase_infra/event_bus/event_bus_kafka.py awaited _publish_raw_to_dlq(...) on the deserialization-failure path, discarded the returned bool, then continued:

await self._publish_raw_to_dlq(...)   # returns bool; return value DISCARDED
continue  # Skip this message but continue consuming

Consumers built by this class pass enable_auto_commit=self._config.enable_auto_commit, which defaults to True, and this loop never commits at all — the client commits the fetch position on its own cadence. So a failed DLQ publish on a poison message was a silent, committed, unrecoverable drop.

Same defect class as OMN-14936, at a call site that fix did not cover: the OMN-14936 gate landed only in runtime/event_bus_subcontract_wiring.py as 5x if not dlq_persisted: return.

Second ungated site found by the required module audit

_dispatch_to_subscriber discarded the DLQ result on the retries-exhausted branch — and MixinKafkaDlq._publish_to_dlq returned None unconditionally, so no caller could have gated on it. The static ratchet added here flagged both sites (lines 2075 and 2233) on the unfixed tree.

Fix — and why a rewind, not "withhold the commit"

The OMN-14936 gate works by not calling commit, because that path is a manual-commit model. That mechanism is a no-op in this loop: it has no commit call to withhold; the aiokafka client commits for it. Declining to commit here would leave the fail-open behaviour exactly as-is.

The fail-closed action that works under both commit models is a rewind of the fetch position:

consumer.seek(TopicPartition(msg.topic, msg.partition), msg.offset)

This is the same "does NOT advance the committed offset" idiom KafkaTransport.nack already uses in event_bus/kafka_transport.py. Under auto-commit the committer then commits the rewound position, which cannot be past the un-persisted message; under manual commit the message is simply refetched. Either way Kafka redelivers and the DLQ write is retried instead of the record being lost. Only the failed message's own partition is rewound — siblings untouched, matching the OMN-14757 per-partition discipline.

Changes

file	change
mixin_kafka_dlq.py	_publish_to_dlq returns bool (the success it already computed), mirroring the _publish_raw_to_dlq contract from OMN-14936
event_bus_kafka.py	_dispatch_to_subscriber returns "safe to advance offset"; False only when retries exhausted AND DLQ unconfirmed
event_bus_kafka.py	_consume_loop gates both sites; new _rewind_after_unpersisted_dlq helper
event_bus_kafka.py	DLQ_UNPERSISTED_REWIND_BACKOFF_SECONDS bounds the retry rate so an unreachable DLQ cannot hot-spin

Only an explicit False counts as a confirmed non-persist — duck-typed hosts/test doubles returning None keep prior behaviour, matching the allowance the OMN-14936 gate makes at event_bus_subcontract_wiring.py:607.

Accepted trade-off: with the DLQ down, a poison message now stalls its partition instead of being dropped. That is the intended fail-closed semantics and is what OMN-14936 already chose ("Kafka will redeliver this message and we retry the DLQ write rather than silently losing it"). On the fan-out path, redelivery may duplicate for subscribers that already succeeded — at-least-once, which is the delivery contract here, and strictly preferable to losing the record.

Seams

Two private return types widen None -> bool. Both are private, have no external callers, and mypy src/omnibase_infra/ is clean across 2615 source files.

• MixinKafkaDlq._publish_to_dlq: None → bool
• EventBusKafka._dispatch_to_subscriber: None → bool

EventBusSubcontractWiring._publish_to_dlq is a different method on a different class (already bool under OMN-14936) and is untouched. ProtocolKafkaDlqHost does not declare either method, so no protocol surface moved.

RED-first proof

tests/unit/event_bus/test_deser_dlq_fail_closed_omn15232.py drives the real _consume_loop against a fake consumer — the artifact that runs, not a surrogate.

Against unfixed dev (4 failed, 1 passed):

test_unpersisted_dlq_write_rewinds_offset_instead_of_advancing FAILED
test_confirmed_dlq_write_lets_offset_advance                   PASSED   <- intentional control
test_handler_path_unpersisted_dlq_also_rewinds                 FAILED
test_publish_to_dlq_reports_persistence_outcome                FAILED
test_no_dlq_publish_call_site_discards_its_persistence_result  FAILED

The static ratchet named both defects precisely:

AssertionError: event_bus_kafka.py discards the DLQ persistence result at line(s) [2075, 2233].

With the fix: 5 passed.

The one pre-passing test is the deliberate control — it proves the fix is a gate, not a blanket rewind (a confirmed DLQ persist must still let the offset advance, or poison messages replay forever).

Verification (rule 11a — patch-transferred to .200, gates + push from .200)

Diff was generated locally, scp'd, git apply'd on .200, and sha256-verified identical on both hosts for all three files before any gate ran (no vacuous green). Both worktrees on base f7fb7cde. Commit and push executed on .200.

uv run ruff check src/ tests/          All checks passed!
uv run ruff format --check src/ tests/ 4526 files already formatted
uv run mypy src/omnibase_infra/        Success: no issues found in 2615 source files
uv run pytest tests/unit/ -q -n auto   22058 passed, 40 skipped
pre-commit run --files <3 changed>     all Passed, no files modified by hooks

Full-suite run (pytest tests/ -q -n auto): 26351 passed, 25 failed, 42 errors. Every failure/error is in tests/integration, tests/performance, or tests/replay and needs live infra absent on .200 — Kafka on localhost:19092, Postgres, and the LLM SLO endpoints. Zero failures in tests/unit/ when the unit tree is run on its own (22058 passed). The 4 unit failures that appeared inside the combined xdist session are cross-suite env-var pollution from the integration tests sharing the worker pool: those same 3 files pass 65/65 both in this patched worktree and in the clean dev checkout on .200.

No gate was bypassed; no --no-verify, no skip tokens.

OCC

Companion (net-new-file-only): OmniNode-ai/onex_change_control#5115 — opens and merges before this PR lands, per strict OCC ordering (OMN-15222). Its three evidence probes were verified present at ceb33d341 and absent on dev.

Closes OMN-15232

Evidence-Ticket: OMN-15232
Evidence-Source: OCC#5118

Summary by CodeRabbit

• Bug Fixes
  • Improved Kafka event handling to prevent messages from being lost when delivery to the dead-letter queue (DLQ) cannot be confirmed.
  • Failed messages are now retried by rewinding the consumer position when DLQ persistence is unsuccessful or uncertain.
  • Consumer offsets advance only after successful DLQ delivery is confirmed.
  • Applies to both deserialization failures and messages exhausted after handler retries.

[coderabbitai]

📝 Walkthrough

Walkthrough

Kafka DLQ publishing now returns a persistence result. The Kafka consumer uses that result to allow offset advancement only after confirmed DLQ persistence, otherwise rewinding to the failed record for deserialization and exhausted-handler failures.

Changes

Kafka DLQ fail-closed flow

Layer / File(s)	Summary
DLQ persistence outcome contract src/omnibase_infra/event_bus/mixin_kafka_dlq.py	_publish_to_dlq returns bool, reports invalid topics as unsuccessful, and exposes the final persistence outcome.
Dispatch and offset rewind flow src/omnibase_infra/event_bus/event_bus_kafka.py	Subscriber dispatch and deserialization handling gate offset advancement on DLQ durability and seek back to the failed partition and offset when persistence is unconfirmed.
Fail-closed behavior validation tests/unit/event_bus/test_deser_dlq_fail_closed_omn15232.py	Tests cover rewind, confirmed DLQ advancement, handler retry exhaustion, persistence results, and unchecked DLQ call sites.

Estimated code review effort: 4 (Complex) | ~45 minutes

Sequence Diagram(s)

sequenceDiagram
  participant KafkaConsumer
  participant EventBusKafka
  participant MixinKafkaDlq
  KafkaConsumer->>EventBusKafka: deliver failed record
  EventBusKafka->>MixinKafkaDlq: publish record to DLQ
  MixinKafkaDlq-->>EventBusKafka: return persistence result
  alt result is true
    EventBusKafka-->>KafkaConsumer: allow offset advancement
  else result is false
    EventBusKafka->>KafkaConsumer: seek to failed partition and offset
  end

Loading

Possibly related PRs

• OmniNode-ai/omnibase_infra#2314: Extends the DLQ publishing interface with failure metadata and validation details.
• OmniNode-ai/omnibase_infra#2403: Uses explicit DLQ persistence results to gate offset commits in subcontract wiring.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly captures the main change: fail-closing DLQ publishing in event_bus_kafka for the deserialization path.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch jonah/omn-15232-deser-dlq-fail-closed

---

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

[coderabbitai]

🧹 Nitpick comments (1)

src/omnibase_infra/event_bus/event_bus_kafka.py (1)

2139-2242: 🩺 Stability & Availability | 🔵 Trivial

Rewind mechanism is correct; consider a rewind counter for observability.

Verified against aiokafka's own documented pattern (consumer.seek(tp, msg.offset) immediately followed by getone() returning the same message) — the seek-based rewind here is a supported, immediate-effect API, not a buffered/delayed one.

One operational gap: if the DLQ stays unreachable, this partition will loop seek → sleep(1s) → refetch → fail indefinitely (by design, to avoid silent loss), but there's currently no counter/metric distinguishing this from normal traffic — only ERROR-level logs. Consider incrementing a metric (e.g. via the existing DLQ metrics mechanism) on each dlq_unpersisted_offset_rewound so an operator can alert on a stuck partition rather than relying on log-grepping.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/omnibase_infra/event_bus/event_bus_kafka.py` around lines 2139 - 2242,
The _rewind_after_unpersisted_dlq method lacks a metric for repeated failed DLQ
persistence retries. Using the existing DLQ metrics mechanism, increment a
counter each time the successful rewind path emits
dlq_unpersisted_offset_rewound, including suitable topic, group, partition, and
failure-stage dimensions if supported, while leaving the rewind and backoff
behavior unchanged.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@src/omnibase_infra/event_bus/event_bus_kafka.py`:
- Around line 2139-2242: The _rewind_after_unpersisted_dlq method lacks a metric
for repeated failed DLQ persistence retries. Using the existing DLQ metrics
mechanism, increment a counter each time the successful rewind path emits
dlq_unpersisted_offset_rewound, including suitable topic, group, partition, and
failure-stage dimensions if supported, while leaving the rewind and backoff
behavior unchanged.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: c0698441-42be-4213-a6f7-2179915ef2df

📥 Commits

Reviewing files that changed from the base of the PR and between f7fb7cd and ceb33d3.

📒 Files selected for processing (3)

• src/omnibase_infra/event_bus/event_bus_kafka.py
• src/omnibase_infra/event_bus/mixin_kafka_dlq.py
• tests/unit/event_bus/test_deser_dlq_fail_closed_omn15232.py

[github-actions]

⚠️ Hostile Reviewer — DEGRADED (informational)

Blocking findings (critical): 0
Total findings: 0
Models succeeded: none

Note: All reviewer models failed or were unavailable. Degraded results are informational during the pilot phase (OMN-8468/OMN-8524) and do not block merge. Error: all review endpoints [192.168.86.201:8000 192.168.86.201:8001 ] unreachable — preflight short-circuit (no models available)

---

Gate semantics (pilot phase)

Verdict	Meaning	Blocks merge?
passed	No critical findings	No
blocked	CRITICAL findings found	Yes
degraded	All models unavailable (infra)	No (pilot)

Powered by omniintelligence.review_pairing.cli_review — node-based adversarial review via HandlerLlmCliSubprocess (OMN-8468/OMN-8524)