Skip to content

feat: PostgresDataSource adapter for data verification [OMN-7624] - #6

Merged
jonahgabriel merged 2 commits into
mainfrom
jonah/omn-7624-task-3-post-pipeline-data-validation-gate
Apr 6, 2026
Merged

jonahgabriel merged 2 commits into
mainfrom
jonah/omn-7624-task-3-post-pipeline-data-validation-gate

Conversation

@jonahgabriel

@jonahgabriel jonahgabriel commented Apr 6, 2026 •

Copy link
Copy Markdown
Contributor

Summary

  • Implements PostgresDataSource class satisfying the existing DataSource protocol, backed by psycopg2 connecting to omnidash_analytics on .201:5436
  • SQL injection prevention via strict regex table name validation (^[a-zA-Z_][a-zA-Z0-9_]*$)
  • Reads DSN from OMNIDASH_ANALYTICS_DB_URL env var or constructor parameter
  • Adds psycopg2-binary>=2.9.10 to project dependencies
  • 6 unit tests (mocked psycopg2, no DB needed) + 4 integration tests (require real DB, auto-skipped in CI)

Test plan

  • 6 unit tests pass: row count, columns, sample rows, None→empty string, handler protocol check, missing DSN error
  • 4 integration tests auto-skip without OMNIDASH_ANALYTICS_DB_URL
  • All 194 existing tests still pass (0 regressions)

Closes OMN-7624 (sub-task 3a: runtime adapter)

Summary by CodeRabbit

  • Refactor

    • PostgreSQL data source now uses DSN string for configuration instead of individual parameters.
    • Improved connection management and lifecycle handling.
    • Enhanced table name validation and schema filtering.
  • Tests

    • Added comprehensive unit and integration tests for PostgreSQL data source functionality.
  • Chores

    • Added psycopg2-binary dependency.

@linear

linear Bot commented Apr 6, 2026

Copy link
Copy Markdown

OMN-7624

@coderabbitai

coderabbitai Bot commented Apr 6, 2026 •

Copy link
Copy Markdown

Caution

Review failed

The pull request is closed.

ℹ️ Recent review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Lite

Run ID: 78b3a442-5e3f-4736-ad0b-5b1f577154b2

📥 Commits

Reviewing files that changed from the base of the PR and between 6da5507 and 67e6b6f.

