feat(OMN-15789): EventBusKafka.subscribe(auto_offset_reset=) per-call override by jonahgabriel · Pull Request #2705 · OmniNode-ai/omnibase_infra

jonahgabriel · 2026-08-09T21:44:04Z

Summary

The real-broker-side half of the event_bus_substrate dual-substrate fixture (omnibase_core PR #1550, merged to dev). Adds an optional auto_offset_reset kwarg to EventBusKafka.subscribe(), threaded through to _start_consumer_for_topic_unlocked's new auto_offset_reset_override parameter.

None (the default, every pre-existing call site) preserves current behavior exactly -- falls back to self._config.auto_offset_reset.
Each (topic, group_id) pair already gets its own dedicated AIOKafkaConsumer, so the override is scoped to exactly the one consumer a given subscribe() call creates -- it does not affect any other subscription's offset-reset behavior.
New unit test test_subscribe_threads_auto_offset_reset_override_to_consumer_start proves the override threads through correctly (and that a None override is not silently widened to a truthy value).
Fixed a real regression the governed pre-push selector caught: four TestKafkaEventBusSubscribe tests hand-roll a fake_start(topic, group_id) async double patched over _start_consumer_for_topic_unlocked; the new keyword-only parameter broke their fixed signature. Widened every fake_start/fake_start_unlocked double in that file.

Why split from the fixture/conftest/exemplar-test half

omnibase_infra pins omnibase-core from PyPI only (OMN-14628 removed the git-rev override -- pure-PyPI cross-repo seam). The event_bus_substrate fixture's real_broker leg (tests/integration/event_bus/substrate/) imports omnibase_core.event_bus.testing.* and omnibase_core.event_bus.event_bus_semantic_fake, which are new in omnibase_core PR #1550 and not yet published to PyPI as of this PR. That half is prepared on this same branch but held back (not pushed in this commit) until an omnibase-core release is cut and this repo's pin is bumped -- otherwise the substrate test directory's conftest.py fails to import and breaks collection for its whole subtree.

This EventBusKafka.subscribe() extension itself has no such dependency -- it's independently useful, independently CI-green, and unblocks nothing by landing now.

Test plan

uv run pytest tests/unit/event_bus/test_kafka_event_bus.py -- 121/121 passed
uv run pytest tests/unit/event_bus/test_event_bus_kafka_error_paths.py -- 16/16 passed (includes the new override-threading test)
uv run pytest tests/unit/event_bus/ tests/unit/adapters/ tests/unit/backends/ tests/unit/services/ -- 2984/2984 passed
uv run mypy src/omnibase_infra/event_bus/event_bus_kafka.py --strict -- clean
pre-commit run --all-files on every changed file -- all hooks green
Governed impacted-test selector (pre-push, OMN-13973) -- 9020 passed, 30 skipped, 0 failed

Evidence-Ticket: OMN-15789
Evidence-Source: OCC#6276

… override Adds an optional auto_offset_reset kwarg to EventBusKafka.subscribe(), threaded through to _start_consumer_for_topic_unlocked's new auto_offset_reset_override parameter. None (the default, every pre-existing call site) preserves current behavior exactly -- falls back to self._config.auto_offset_reset. Each (topic, group_id) pair already gets its own dedicated AIOKafkaConsumer, so the override is scoped to exactly the one consumer a given subscribe() call creates. This is the real_broker-side half of the event_bus_substrate fixture's fidelity contract (OMN-15789, omnibase_core PR #1550, merged): the fixture's real_broker leg needs the same per-call auto_offset_reset surface EventBusSemanticFake (omnibase_core) already exposes, so a single EventBusKafka instance can be reused across earliest/latest test legs instead of needing a separate instance per offset policy. New unit test proves the override threads through correctly (and that a None override is not silently widened to a truthy value): test_subscribe_threads_auto_offset_reset_override_to_consumer_start. Landed as its own PR (split from the fixture/conftest/exemplar-test half, which imports omnibase_core.event_bus.testing.* -- unreleased on PyPI as of this commit; omnibase_infra pins omnibase-core from PyPI only, OMN-14628 pure-PyPI seam, no git-rev override) so this independently-useful, independently-CI-green production change is not blocked on that release. The fixture-consuming half is prepared on this same branch's follow-up commit, held pending the release. Evidence-Ticket: OMN-15789

…override kwarg The governed impacted-test selector (pre-push) caught a real regression in 5d55a94: four TestKafkaEventBusSubscribe tests hand-roll a fake_start(topic, group_id) async double patched over _start_consumer_for_topic_unlocked, and the new auto_offset_reset_override keyword-only parameter broke their fixed signature (TypeError: unexpected keyword argument). Widens every fake_start/fake_start_unlocked double in this file to accept the new kwarg. 121/121 passed in tests/unit/event_bus/test_kafka_event_bus.py after the fix; 2984/2984 passed across tests/unit/event_bus/+adapters/+backends/+services/. Evidence-Ticket: OMN-15789

coderabbitai · 2026-08-09T21:44:15Z

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 44 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: f35ee9e1-7eed-4411-b19b-5afbe3ac4244

📥 Commits

Reviewing files that changed from the base of the PR and between 5c43b33 and c721ac4.

📒 Files selected for processing (3)

src/omnibase_infra/event_bus/event_bus_kafka.py
tests/unit/event_bus/test_event_bus_kafka_error_paths.py
tests/unit/event_bus/test_kafka_event_bus.py

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

…ibase_infra#2705 (#6276) * evidence(OMN-15789): author OCC companion for OmniNode-ai/omnibase_infra#2705 OCC companion by node_pr_lifecycle_fix_effect (OMN-13317 F1 / OMN-13990 / OMN-14285). Product PR head c721ac451f79ea7bdf2b84df646288b011e88b4f. * evidence(OMN-15789): self-bind OCC#6276 + rebind contract_sha256 * fix(OMN-15789): revert dod-occ-evidence-admissibility-validator receipt to its merged (OCC#6264) content The auto-bind automation rewrote this shared, generic evidence entry (rebinding it to PR #2705 / this OCC PR's commit) instead of leaving the already-merged receipt untouched, tripping the OCC Append-Only Gate ("merged receipts are immutable. Corrections must be net-new add-only supersession files."). Reverts this one file to its origin/dev (merged, OCC#6264) content; the four genuinely new evidence items for this PR (dod-omnibase_infra-pr-2705, dod-omnibase_infra-pr-2705-ci, occ-self-bind-pr-6276, and their contracts/OMN-15789.yaml entries) are untouched net-new additions. Evidence-Ticket: OMN-15789 --------- Co-authored-by: omnimarket-bot <bot@omninode.ai> Co-authored-by: Jonah Gray <jonah@omninode.ai>

github-actions · 2026-08-09T22:55:47Z

✅ Hostile Reviewer — PASSED

Blocking findings (critical): 0
Total findings: 0
Models succeeded: qwen3-review,qwen3-review-b

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)

…t_reset_override kwarg Test-Failure Ratchet Gate flagged a new failure cluster on PR #2665 (tests.integration.event_bus.test_kafka_concurrent_subscribe_startup). Root cause: dev PR #2705 (OMN-15789, merged 2026-08-09T23:11:14Z) added a keyword-only auto_offset_reset_override param to EventBusKafka._start_consumer_for_topic_unlocked and now passes it from subscribe(); the two tests that stub that method via subscribe() had stale fake_start() signatures that reject the new kwarg. Unrelated to this PR's DB-delegation-workflow-state change (dev drift from #2705, confirmed via git log on both files). Fix: accept auto_offset_reset_override in both fake_start() stubs.

…e signature parity Split 1/15 on head 2c0e206 (run 31351049303/job 93344494250) had 4 reds. Two were dev drift that landed on origin/dev after this branch's last rebase (OMN-15384 #2702 shape-parity gate, OMN-15789 #2705 per-call auto_offset_reset override) -- merged dev in first (no conflicts), which alone fixed the dep-provenance-timeout-adjacent kafka fakes already touched by #2705's own diff. Two fakes in this repo's own test_kafka_concurrent_subscribe_startup.py (not touched by #2705, since that PR only updated tests/unit/event_bus/test_kafka_event_bus.py) still called EventBusKafka._start_consumer_for_topic_unlocked's fake without the new kwarg and broke the two tests that exercise bus.subscribe() directly: TypeError: fake_start() got an unexpected keyword argument 'auto_offset_reset_override' Fixed signature-parity-only, matching the precedent already established in test_kafka_event_bus.py: accept and discard the kwarg (`*, auto_offset_reset_override: str | None = None`), zero assertion changes. The other 5 fake_start doubles in the same file drive bus._subscribers + bus.start_consuming()/shutdown() directly, never through subscribe(), so the kwarg is never passed to them -- left as-is (matches live call-graph, not a latent gap). live_events (the table this PR's own 099 migration creates in omninode_internal, transform-copied from public.live_events) is newly dual-produced now that #2702's flat-vs-node shape-parity gate exists on dev: both docker/migrations/forward/099_create_omninode_internal_live_events.sql and docker/migrations/forward/nodes/node_projection_live_events/0000_create_live_events.sql declare CREATE TABLE IF NOT EXISTS live_events (bare name, schema-qualification stripped by the gate's own table_column_shapes()). Diffed the two column sets by hand: 10/10 columns present on both sides with byte-identical normalized types (id UUID, event_id/type/source/topic/summary/payload/ correlation_id TEXT, timestamp/created_at TIMESTAMPTZ) and identical NOT NULL/DEFAULT text. node_projection_live_events/0001 (the other file in that node dir) is data-only (UPDATE, no CREATE TABLE), so it does not affect the comparison. Added a `status: identical` ledger entry -- no accepted divergence, no reason needed. Third red (test_dep_provenance_lineage_live_omn15604, live GitHub API read timeout) is the documented transient class; not touched here, expected to clear on the fresh push-triggered run per rule 2's second-failure escalation if it doesn't. Local gates: ruff format/check clean, mypy --strict clean on the touched test file (0 new errors; the 7 reported are pre-existing in unrelated helper/conftest files), pre-commit run on touched files all Passed, targeted pytest (tests/integration/event_bus/test_kafka_concurrent_subscribe_startup.py + tests/unit/event_bus/test_kafka_event_bus.py + tests/ci/test_flat_node_migration_shape_parity.py) 155 passed. OMN-15359

…ansform-copy public rows (#2704) * feat(OMN-15359): physically create omninode_internal.live_events + transform-copy public rows P5 build slice, follow-on to #2674's P2-P4 schema-build. Absorbs the handler_wiring UndefinedTable gap the mergesweep-0809-projplane ground phase + adversarial verify live-reconfirmed on onex-dev: handler_wiring issues schema-qualified SQL against contract-declared omninode_internal for live_events (db_io.db_tables[0].schema has said omninode_internal since before this PR), and no migration had ever physically created that relation anywhere in the corpus. - docker/migrations/forward/099_create_omninode_internal_live_events.sql: physically creates omninode_internal.live_events (identical shape to public.live_events), transform-copies every existing row (idempotent, ON CONFLICT DO NOTHING), and reconciles count + key-set + row-content-hash parity in a fail-closed DO block (RAISE EXCEPTION aborts the migration Job on any mismatch). public.live_events is PRESERVED -- no ALTER TABLE SET SCHEMA, no DROP -- per this ticket's own scope text. Live-recounted before writing this migration: 26688 rows, unchanged from the prior same-day ground-phase readback. - physical_schema_mapping.py: removes "live_events" from INTERNAL_TABLES_PHYSICALLY_IN_PUBLIC_UNTIL_OMN15359 now that the physical table exists, so physical_grant_schema_for_table('omninode_internal', 'live_events') agrees with the real INSERT-target schema instead of silently reasserting the closed gap. - Shipped topology TABLE grants regenerated via generate_application_database_table_grants.py --write (57 declarations, --check --prove clean across all 7 profiles) + rendered catalogs regenerated via render_application_database_topology.py: the omninode_runtime principal's live_events grant moves from schema: public to schema: omninode_internal across local/onex-dev/onex-prod instances. - Cross-boundary regression test (the seam gap the mergesweep-0809-projplane-verify adversarial verify named as missing): tests/integration/runtime/test_live_events_projection_write_path_omn15359.py drives a real event through the REAL, unpatched infra dispatch bridge (_make_projection_dispatch_callback -> _build_projection_db_adapter -> ProjectionDatabaseOperations, real psycopg2, real connection-identity attestation against a real omninode_runtime LOGIN role) against a real ephemeral Postgres 16 cluster with 098+099 applied via the same psql -f path production uses. Asserts the row lands in AND reads back from omninode_internal.live_events via an independently-opened connection, and that physical_grant_schema_for_table('omninode_internal', 'live_events') agrees with the resolved INSERT-target schema post-migration. - Migration proof (test_099_omninode_internal_live_events_omn15359.py, same pattern as test_098): creates, copies every row, preserves the source, reproduces row content faithfully, is idempotent on reapply, picks up only the delta on a later reapply, fails closed on a corrupted partial copy (content-hash mismatch), and rollback drops the table while preserving the source. TDD: both new integration test files were run RED-before (099 migration file + rollback moved aside, physical_schema_mapping.py bridge-removal stashed) -- 9 of 10 new tests failed / 1 errored, including the exact live 'ValueError: ... lacks declared write privileges ... INSERT, SELECT, UPDATE on table omninode_internal.live_events' the ground-phase readback predicted -- then GREEN after restoring the fix (10/10 pass). Coordination: this PR alone does NOT fix the schema_migrations 098-never-applied onex-dev deploy gap (mergesweep-0809-projplane-verify, D1) or the runtime-image-not-rebuilt gap (D3) -- those are the omninode_infra Deploy onex-staging workflow's mechanism defects, out of this PR's scope, and must be re-verified live before this migration is treated as deployed on onex-dev. Does not touch ONEX_CORE_081/OMN-15425 (tenant projection authority -- a separate, unrelated 5-point wiring gap, not started). Deferred (named, not silent): the projection_api.exposures[0].schema seam (still `public` in node_projection_live_events/contract.yaml) is fixed in a companion omnimarket PR, held open per its own landing-order gate (read path must keep working against whichever schema is actually live at each deploy step) rather than merged blind alongside this one. Requires (merged first, unblocks this PR's CI): omnimarket#2031, which registers omninode_internal.live_events catalog ownership so this PR's schema-qualified CREATE TABLE clears the OMN-15361 SQL ownership gate. OMN-15359 Evidence-Source: OCC#6273 dod_evidence: uv run pytest tests/integration/migrations/test_099_omninode_internal_live_events_omn15359.py tests/integration/runtime/test_live_events_projection_write_path_omn15359.py -> 10 passed (RED-before captured, see body); scripts/generate_application_database_table_grants.py --check --prove -> 7/7 profiles PASS; scripts/check_schema_fingerprint.py stamp -> regenerated (84 migration files). * fix(OMN-15359): exempt 099 from the schema-qualification SQL scanner 099 is the governed OMN-15359 physical-schema cutover itself: it creates NEW omninode_internal-domain authority (ownership declared via omnimarket#2031) AND, in the same file, transform-copies from the existing legacy public.live_events table. The OMN-15361 schema-qualification scanner admits neither reference form for that cross-schema read (unqualified 'live_events' -> "must be schema-qualified"; explicit 'public.live_events' -> "prohibited in public") -- there is no third form it accepts. This is exactly the class of migration the OMN-15420 cutover-journal machinery (omnibase_infra.migration.cutover) is designed for, but that machinery has no runnable CLI yet (P5 scope, OMN-15426/OMN-15360, not started). Exempted via _LEGACY_DEFAULT_SCHEMA_SQL_EXACT_PATHS rather than blocked on building that machinery from scratch inside this slice; 099's own count/key-set/ row-content-hash reconciliation (fail-closed via RAISE EXCEPTION, proven live in tests/integration/migrations/test_099_omninode_internal_live_events_omn15359.py) is the substitute proof this gate would otherwise provide. Verified: uv run python scripts/ci/check_application_database_sql.py --base-revision origin/dev --head-revision HEAD -> application_database_sql_gate=PASS (previously FAIL on 3 violations: prohibited-in-public, must-be-schema-qualified, dynamic-SQL-DO-block). Full mandatory-source-assertion suite (434 tests) still green. OMN-15359 * chore(OMN-15359): retrigger CI (occ-preflight run stale, workflow file rerun refused) * fix(OMN-15359): address CodeRabbit findings on 099 (3 threads) 1. Reconciliation now requires EXACT row-count parity (<>, not <): a destination with MORE rows than the source is exactly as much a failure as one with fewer. 2. Content-hash reconciliation now includes `id` (the primary key), not just event_id (the dedup key). Without it, a stale destination row whose event_id matches but whose id diverges (e.g. left behind by a prior partial/corrupted copy that ON CONFLICT (event_id) DO NOTHING preserved) would hash identically to the correct row. 3. check_application_database_sql.py: narrowed the 099 exemption to the schema-qualification LINT only. Ownership validation for objects the file CREATES stays active -- an exempted file that creates new schema-qualified authority must still carry a declared owner. This makes the omnimarket#2031 ownership-manifest work load-bearing again (the prior blanket `continue` skipped it entirely). Verified: application_database_sql_gate=PASS with the narrowed exemption. 4. test_099_row_content_is_faithfully_reproduced now compares id, timestamp, and created_at in addition to the previously-checked columns -- a migration that silently changed any of these would previously have passed this test. All 10 migration + write-path tests re-verified green after the fixes. OMN-15359 * fix(OMN-15359): narrow the CodeRabbit ownership fix to created-object checks only The prior commit's exemption narrowing (skip lint only, run BOTH ownership checks) broke the pre-existing OMN-15655 legacy-shape-repair exemption tests: target_requirements fires on ALTER (not just CREATE), and 031/050's own ALTER TABLE public.legacy_shape statements are exactly the "modifying an already-existing legacy table is not new authority" case that exemption category exists to cover -- they were never required to carry an ownership declaration and still shouldn't be. Correct scope: the legacy exemption skips lint AND target_requirements (ALTER/DROP/GRANT/etc against pre-existing relations), but created_catalog_identities (CREATE ... schema.name) stays unconditional -- that's the actual gap CodeRabbit found, and it's what 099's CREATE TABLE omninode_internal.live_events needs enforced. Verified: application_database_sql_gate=PASS on the real diff; tests/ci/test_application_database_sql_gate.py + test_application_database_sql_authority_isolation.py + test_application_database_domain_enforcement_contract.py -> 39 passed (including the previously-broken OMN-15655 exemption test). OMN-15359 * fix(OMN-15359): 099 tolerates public.live_events being absent (Migration Integration Test) Live CI failure on this PR: omnibase_infra's standalone "Migration Integration Test" gate applies only the numbered top-level docker/migrations/forward/*.sql files (024...098, no nodes/ subtree) against a fresh database -- it does not run the node-owned migration stream that vendors public.live_events's own CREATE TABLE. 099 unconditionally referenced public.live_events and failed UndefinedTable in that scope: asyncpg.exceptions.UndefinedTableError: relation "public.live_events" does not exist Guarded the transform-copy + reconciliation behind `to_regclass('public.live_events') IS NULL` (a plain plpgsql IF, not dynamic SQL/EXECUTE -- the embedded INSERT/SELECT statements are only planned when control flow actually reaches them, so an unreached branch never resolves the possibly-missing table). When absent, the migration RAISE NOTICEs and succeeds with omninode_internal.live_events created empty. Production apply (node migrations already ran, public.live_events has real rows) is unaffected -- same copy + reconciliation logic, now reached because the guard is true. Added test_099_succeeds_when_public_live_events_does_not_exist, reproducing the CI gate's own scope (098 applied, no node migrations, no seeded source table) and asserting 099 still succeeds. All 11 migration + write-path tests green. OMN-15359 * fix(OMN-15359): 099 issues the omninode_runtime grant it was missing (projplane-slice-verify) 099 physically created omninode_internal.live_events but never granted anything on it -- proven 4 ways on a live RDS snapshot: the role carries zero table privileges, no CREATE on the schema, and pg_default_acl is empty. The only deploy-time grant mechanism (000_create_multiple_databases.sh) only ever touches schema public, and 098's own docstring explicitly defers "owner_omninode_internal/omninode_runtime role creation" as P5 scope. Because live_events is simultaneously removed from the physical bridge in this same PR, it is -- unlike the other ~40 still-bridged families -- live TODAY: deploying 099 as shipped converts the original UndefinedTable failure into InsufficientPrivilege on the very first write. 099 now: - guard-creates omninode_runtime (NOLOGIN, matching the 094/096 precedent exactly -- role provisioned only if absent, RAISE EXCEPTION if the executing role lacks CREATEROLE and the role is not already there; LOGIN + password stays a deployment-owned attach, never re-asserted here even though the topology declares login: true) - grants USAGE ON SCHEMA omninode_internal + SELECT/INSERT/UPDATE ON omninode_internal.live_events, derived from and matching src/omnibase_infra/topology/instances/*.yaml's principals.omninode_runtime.grants declaration byte for byte - adds ALTER DEFAULT PRIVILEGES IN SCHEMA omninode_internal so the ~40 remaining OMNINODE_INTERNAL-domain families 098 defers do not each need to repeat this same GRANT boilerplate by hand - adds a statically-provable post-condition asserting the INSERT grant landed, matching the file's existing assertion pattern Chose to grow 099 in place rather than a companion 100: 099 has not merged to dev and has never applied anywhere, so a companion migration would be pure churn for a migration with zero deployed history. De-masks the cross-boundary write-path test in the same PR (test_live_events_projection_write_path_omn15359.py): the fixture previously hand-issued CREATE ROLE ... LOGIN + GRANT statements independent of 099, proving the dispatch/adapter path while silently supplying the one thing 099 itself never produced. The fixture now applies 098 + 099 exactly as production does and only attaches LOGIN + a password afterward (the one deployment-owned step a migration never performs, mirroring 094/096's own stated invariant). Two new tests derive the expected grant set from the shipped topology (application_topology(), never hand-typed) and assert it against live information_schema.role_table_grants / has_schema_privilege state -- exact equality, not a subset check. RED-first, verified live (2026-08-10): reverted 099 to its pre-fix content, applied 098 + unfixed 099 against a fresh ephemeral Postgres 16 cluster, manually provisioned omninode_runtime out-of-band (mirroring RDS), and confirmed the write path fails with psycopg2.errors.InsufficientPrivilege: permission denied for schema omninode_internal -- not merely UndefinedTable, confirming this is specifically a grant gap. Restored the fix; the full de-masked test file (4 tests) then passes green. Also verified the fixed 099 through the exact CI "Migration Integration Test" path (scripts/run-migrations.py against a completely fresh cluster, migrations 001-099 in sequence): omninode_runtime ends with exactly SELECT/INSERT/UPDATE on omninode_internal.live_events, USAGE on the schema, and NOLOGIN/NOSUPERUSER/NOBYPASSRLS. Local gates: pre-commit run --all-files on touched files (all Passed), ruff format + check --fix, mypy (zero new errors -- the 6 reported errors are pre-existing, unrelated to this change), schema_fingerprint stamp regenerated, tests/integration/migrations/ (64 passed / 2 opt-in skipped), tests/unit/topology/ (107 passed), tests/integration/runtime/ (450 passed; 31 pre-existing/environmental errors from an unrelated shared local Postgres instance in recovery mode, not the ephemeral-cluster fixture this change touches). OMN-15359 * fix(OMN-15359): shape-parity ledger entry for live_events + kafka fake signature parity Split 1/15 on head 2c0e206 (run 31351049303/job 93344494250) had 4 reds. Two were dev drift that landed on origin/dev after this branch's last rebase (OMN-15384 #2702 shape-parity gate, OMN-15789 #2705 per-call auto_offset_reset override) -- merged dev in first (no conflicts), which alone fixed the dep-provenance-timeout-adjacent kafka fakes already touched by #2705's own diff. Two fakes in this repo's own test_kafka_concurrent_subscribe_startup.py (not touched by #2705, since that PR only updated tests/unit/event_bus/test_kafka_event_bus.py) still called EventBusKafka._start_consumer_for_topic_unlocked's fake without the new kwarg and broke the two tests that exercise bus.subscribe() directly: TypeError: fake_start() got an unexpected keyword argument 'auto_offset_reset_override' Fixed signature-parity-only, matching the precedent already established in test_kafka_event_bus.py: accept and discard the kwarg (`*, auto_offset_reset_override: str | None = None`), zero assertion changes. The other 5 fake_start doubles in the same file drive bus._subscribers + bus.start_consuming()/shutdown() directly, never through subscribe(), so the kwarg is never passed to them -- left as-is (matches live call-graph, not a latent gap). live_events (the table this PR's own 099 migration creates in omninode_internal, transform-copied from public.live_events) is newly dual-produced now that #2702's flat-vs-node shape-parity gate exists on dev: both docker/migrations/forward/099_create_omninode_internal_live_events.sql and docker/migrations/forward/nodes/node_projection_live_events/0000_create_live_events.sql declare CREATE TABLE IF NOT EXISTS live_events (bare name, schema-qualification stripped by the gate's own table_column_shapes()). Diffed the two column sets by hand: 10/10 columns present on both sides with byte-identical normalized types (id UUID, event_id/type/source/topic/summary/payload/ correlation_id TEXT, timestamp/created_at TIMESTAMPTZ) and identical NOT NULL/DEFAULT text. node_projection_live_events/0001 (the other file in that node dir) is data-only (UPDATE, no CREATE TABLE), so it does not affect the comparison. Added a `status: identical` ledger entry -- no accepted divergence, no reason needed. Third red (test_dep_provenance_lineage_live_omn15604, live GitHub API read timeout) is the documented transient class; not touched here, expected to clear on the fresh push-triggered run per rule 2's second-failure escalation if it doesn't. Local gates: ruff format/check clean, mypy --strict clean on the touched test file (0 new errors; the 7 reported are pre-existing in unrelated helper/conftest files), pre-commit run on touched files all Passed, targeted pytest (tests/integration/event_bus/test_kafka_concurrent_subscribe_startup.py + tests/unit/event_bus/test_kafka_event_bus.py + tests/ci/test_flat_node_migration_shape_parity.py) 155 passed. OMN-15359

…set_override seam PR #2681 (OMN-15700) landed on top of dev's #2705 (OMN-15789), which made EventBusKafka.subscribe() unconditionally pass auto_offset_reset_override= into _start_consumer_for_topic_unlocked(). The contractor's fake capture_consumer_start(topic, group_id) in test_savings_estimator_topics_keep_distinct_effective_consumer_groups predates that seam and only accepted the original two positional params, so the keyword-only argument raised TypeError deterministically in CI (Tests Split 3/15) — a seam collision between two individually correct, individually green PRs (OMN-14208 class), not a defect in either. Widen the fake's signature to field-match the real _start_consumer_for_topic_unlocked seam exactly (name + keyword-only + default None) and record the captured value alongside topic/group_id so the double mirrors the seam instead of swallowing it. No production code touched; the test's consumer-group-distinctness assertions are unchanged.

* fix(OMN-15700): derive savings consumer groups canonically * test(OMN-15700): harden savings subscription guard * test(OMN-15700): prove per-topic consumer isolation * test(OMN-15700): exercise public subscription path * fix(OMN-15700): match savings-estimator test double to auto_offset_reset_override seam PR #2681 (OMN-15700) landed on top of dev's #2705 (OMN-15789), which made EventBusKafka.subscribe() unconditionally pass auto_offset_reset_override= into _start_consumer_for_topic_unlocked(). The contractor's fake capture_consumer_start(topic, group_id) in test_savings_estimator_topics_keep_distinct_effective_consumer_groups predates that seam and only accepted the original two positional params, so the keyword-only argument raised TypeError deterministically in CI (Tests Split 3/15) — a seam collision between two individually correct, individually green PRs (OMN-14208 class), not a defect in either. Widen the fake's signature to field-match the real _start_consumer_for_topic_unlocked seam exactly (name + keyword-only + default None) and record the captured value alongside topic/group_id so the double mirrors the seam instead of swallowing it. No production code touched; the test's consumer-group-distinctness assertions are unchanged. --------- Co-authored-by: omarashrafwellx <omar@wellxai.com> Co-authored-by: jonahgabriel <jonah.gabriel@gmail.com>

…ERNAL_CONTEXTS (#2730) The "Integration Test Removal Gate" job (OMN-8732) hard-blocks a PR that deletes a tests/integration/*.py file without a replacement and its own workflow header says "No override mechanism" -- but before this change it was absent from all three omnibase_infra dev enforcement surfaces: branch protection (required_status_checks = exactly ["CI Summary"]), EXPECTED_EXTERNAL_CONTEXTS, and MEASURED_NOT_ENFORCED_CONTEXTS. PR #2720 (head ce4e88f) merged 2026-08-11T04:47:58Z with this context reporting `failure`. Measured 16/16 present, 15/16 green over two independent 16-PR windows: the original OMN-15496 seed window 2026-07-29T23:04Z -> 2026-07-30T14:54Z (#2546...#2567, backfilled into omn15496_merge_time_external_check_runs.json -- 16/16 green there) and the current window 2026-08-09T23:11Z -> 2026-08-11T04:47:58Z (#2705...#2720, tests/ci/fixtures/omn15979_merge_time_external_check_runs.json -- 15/16 green). The one red, #2720, is root-caused: its job (id 93668844708) recorded ZERO steps -- a self-hosted-runner dispatch failure -- and the PR's diff touched only .github/workflows/build-and-push-runtime.yml (no tests/integration files), so the gate's own substantive check logic was never at risk of a legitimate red. TDD: TestIntegrationTestRemovalGateExternalContext added RED (verified by stashing the tuple entry and re-running -- 5 failures for the right reasons) before the fix went GREEN. Includes a synthetic-red regression (test_synthetic_red_flips_a_clean_pr_to_failure) proving a red check-run on this context flips CI Summary's verdict, and a live-replay of PR #2720's real payload proving it would now be blocked end-to-end. Ticket: OMN-15979

… block on infra (#2973) OMN-16876 census item 2. Both checks ran on every omnibase_infra PR and could not block a merge. That is sharper here than in any other repo: `dev` requires exactly ONE status check, "CI Summary" (the OMN-4497 single-umbrella design), so EXPECTED_EXTERNAL_CONTEXTS in scripts/ci/ci_summary_gate.py IS the entire external enforcement surface. A context missing from that tuple has no second surface to fall back on and no branch-protection signal that it is missing. Both entries close a Done ticket that live state contradicts: - OMN-13328 ("receipt-honesty REQUIRED on all repos that produce receipts") -- its OCC contract asserted omnibase_infra, omniclaude and omnimarket "were already flipped". Live readback 2026-08-28: none of the three was enforced on any surface. - OMN-13326 ("contract-validation REQUIRED across all 6 repos") -- genuinely required on omnibase_core, omnibase_compat, omnibase_spi, omniintelligence and omnidash; wired on neither omnibase_infra nor omniclaude. Admission, measured not assumed: - #2955..#2970 (2026-08-28): 16/16 present, 16/16 green for both. - Backfilled BOTH existing merge-time fixtures with real API reads under the fixtures own merge-time filter (started_at <= mergedAt), not synthesised rows: omn15496 (#2546..#2567) and omn15979 (#2705..#2720). The second window independently measured 32/32 present, 32/32 green across both contexts. Three independent windows agree; no freeze-baseline ratchet is indicated. Non-vacuity (OMN-16876 finding 5) -- 16/16 green is also exactly the vacuous pass shape, so each was proven able to FAIL on real input: receipt-honesty gamed receipt (verifier == runner) -> exit 1; real -> 0 contract-validation schema-invalid contract -> exit 1; OMN-10041.yaml -> 0 Both producers declare pull_request AND merge_group and carry no `needs:` and no job-level `if:`, so no upstream failure can skip them; and this layer admits only "success", so a skip fails closed rather than passing. Evidence-Source: OCC#7433

jonahgabriel added 2 commits August 9, 2026 17:25

onexbot-occ-writer Bot mentioned this pull request Aug 9, 2026

evidence(OMN-15789): OCC Evidence-Source autobind for OmniNode-ai/omnibase_infra#2705 OmniNode-ai/onex_change_control#6276

Merged

jonahgabriel merged commit 603f1ae into dev Aug 9, 2026
147 of 158 checks passed

jonahgabriel deleted the jonahgabriel/omn-15789-dual-substrate-real-broker-leg branch August 9, 2026 23:11

jonahgabriel mentioned this pull request Aug 10, 2026

feat(OMN-15359): physically create omninode_internal.live_events + transform-copy public rows #2704

Merged

5 tasks done

jonahgabriel mentioned this pull request Aug 12, 2026

fix(OMN-15979): admit Integration Test Removal Gate into EXPECTED_EXTERNAL_CONTEXTS #2730

Merged

jonahgabriel mentioned this pull request Aug 28, 2026

fix(OMN-16878): make receipt-honesty and contract-validation actually block on infra #2973

Merged

Navigation Menu

Repository navigation

feat(OMN-15789): EventBusKafka.subscribe(auto_offset_reset=) per-call override - #2705

feat(OMN-15789): EventBusKafka.subscribe(auto_offset_reset=) per-call override#2705
jonahgabriel merged 2 commits into
devfrom
jonahgabriel/omn-15789-dual-substrate-real-broker-leg

jonahgabriel commented Aug 9, 2026 •

edited by onexbot-occ-writer Bot

Loading

Uh oh!

coderabbitai Bot commented Aug 9, 2026

Review limit reached

Uh oh!

github-actions Bot commented Aug 9, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

jonahgabriel commented Aug 9, 2026 • edited by onexbot-occ-writer Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Why split from the fixture/conftest/exemplar-test half

Test plan

Uh oh!

coderabbitai Bot commented Aug 9, 2026

Review limit reached

Uh oh!

github-actions Bot commented Aug 9, 2026

✅ Hostile Reviewer — PASSED

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jonahgabriel commented Aug 9, 2026 •

edited by onexbot-occ-writer Bot

Loading