[jonahgabriel]

OMN-18418

The fail-closed gate in front of every dispatched delegation resolved consumer-group
liveness through a synchronous confluent admin client, while every producer, consumer
and admin client the runtime wires is aiokafka via build_aiokafka_auth_kwargs.

AWS_MSK_IAM and OAUTHBEARER authenticate through a token callback on the client,
which librdkafka cannot take as a flat config entry, so build_confluent_auth_config
refuses both by design (src/omnibase_infra/event_bus/kafka_auth.py:239-250). That
refusal arrived at live_consumer_groups as an exception and left as UNKNOWN, and
resolve_delegate_locus fails closed on UNKNOWN by design — so on onex-dev, the only
AWS_MSK_IAM lane, every in-cluster onex delegate --bus kafka refused before
publishing, with a message about the lane rather than about the client:

cannot confirm a deployed orchestrator is consuming 'onex.cmd.omnimarket.delegate-skill.v1'
... [ONEX_CORE_041_INVALID_CONFIGURATION] sasl_mechanism='AWS_MSK_IAM' needs a token
callback on the confluent client and cannot be expressed as config entries; use an
aiokafka client for this lane

Reproduced in-cluster read-only 2026-09-15T23:38Z on the dev-system cluster, in pod
omninode-runtime-7946cbb694-wv8v7, as the deployed container, while the runtime in that
same pod was consuming the command topic throughout.

What changed

live_consumer_groups (src/omnibase_infra/backends/backend_probe.py) resolves its
client from the same ModelKafkaEventBusConfig and the same auth builder the runtime
uses. One client resolution path, no environment switch deciding the family, and exactly
one implementation of the MSK token callback — the runtime's MSKTokenProvider.

The group state is not on aiokafka's listing, so the probe narrows by the group id's
topic-scope suffix first and then describes only those candidates — a handful, not the
~630 groups a real cluster lists. Kafka spells the state Stable on the wire where the
confluent enum spelled it STABLE; the comparison is case-folded, which is the one
detail a family swap silently gets wrong.

Fail-closed semantics are unchanged and now pinned by tests: an unreachable broker, a
per-group error in the describe, and a missing broker address all raise rather than
returning an empty tuple. The metadata question is still asked first and deliberately.

This is the same class OMN-17304 hit and fixed by threading confluent credentials into
the same construction, which is why it reproduced the moment a mechanism arrived that the
family cannot express at all.

Tests

tests/unit/backends/test_omn18418_lane_probe_client_family.py, red-first:

Test	At parent commit	After
IAM lane resolves liveness	FAIL, the exact live ONEX_CORE_041_INVALID_CONFIGURATION text	pass
SCRAM lane threads credentials, never logs them	FAIL	pass
PLAINTEXT lane acquires no auth parameters	FAIL	pass
non-Stable group is not a live consumer	FAIL	pass
per-group describe error is UNKNOWN	pass (property preserved)	pass
unreachable broker is UNKNOWN, not empty	pass (property preserved)	pass
no broker address is UNKNOWN	pass (property preserved)	pass

RED run at the parent commit: 4 failed, 3 passed. After: 7 passed.

The two OMN-18012 tests that pinned the confluent construction of this one probe moved
here and now assert the aiokafka construction; their property is unchanged, only the
client. The two sibling confluent clients in that module (probe_kafka and the runtime
health monitor) are untouched and still pinned there, with a comment at the seam saying
where the moved coverage went and why.

Verification

• pytest tests/unit/backends/ tests/unit/cli/test_delegate_locus.py tests/unit/event_bus/test_omn18012_confluent_admin_transport.py — 72 passed
• mypy src/omnibase_infra/backends/backend_probe.py --strict — clean
• ruff format + ruff check — clean
• pre-commit run on the staged set — every hook Passed or Skipped
• Pre-push hooks ran on the push (this repo retired its pre-push test leg on 2026-09-10; CI is the enforced gate)

Lab-first note

The .201 dev lane is SCRAM, and SCRAM is expressible as confluent config entries, so
this defect is structurally invisible there — a lab pass cannot cover it. The lane
shapes are covered by unit tests instead, and the live proof is the in-cluster
onex delegate --bus kafka on onex-dev once this reaches it through the normal delivery.
That readback is stated as pending rather than borrowed.

Residual, stated not fixed

probe_kafka and the runtime health monitor's _list_consumer_group_snapshots are still
confluent and still cannot describe an AWS_MSK_IAM lane. They degrade rather than
refuse, so they are noise rather than a blocker, and they already have an owner in
OMN-15364 — whose own live evidence (652 groups enumerated in-pod through aiokafka +
build_aiokafka_auth_kwargs_from_env, describe_consumer_groups returning
state='Stable') is independent confirmation that this is the construction that works
against MSK IAM. Not folded in here: different failure mode, different ticket, and one
ticket per PR.

Evidence-Ticket: OMN-18418
Evidence-Source: OCC#9748

[github-actions]

Hostile Reviewer — adversarial findings (OMN-17492)

Models succeeded: glm-review
Models failed: codex
New finding threads: 0
Deduped (already posted on this PR): 0
Nit-level findings suppressed: 2

The model is the FINDER, never the gate: merge is gated only by the
deterministic Hostile Review Thread Gate, which blocks while
hostile-reviewer threads are unresolved. Resolve each thread after
addressing (or rejecting, with a reply) its finding.

Findings not anchored to a changed file

• [MAJOR] hostile-reviewer (glm-review)

  asyncio.run makes live_consumer_groups unusable from any running loop | The synchronous wrapper now calls asyncio.run unconditionally. Any caller already inside an event loop (a likely future state given the entire runtime is aiokafka-based) gets RuntimeError: asyncio.run() cannot be called from a running event loop, which the broad except converts into ConsumerGroupLivenessUnknownError and the gate fails closed. The docstring acknowledges this and defers it. This is a latent API contract violation in a fun

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MAJOR] hostile-reviewer (glm-review)

  No wall-clock bound on the aiokafka admin round trips | The confluent implementation enforced explicit timeouts at every step: socket.timeout.ms, request.timeout.ms on the client, timeout on list_topics, and future.result(timeout + 1.0). The aiokafka implementation passes only request_timeout_ms to the constructor; describe_cluster and list_consumer_groups receive no timeout argument and aiokafka's internal retry behavior is not bounded by request_timeout_ms alone. A half-open broker connection can now hang

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  admin.close() skipped when start() fails partway | await admin.start() sits outside the try/finally. If start() raises after establishing the connection but before completing initialization (aiokafka's start performs a bootstrap and metadata fetch that can fail late), the client is never closed, leaking the connection and its background task. The finally block only protects the post-start section. | Evidence: await admin.start()
  try:
  ...
  finally:
  await admin.close() | Fix: Move start() inside the tr

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  Positional tuple unpacking of describe responses is fragile | The code assumes each group entry in the describe response is indexable with error_code at 0, group_id at 1, state at 2. aiokafka's DescribeGroupsResponse field layout is an implementation detail of the generated protocol structs and varies across versions; the confluent path used named attributes (group.group_id, group.state). A version bump that reorders or renames fields silently misreads error codes as states or vice versa, and because state

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  Per-group listing errors are no longer detected at the listing stage | The old code inspected listing.errors and failed closed on partial listings. The new code inspects errors only in describe_consumer_groups responses. aiokafka's list_consumer_groups can raise for coordinator-level failures but the diff adds no test or handling for a listing that silently omits groups the caller cannot list (ACL-scoped service accounts will see a filtered, incomplete listing that looks like an authoritative empty answer f

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  Stub class state stored as mutable class attributes | _RecordingAdminClient keeps seen_kwargs, listed_groups, described_states, and closed as class-level mutable attributes mutated across tests. This creates hidden inter-test coupling and order dependence; parallel test execution (pytest-xdist distributes by test, but any in-process ordering change or a future parametrization) can leak state between tests. The init reset covers only seen_kwargs and closed. | Evidence: class _RecordingAdminClient:
  se

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  Unit tests now hard-import aiokafka.admin | The fixture does import aiokafka.admin at runtime rather than synthesizing the module in sys.modules as the deleted confluent tests did with types.ModuleType. In an environment where aiokafka is an optional extra for backend consumers, these 'unit' tests fail on import rather than skipping, changing the test suite's dependency profile as a side effect of the fix. | Evidence: import aiokafka.admin

  bound = _bound_group_id()
  monkeypatch.setattr(aiokafka.admin, "AIOK

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

[github-actions]

⚠️ Hostile Reviewer — DEGRADED (informational)

Critical findings: 1
Major findings: 3
Total findings: 10
Models succeeded: glm-review

Note: Fewer than 2 reviewer models succeeded. Degraded results are informational (OMN-8468/OMN-8524) and do not block merge. Error: cli_review exit 2 (fewer than 2 models succeeded — partial/total outage)

---

Semantics (OMN-17492 — the model finds, thread resolution gates)

Surface	Meaning	Blocks merge?
Review threads	Per-finding, posted by the reviewer	No (informational)
Hostile Review Thread Gate	Deterministic: unresolved hostile-reviewer threads exist	Fails until resolved (not yet a required context)
degraded verdict	Fewer than 2 models succeeded (infra)	No

Powered by omniintelligence.review_pairing.cli_review — multi-model adversarial review: qwen3-review, qwen3-review-b, glm-review (OMN-8468/OMN-8524/OMN-17492)

[github-actions]

Hostile Reviewer — adversarial findings (OMN-17492)

Models succeeded: glm-review
Models failed: codex
New finding threads: 4
Deduped (already posted on this PR): 0
Nit-level findings suppressed: 1

The model is the FINDER, never the gate: merge is gated only by the
deterministic Hostile Review Thread Gate, which blocks while
hostile-reviewer threads are unresolved. Resolve each thread after
addressing (or rejecting, with a reply) its finding.

Findings not anchored to a changed file

• [CRITICAL] hostile-reviewer (glm-review)

  Describe-response shape parsing can silently yield an empty answer | The code iterates for response in described: for group in getattr(response, "groups", []). In aiokafka, describe_consumer_groups returns one DescribeGroupsResponse per requested group id; whether each element exposes a .groups attribute containing the tuple, or is itself a tuple of group fields, varies by aiokafka version. If the attribute access fails, getattr defaults to [] and the loop body never runs, so the function returns () for

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MAJOR] hostile-reviewer (glm-review)

  No wall-clock bound on the probe; gate can hang indefinitely | The confluent implementation enforced timeout three ways (socket.timeout.ms, request.timeout.ms, future.result(timeout + 1.0)). The aiokafka rewrite passes only request_timeout_ms to the client constructor and calls describe_cluster() and list_consumer_groups() with no await-side timeout. A connection that establishes but stalls (half-open TCP, slow coordinator) is not covered by request_timeout_ms for the metadata path, and asyncio.run imposes

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MAJOR] hostile-reviewer (glm-review)

  asyncio.run contract is fragile under any future async or loop-bound caller | The Note asserts there is no async caller today and that one would call _live_consumer_groups_async directly, but the function provides no guard or dispatch. If live_consumer_groups is ever invoked from a thread with a running event loop, or from an async context via a synchronous wrapper, asyncio.run raises RuntimeError, which the generic except converts to ConsumerGroupLivenessUnknownError: a client-side programming error masque

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  Admin client leaked if start() fails after partial connection | The try/finally begins after await admin.start(). If start() raises mid-handshake (auth failure after TCP connect), no close() is called; the aiokafka client holds open sockets and background tasks until GC. The stub-based unit test cannot observe this. | Evidence: await admin.start()
  try:
  await admin.describe_cluster()
  ...
  finally:
  await admin.close() | Fix: Include start() inside the try, or wrap start in its own try/except that c

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  Per-group error handling no longer covers listing-level errors | The old code raised UNKNOWN on listing.errors before narrowing. The new code handles per-group describe errors explicitly, but list_consumer_groups itself can report errors (aiokafka raises or returns error-carrying entries depending on version); any listing-level error now falls into the outer generic except and is converted with a generic message, losing the structured 'incomplete answer' semantics the old message carried. Behavior is still

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

Findings demoted from threads (anchor rejected)

• [MAJOR] hostile-reviewer (glm-review)

  Unit stubs pin a hand-written contract instead of the library's | _RecordingAdminClient, _FakeDescribeResponse, and the listing tuple shape (entry[0] as group id) are invented by this diff. The production code's parsing of list_consumer_groups and describe_consumer_groups return values is validated only against these fakes in unit tests; the fake's describe_cluster returns a dict, which real aiokafka does not. Combined with the getattr-default parsing above, a version bump in aiokafka can break the probe in

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  Shared class-level mutable state across stub instances | _RecordingAdminClient keeps seen_kwargs, listed_groups, described_states, and closed as class attributes mutated by init and async methods. Tests that do not reset listed_groups inherit state from fixture setup order; this works only because _stub_admin reassigns every field. Any future test instantiating the client without the fixture reads stale group listings. The subclass fixtures (Erroring, Unreachable) also mutate the parent's class state vi

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  test_no_broker_address_is_unknown leaves auth env vars set | The test clears only KAFKA_BOOTSTRAP_SERVERS. If the surrounding environment exports KAFKA_SASL_* variables, build_aiokafka_auth_kwargs_from_env runs with foreign credentials. The assertion still holds because the function raises before client construction, but the test no longer isolates the no-broker path from the auth path, weakening its diagnostic value if it fails. | Evidence: def test_no_broker_address_is_unknown(monkeypatch: pytest.MonkeyPa

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).

• [MINOR] hostile-reviewer (glm-review)

  New event loop and full client lifecycle per probe call | The confluent client was constructed per call without loop overhead; the rewrite now creates and tears down an asyncio event loop, an aiokafka admin client, and its background tasks on every invocation. _await_groups in the integration test makes this visible: it polls up to 60 times at one call each, each call paying asyncio.run plus client start/close. In production the delegate gate's polling patterns would multiply this overhead. | Evidence: retu

  Resolve this thread when addressed — the Hostile Review Thread Gate blocks while hostile-reviewer threads are unresolved (OMN-17492).