⛔ Files ignored due to path filters (1)
  • uv.lock is excluded by !**/*.lock
📒 Files selected for processing (3)
  • pyproject.toml
  • src/omnimarket/nodes/node_data_verification/handlers/datasource_postgres.py
  • tests/test_datasource_postgres.py

📝 Walkthrough

Walkthrough

This pull request adds psycopg2-binary as a runtime dependency and refactors the PostgresDataSource class to use DSN-based connection configuration instead of individual host/port/database parameters. The connection lifecycle is redesigned to handle auto-reconnection and query execution improvements including stricter SQL identifier validation and enhanced result normalization. A comprehensive test suite is introduced covering unit and integration scenarios.

Changes

Cohort / File(s) Summary
Dependency Addition
pyproject.toml
Added psycopg2-binary>=2.9.10 to the package's runtime dependencies.
PostgreSQL DataSource Refactoring
src/omnimarket/nodes/node_data_verification/handlers/datasource_postgres.py
Redesigned PostgresDataSource.__init__() to accept dsn: str | None with validation against OMNIDASH_ANALYTICS_DB_URL; removed from_env() classmethod and parameterized constructor. Replaced connection management with lazy initialization and auto-reconnection in _get_conn() with autocommit = True. Introduced _quote_table() with stricter regex validation for table names. Updated query methods: get_row_count() uses SELECT COUNT(*), get_sample_rows() uses RealDictCursor with value stringification and fallback from ORDER BY created_at DESC to LIMIT-only on undefined column errors. Enhanced schema introspection in get_columns() with table_schema = 'public' filtering. Added __del__() and improved close() logging.
Test Coverage
tests/test_datasource_postgres.py
New comprehensive test file with unit tests mocking psycopg2.connect to validate initialization errors, row counting, column retrieval, sample row normalization, and integration with HandlerDataVerification. Integration tests (gated by OMNIDASH_ANALYTICS_DB_URL) verify real database interactions including row counts, column schemas, value stringification, and end-to-end verification workflows with proper resource cleanup.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Poem

🐰 A rabbit hops through DSN strings,
No more host params—cleaner things!
psycopg2 connects with flair,
From env to DSN, handled with care,
Tests verify each hop and bound! 🎉


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

Wire real DB connectivity for node_data_verification so it can run
against omnidash_analytics on .201:5436. Implements the DataSource
protocol using psycopg2 with SQL injection prevention via regex-based
table name validation. Includes 6 unit tests (mocked) and 4 integration
tests (skipped without DB).
- Add type: ignore[import-untyped] for psycopg2
- Fix import sorting (I001)
- Combine nested with statements (SIM117)
- Add SLF001 to per-file test ignores
- Run ruff format
@jonahgabriel
jonahgabriel force-pushed the jonah/omn-7624-task-3-post-pipeline-data-validation-gate branch from 499f0f0 to 67e6b6f Compare April 6, 2026 15:51
@jonahgabriel
jonahgabriel merged commit 5f88d0d into main Apr 6, 2026
3 of 4 checks passed
jonahgabriel added a commit that referenced this pull request Apr 16, 2026
Adds test_runtime_local_publishes_terminal_event_to_bus which asserts
RuntimeLocal._events_received['(terminal)'] >= 1 after an L1 run on
node_merge_sweep. Depends on the runtime behavior decision in the paired
omnibase_core commit (publish-after-classify in _run_single_handler).

Proves L1B plan acceptance gate #6 — the terminal topic
onex.evt.omnimarket.merge-sweep-completed.v1 participates in the bus
during sync-return handler invocations.

All 3 proof tests green (baseline + L1A + L1B). Plan § Task 5.
jonahgabriel added a commit that referenced this pull request Apr 16, 2026
Adds test_runtime_local_publishes_terminal_event_to_bus which asserts
RuntimeLocal._events_received['(terminal)'] >= 1 after an L1 run on
node_merge_sweep. Depends on the runtime behavior decision in the paired
omnibase_core commit (publish-after-classify in _run_single_handler).

Proves L1B plan acceptance gate #6 — the terminal topic
onex.evt.omnimarket.merge-sweep-completed.v1 participates in the bus
during sync-return handler invocations.

All 3 proof tests green (baseline + L1A + L1B). Plan § Task 5.
github-merge-queue Bot pushed a commit that referenced this pull request Apr 16, 2026
…payload [OMN-8939, OMN-8940] (#317)

* test: baseline L1 probe — RuntimeLocal runs merge_sweep with defaults [OMN-8937]

* feat(proof): L1A — onex node node_merge_sweep runs end-to-end on real payload [OMN-8939]

* test(proof): L1B — terminal event transits EventBusInmemory [OMN-8940]

Adds test_runtime_local_publishes_terminal_event_to_bus which asserts
RuntimeLocal._events_received['(terminal)'] >= 1 after an L1 run on
node_merge_sweep. Depends on the runtime behavior decision in the paired
omnibase_core commit (publish-after-classify in _run_single_handler).

Proves L1B plan acceptance gate #6 — the terminal topic
onex.evt.omnimarket.merge-sweep-completed.v1 participates in the bus
during sync-return handler invocations.

All 3 proof tests green (baseline + L1A + L1B). Plan § Task 5.

* fix(ci): update E2E runner to use onex node instead of onex run [OMN-8938]

onex run was removed in OMN-8938 in favour of onex node <name>.
andywu42 pushed a commit to andywu42/omnimarket that referenced this pull request Jul 6, 2026
* feat(OMN-13860): add hardcoded-paths check to node_aislop_sweep

aislop_sweep's SKILL.md already documented a `hardcoded-paths` check
category in its --checks argument, but the handler never implemented
it — a live enforcement hole for CLAUDE.md rule OmniNode-ai#6 (no /Users/... or
/Volumes/... literals in source). Adds the check, mirroring the
ARCH-005 invariant already enforced by node_architectural_invariant_loop,
and honors the existing `# local-path-ok` / `# onex-allow-internal-ip`
/ `# noqa` allowlist annotations on the same line as the flagged
literal.

Verified against a planted /Users/ canary: un-annotated literal is
flagged (ERROR/HIGH), annotated literal is suppressed.

* fix(OMN-13860): avoid leaked-literal canary in hardcoded-paths test

The /Volumes/PRO-G40 canary in test_detects_volumes_path tripped the
repo's Leaked Literals Gate (OMN-10580), which specifically matches
that real machine-mount literal. Swap to a placeholder disk name that
still exercises the /Volumes/ detection pattern without leaking a real
identifier.
andywu42 pushed a commit to andywu42/omnimarket that referenced this pull request Jul 6, 2026
…erministic ruff-fix path (OmniNode-ai#1612)

* feat(OMN-13940): WS-D/D2 merge-sweep delegation harness Slice 0 — deterministic ruff-fix path

Builds the delegation harness end-to-end with zero LLM risk, proving the
plumbing before any model-authored diff lands in Slice 1:

- Splits RECEIPT_FAILURE out of the CODE_FAILURE/CHANGES_REQUESTED branch in
  handler_pr_lifecycle_fix._route (safety bar OmniNode-ai#5) — it stays on the Claude
  agent path unconditionally.
- Adds a delegation eligibility gate (path/keyword denylist + blast-radius
  size gate + persistent two-strike escalation) that only routes small,
  non-sensitive CODE_FAILURE/CHANGES_REQUESTED fixes to a delegated path.
- New node node_pr_delegated_fix_effect: resolves a worktree, runs
  deterministic `ruff format` + `ruff check --fix` (no LLM), re-checks the
  denylist/size gate against the ACTUAL diff produced, commits with a
  `delegated-by: ruff-deterministic` trailer, then shells out to the
  existing node_pr_polish CLI (--skip-repair-dispatch --no-automerge) so the
  established gate/verify/push/coderabbit-triage flow runs unchanged and a
  delegated fix never self-arms auto-merge (safety bar OmniNode-ai#6).
- Fixes a pre-existing bug: node_pr_polish always compiled/dispatched a
  repair-worker (Claude) agent spec even when an already-fixed worktree_path
  was supplied. New `skip_repair_dispatch` command flag closes it.
- Wires delegation counters (delegated/outcome/cost) through
  ModelPrLifecycleFixResult, the orchestrator's FixResult aggregate, and the
  sweep-level ModelPrLifecycleResult.
- Shared command/result models live in omnimarket.events.pr_delegated_fix
  (not node-local) so the fix-effect bridge adapter and the new node avoid a
  cross-node model reach-in; the pr_polish re-entry shells to its CLI rather
  than importing its models in-process for the same reason.

Safety-bar tests: eligible fix accepted with agent never called, denylisted
path/content refused before any attempt, two-strike permanently escalates
after a second failure, and a gate failure never force-pushes.

* fix(OMN-13940): declare skip_repair_dispatch in contract + baseline new dep-health findings

- node_pr_polish/contract.yaml was missing the skip_repair_dispatch input
  declaration added to ModelPrPolishStartCommand/CLI in the previous commit,
  causing market-skill-baseline's input-drift check to report pr_polish as
  degraded. Declaring it restores contract/model parity (0 degraded, was 1).
- Baseline the two dep-health findings intrinsic to any new node
  (UNTESTED_HANDLER — the sweep only scans src/, never tests/, so it can
  never see this PR's real test coverage; and the terminal-event
  MISSING_TOPIC_EDGE — node_pr_delegated_fix_effect's completed event has no
  bus subscriber in this in-process Slice 0). Same pattern as
  node_env_parity_collect_effect (OMN-13925) and
  node_pr_lifecycle_worktree_prune_effect.

* fix(OMN-13940): clear stale _health_cache before delegation transport tests

CI's "Tests (Split N/20)" shard reliably failed
test_local_dispatch_evidence.py after pytest-split reshuffled shard
boundaries because of this PR's new test files. Root cause: pytest-split
merely changed which pre-existing test now lands in the same worker process
ahead of these tests; handler_llm_delegation_call caches probe_health
results in a module-level dict keyed by endpoint_url with a TTL, and the
cache lookup happens before the monkeypatched fake_probe_health in this file
is ever called — a cached negative result from an earlier test in the same
worker silently short-circuits these tests' own transport patch. Clearing
the cache before patching makes each test's own probe result authoritative
regardless of shard/run order. Confirmed locally: full suite and this file
in isolation both pass; the fix is scoped to a pre-existing cross-test
global-state hazard, unrelated to this PR's actual delegation-harness scope.

* fix(OMN-13940): make health-cache clear autouse — one prior fix site was missed

The previous commit only cleared _health_cache inside _patch_transport and
_patch_transport_with_content. test_local_dispatch_reaches_lan_endpoint_via_curl_on_macos_profile
patches subprocess.run directly (the curl transport path) and never calls
either helper, so it still failed under the same shard-ordering hazard.
Converts the fix to an autouse fixture so every test in the file clears the
cache regardless of which transport-patching path it uses.

* fix(OMN-13940): declare runtime closeout external command topics
github-actions Bot pushed a commit that referenced this pull request Jul 8, 2026
* feat(OMN-13940): WS-D/D2 merge-sweep delegation harness Slice 0 — deterministic ruff-fix path

Builds the delegation harness end-to-end with zero LLM risk, proving the
plumbing before any model-authored diff lands in Slice 1:

- Splits RECEIPT_FAILURE out of the CODE_FAILURE/CHANGES_REQUESTED branch in
  handler_pr_lifecycle_fix._route (safety bar #5) — it stays on the Claude
  agent path unconditionally.
- Adds a delegation eligibility gate (path/keyword denylist + blast-radius
  size gate + persistent two-strike escalation) that only routes small,
  non-sensitive CODE_FAILURE/CHANGES_REQUESTED fixes to a delegated path.
- New node node_pr_delegated_fix_effect: resolves a worktree, runs
  deterministic `ruff format` + `ruff check --fix` (no LLM), re-checks the
  denylist/size gate against the ACTUAL diff produced, commits with a
  `delegated-by: ruff-deterministic` trailer, then shells out to the
  existing node_pr_polish CLI (--skip-repair-dispatch --no-automerge) so the
  established gate/verify/push/coderabbit-triage flow runs unchanged and a
  delegated fix never self-arms auto-merge (safety bar #6).
- Fixes a pre-existing bug: node_pr_polish always compiled/dispatched a
  repair-worker (Claude) agent spec even when an already-fixed worktree_path
  was supplied. New `skip_repair_dispatch` command flag closes it.
- Wires delegation counters (delegated/outcome/cost) through
  ModelPrLifecycleFixResult, the orchestrator's FixResult aggregate, and the
  sweep-level ModelPrLifecycleResult.
- Shared command/result models live in omnimarket.events.pr_delegated_fix
  (not node-local) so the fix-effect bridge adapter and the new node avoid a
  cross-node model reach-in; the pr_polish re-entry shells to its CLI rather
  than importing its models in-process for the same reason.

Safety-bar tests: eligible fix accepted with agent never called, denylisted
path/content refused before any attempt, two-strike permanently escalates
after a second failure, and a gate failure never force-pushes.

* fix(OMN-13940): declare skip_repair_dispatch in contract + baseline new dep-health findings

- node_pr_polish/contract.yaml was missing the skip_repair_dispatch input
  declaration added to ModelPrPolishStartCommand/CLI in the previous commit,
  causing market-skill-baseline's input-drift check to report pr_polish as
  degraded. Declaring it restores contract/model parity (0 degraded, was 1).
- Baseline the two dep-health findings intrinsic to any new node
  (UNTESTED_HANDLER — the sweep only scans src/, never tests/, so it can
  never see this PR's real test coverage; and the terminal-event
  MISSING_TOPIC_EDGE — node_pr_delegated_fix_effect's completed event has no
  bus subscriber in this in-process Slice 0). Same pattern as
  node_env_parity_collect_effect (OMN-13925) and
  node_pr_lifecycle_worktree_prune_effect.

* fix(OMN-13940): clear stale _health_cache before delegation transport tests

CI's "Tests (Split N/20)" shard reliably failed
test_local_dispatch_evidence.py after pytest-split reshuffled shard
boundaries because of this PR's new test files. Root cause: pytest-split
merely changed which pre-existing test now lands in the same worker process
ahead of these tests; handler_llm_delegation_call caches probe_health
results in a module-level dict keyed by endpoint_url with a TTL, and the
cache lookup happens before the monkeypatched fake_probe_health in this file
is ever called — a cached negative result from an earlier test in the same
worker silently short-circuits these tests' own transport patch. Clearing
the cache before patching makes each test's own probe result authoritative
regardless of shard/run order. Confirmed locally: full suite and this file
in isolation both pass; the fix is scoped to a pre-existing cross-test
global-state hazard, unrelated to this PR's actual delegation-harness scope.

* fix(OMN-13940): make health-cache clear autouse — one prior fix site was missed

The previous commit only cleared _health_cache inside _patch_transport and
_patch_transport_with_content. test_local_dispatch_reaches_lan_endpoint_via_curl_on_macos_profile
patches subprocess.run directly (the curl transport path) and never calls
either helper, so it still failed under the same shard-ordering hazard.
Converts the fix to an autouse fixture so every test in the file clears the
cache regardless of which transport-patching path it uses.

* fix(OMN-13940): declare runtime closeout external command topics

* fix(OMN-13940): stop merge sweep queue spin

* fix(OMN-13940): run merge sweep every thirty minutes

* fix(OMN-13940): normalize merge sweep repo filters

* chore(OMN-13940): format merge sweep handler on current dev

* fix(OMN-13940): route network check flakes to rerun

* fix(OMN-13940): classify service container flakes for rerun

* fix(OMN-13940): classify docker extraction flakes for rerun

* fix(OMN-13940): classify package network flakes for rerun

* fix(OMN-13940): keep merge sweep loop standing

* ci(OMN-13940): refresh missing required checks

* fix(OMN-13940): stop open remainder sweep spin

* test(OMN-13940): align sweep shard expectations

* test(OMN-13940): format datasource postgres imports

* fix(OMN-13940): address merge sweep review blockers
jonahgabriel added a commit that referenced this pull request Jul 13, 2026
…rops hardcoded LAN IP

* fix(OMN-14531): data_flow_sweep fails closed on lag-probe failure + drops hardcoded LAN IP

Two structural-blindness defects in node_data_flow_sweep, the detector
backing onex:data_flow_sweep (OMN-14531 detector-shelf audit):

1. Fail-open on rpk failure: probe_consumer_lag() treated any non-zero
   `rpk group describe` return code as lag=0 (healthy), so a broker/group
   probe failure was silently indistinguishable from "no lag" and could let
   an otherwise-broken flow classify FLOWING. It now returns
   (lag, lag_unknown) and fails CLOSED — collect_flow_metadata propagates
   consumer_lag_unknown, and the handler adds EnumFlowStatus.LAG_UNKNOWN so a
   failed probe is never classified healthy.

2. Hardcoded LAN IP (Rule #6): lane_target.py defaulted the .201 runtime
   host to the raw private IP 192.168.86.201 — unroutable off-network, the
   same bug class node_database_sweep already fixed under OMN-14526. Its own
   comment also claimed ONEX_DATA_FLOW_RUNTIME_HOST overrode the default, but
   the code never read that env var. Both are fixed: the default is now the
   Tailscale MagicDNS host, and ONEX_DATA_FLOW_RUNTIME_HOST is actually read
   (mirroring node_database_sweep's env-read so the two sweeps agree on which
   host they probe).

RED->GREEN: added tests asserting rpk-failure fails closed (never FLOWING)
and that the env var override actually takes effect — both failed against
the pre-fix code, pass after. Also satisfies the pre-existing
state-coverage-gate baseline debt for this node (summary output / result
topic), promoted to a hard FAIL by --strict because this PR directly
touches the node.

* test(OMN-14531): cover swarm dispatch FSM states
@jonahgabriel
jonahgabriel deleted the jonah/omn-7624-task-3-post-pipeline-data-validation-gate branch August 25, 2026 18:15
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant