[jonahgabriel]

Summary

Fixes two structural-blindness defects in node_data_flow_sweep, the node backing onex:data_flow_sweep, found during the OMN-14531 detector-shelf audit (16/16 sweeps found structurally blind).

1. Fail-open on rpk failure. probe_consumer_lag() treated any non-zero rpk group describe return code as lag=0 — indistinguishable from a genuinely quiescent consumer group, so a broker/group probe failure could let an otherwise-broken flow classify FLOWING (healthy). 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 feat: PostgresDataSource adapter for data verification [OMN-7624] #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 hit and fixed under OMN-14526. The module's own comment also claimed ONEX_DATA_FLOW_RUNTIME_HOST overrode the default, but the code never actually read that env var — the override was inert. Both are fixed: the default is now the Tailscale MagicDNS host (omninode-pc.tail75df5e.ts.net), and ONEX_DATA_FLOW_RUNTIME_HOST is now actually read (mirroring node_database_sweep's _resolve_pg_runtime_host, which already reads this exact shared var, so the two sweeps agree on which host they probe).

DoD evidence (OMN-14531)

RED→GREEN, captured against the pre-fix code before implementing:

• test_probe_consumer_lag_fails_closed_on_rpk_failure / test_collect_flow_metadata_lag_probe_failure_fails_closed / test_handler_lag_unknown_flow_is_not_healthy — all failed against unpatched code (AttributeError/ValidationError on consumer_lag_unknown, then a FLOWING false-green once the field existed), pass after the fix.
• test_dev_lane_resolves_to_remote_host / test_remote_lane_probe_uses_ssh_not_local_docker — updated to assert the Tailscale host; failed against the raw-IP default, pass after.
• test_runtime_host_env_var_overrides_default — new test proving ONEX_DATA_FLOW_RUNTIME_HOST actually takes effect; failed against unpatched code (env var was inert), passes after.

Local verification:

• uv run pytest tests/test_golden_chain_data_flow_sweep_collect.py tests/test_golden_chain_data_flow_sweep.py tests/test_data_flow_sweep_dispatch_resolution.py -v -m "not kafka" — 52/52 passed.
• uv run pytest tests/ -q -m "not kafka and not e2e and not live_model" (full suite) — 13259 passed, 19 skipped, 5 pre-existing failures unrelated to this change (test_no_cross_node_reach_in.py::test_no_new_cross_node_model_reach_ins — known OMN-14534-class cross-node reach-in debt; 3× test_pr_graphql_live.py — live GitHub GraphQL network tests). None touch node_data_flow_sweep.
• uv run ruff format / ruff check --fix / mypy --strict src/omnimarket/nodes/node_data_flow_sweep/ — clean.
• pre-commit run --all-files on the changed set — all hooks passed, including state-coverage-gate (fixed a pre-existing baselined-debt promotion to hard-FAIL for this node by adding test_contract_declares_summary_output / test_contract_publishes_result_topic, mirroring the existing test_contract_declares_collect_input convention).

Test plan

• gh pr checks green
• Focused node tests pass (52/52)
• Full local suite run with no new failures introduced
• mypy --strict clean on touched module
• pre-commit run --all-files clean

OCC companion: owed to Codex per standing policy (agents do not author their own OCC companion/receipt).

Closes OMN-14531 (data_flow_sweep sub-fix; parent detector-shelf ticket remains open for the other 9 sweeps).

OCC Evidence

Evidence-Source: OCC#4108
Evidence-Ticket: OMN-14531
Evidence-Commit: 8c709084d03c5f2875fff8eedb2124052a03f3cb

[coderabbitai]

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: 38 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: aba1c022-fef4-4d5b-9631-e9ab032de255

📥 Commits

Reviewing files that changed from the base of the PR and between b8f12f1 and bb9b0a9.

📒 Files selected for processing (7)

• src/omnimarket/nodes/node_data_flow_sweep/__main__.py
• src/omnimarket/nodes/node_data_flow_sweep/collector.py
• src/omnimarket/nodes/node_data_flow_sweep/handlers/handler_data_flow_sweep.py
• src/omnimarket/nodes/node_data_flow_sweep/lane_target.py
• tests/test_data_flow_sweep_dispatch_resolution.py
• tests/test_golden_chain_data_flow_sweep_collect.py
• tests/test_swarm_schema_alignment.py

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch jonah/omn-14531-dataflow-sweep-failclosed

---

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

[github-actions]

✅ Architectural Review — PASSED

Errors: 0
Warnings: 0

What this checks

Rule	Description
ARCH-TOPIC-001	No hardcoded Kafka topic strings in handler code
ARCH-DI-001	No event_bus=None bypass
ARCH-DI-002	No direct Handler instantiation outside workflow_runner/adapters
ARCH-DI-003	No reinvented DI containers
ARCH-TOPIC-002	contract.yaml topics follow `onex.{cmd

---

No architectural violations found.

Static architectural lint — no model inference (OMN-14176).

[github-actions]

✅ Hostile Reviewer — PASSED

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

---

Gate semantics (pilot phase)

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

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