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

src/omnimarket/nodes/node_data_flow_sweep/__main__.py

@@ -79,7 +79,8 @@ def main() -> None:
         help=(
             "JSON array of pre-collected flow objects (keys: topic, handler_name, "
             "table_name, dashboard_route, producer_status, consumer_lag, "
-            "table_row_count, table_has_recent_data, field_mapping_valid). "
+            "consumer_lag_unknown, table_row_count, table_has_recent_data, "
+            "field_mapping_valid). "
             "Default when --collect is not set: built-in critical chains."
         ),
     )

src/omnimarket/nodes/node_data_flow_sweep/collector.py

@@ -1,7 +1,5 @@
 # SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
 # SPDX-License-Identifier: MIT
-# onex-allow-file-internal-ip OMN-13552 reason="docstring documents the .201 runtime-lane host this collector probes over SSH; lane host is resolved by lane_target, not hardcoded here as a connection string"
-# onex-allow-file OMN-13552 reason="docstring documents the .201 runtime-lane host this collector probes over SSH; lane host is resolved by lane_target, not hardcoded here as a connection string"
 """LiveMetadataCollector — shell-out collection phase for node_data_flow_sweep.
 
 Runs rpk topic/group describe and psql row-count checks to populate
@@ -10,7 +8,8 @@
 
 OMN-13552: probes target a resolved *lane* (``ModelLaneTarget``) rather than
 "whatever host this process runs on". A lane resolving to a remote runtime host
-(the dev/stability/prod/judge lanes on ``192.168.86.201``) is probed via
+(the dev/stability/prod/judge lanes on the .201 runtime host — see
+``lane_target.py`` for the concrete host resolution) is probed via
 ``ssh <user>@<host> docker exec <container> ...`` — the same SSH + ``docker
 exec`` transport already proven in ``node_integration_sweep_orchestrator``
 (OMN-7238 class fix). A lane with an empty ``runtime_host`` (``local``) is probed
@@ -255,16 +254,30 @@ def _probe_newest_message_age(target: ModelLaneTarget, topic: str) -> float | No
     return None
 
 
-def probe_consumer_lag(target: ModelLaneTarget, topic: str) -> int:
-    """Return consumer lag for the lane's default consumer group on the topic."""
+def probe_consumer_lag(target: ModelLaneTarget, topic: str) -> tuple[int, bool]:
+    """Return ``(consumer_lag, lag_unknown)`` for the lane's consumer group.
+
+    OMN-14531: a failed ``rpk group describe`` previously defaulted to
+    ``lag=0`` — indistinguishable from a genuinely quiescent group, so the
+    handler classified the flow healthy even though lag was never actually
+    observed. This now fails CLOSED: ``lag_unknown=True`` signals the probe
+    could not be trusted, so the caller must never treat it as healthy.
+    """
     code, out = _docker_exec(
         target,
         target.redpanda_container,
         ["rpk", "group", "describe", target.consumer_group],
         timeout=20,
     )
     if code != 0:
-        return 0  # Unknown — default 0 (handler won't flag LAGGING)
+        _log.warning(
+            "rpk group describe failed for group %s on lane %s (rc=%d) — "
+            "failing closed (lag unknown, never treated as healthy)",
+            target.consumer_group,
+            target.lane,
+            code,
+        )
+        return 0, True
 
     total_lag = 0
     for line in out.splitlines():
@@ -273,7 +286,7 @@ def probe_consumer_lag(target: ModelLaneTarget, topic: str) -> int:
             if len(parts) >= 5:
                 with contextlib.suppress(ValueError):
                     total_lag += int(parts[-1])
-    return total_lag
+    return total_lag, False
 
 
 def probe_table_row_count(target: ModelLaneTarget, table_name: str) -> tuple[int, bool]:
@@ -351,7 +364,11 @@ def collect_flow_metadata(
     MISSING so the handler classifies it as PRODUCER_DOWN rather than crashing.
     Reachability itself is asserted up-front by the caller via
     :func:`assert_lane_reachable`, so MISSING here means "topic genuinely absent
-    on a lane we could reach", not "wrong host".
+    on a lane we could reach", not "wrong host". A consumer-lag probe failure is
+    the one exception (OMN-14531): it fails CLOSED via
+    ``consumer_lag_unknown=True`` rather than being folded into MISSING, so the
+    handler can distinguish "producer gone" from "lag probe untrustworthy" while
+    still never classifying either as a healthy flow.
     """
     try:
         producer_status, newest_age = probe_producer_status(
@@ -366,17 +383,21 @@ def collect_flow_metadata(
         newest_age = None
 
     consumer_lag = 0
+    consumer_lag_unknown = False
     row_count = 0
     has_recent = False
 
     if producer_status == EnumProducerStatus.ACTIVE:
         try:
-            consumer_lag = probe_consumer_lag(target, flow_descriptor.topic)
+            consumer_lag, consumer_lag_unknown = probe_consumer_lag(
+                target, flow_descriptor.topic
+            )
         except Exception:
             _log.debug(
-                "probe_consumer_lag raised for %s — defaulting to 0",
+                "probe_consumer_lag raised for %s — failing closed (lag unknown)",
                 flow_descriptor.topic,
             )
+            consumer_lag_unknown = True
 
         try:
             row_count, has_recent = probe_table_row_count(
@@ -395,6 +416,7 @@ def collect_flow_metadata(
         dashboard_route=flow_descriptor.dashboard_route,
         producer_status=producer_status,
         consumer_lag=consumer_lag,
+        consumer_lag_unknown=consumer_lag_unknown,
         table_row_count=row_count,
         table_has_recent_data=has_recent,
         field_mapping_valid=flow_descriptor.field_mapping_valid,

src/omnimarket/nodes/node_data_flow_sweep/handlers/handler_data_flow_sweep.py

@@ -51,6 +51,9 @@ class EnumFlowStatus(StrEnum):
     MISSING_TABLE = "MISSING_TABLE"
     PRODUCER_DOWN = "PRODUCER_DOWN"
     TOPIC_STALE = "TOPIC_STALE"
+    LAG_UNKNOWN = (
+        "LAG_UNKNOWN"  # OMN-14531: lag probe failed — fail CLOSED, never FLOWING
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -69,6 +72,7 @@ class ModelFlowInput(BaseModel):
     dashboard_route: str | None = None
     producer_status: EnumProducerStatus = EnumProducerStatus.ACTIVE
     consumer_lag: int = 0
+    consumer_lag_unknown: bool = False  # OMN-14531: True when the live lag probe failed — fail CLOSED, never healthy
     table_row_count: int = 0
     table_has_recent_data: bool = False
     field_mapping_valid: bool = True
@@ -334,6 +338,23 @@ def _verify_flow(self, flow: ModelFlowInput) -> ModelFlowResult:
                 message=f"Messages in topic but 0 rows in {flow.table_name}",
             )
 
+        # OMN-14531: a failed lag probe must fail CLOSED, never fall through to
+        # FLOWING. consumer_lag_unknown is set by the collector when rpk could
+        # not describe the consumer group — the flow's actual lag was never
+        # observed, so it is neither known-healthy nor known-lagging.
+        if flow.consumer_lag_unknown:
+            return ModelFlowResult(
+                topic=flow.topic,
+                handler_name=flow.handler_name,
+                table_name=flow.table_name,
+                producer_status=flow.producer_status,
+                flow_status=EnumFlowStatus.LAG_UNKNOWN,
+                consumer_lag=flow.consumer_lag,
+                table_row_count=flow.table_row_count,
+                field_mapping_valid=flow.field_mapping_valid,
+                message=f"Consumer lag probe failed for {flow.topic} — status unknown",
+            )
+
         if flow.consumer_lag > 0:
             return ModelFlowResult(
                 topic=flow.topic,

src/omnimarket/nodes/node_data_flow_sweep/lane_target.py

@@ -1,16 +1,24 @@
 # SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
 # SPDX-License-Identifier: MIT
-# onex-allow-file-internal-ip OMN-13552 reason="lane registry resolves the canonical .201 runtime-lane host; overridable via --runtime-host; not a shipping connection string"
-# onex-allow-file OMN-13552 reason="lane registry resolves the canonical .201 runtime-lane host; overridable via --runtime-host; not a shipping connection string"
 """Lane/target resolution for node_data_flow_sweep collection probes.
 
 OMN-13552: the collector previously probed *whatever host the CLI process is on*
 (``docker exec omnibase-infra-redpanda ...`` / ``psql -h localhost ...``) rather
 than the lane it was nominally verifying. For the dev/stability/prod lanes that
-run on ``192.168.86.201`` this produced false-broken (no local container =>
+run on the .201 runtime host this produced false-broken (no local container =>
 "Topic does not exist" => PRODUCER_DOWN) or false-clean (a stale local stack)
 verdicts for a lane that was never actually probed.
 
+OMN-14531: the default runtime host was a hardcoded raw private LAN IP that is
+unroutable off-network — the exact bug class node_database_sweep already hit
+and fixed under OMN-14526 by defaulting to the Tailscale MagicDNS host
+instead. This module's own docstring/comment also claimed
+``ONEX_DATA_FLOW_RUNTIME_HOST`` overrode the default, but the code never read
+that env var — the override was inert. Both are fixed here:
+:func:`_resolve_default_runtime_host` reads ``ONEX_DATA_FLOW_RUNTIME_HOST``
+(the same var node_database_sweep reads so the two sweeps agree on which host
+they probe) and falls back to the Tailscale hostname, never the raw IP.
+
 This module resolves a lane name to the concrete broker + Postgres endpoints for
 that lane (container names + the host they live on), mirroring the runtime
 address-registry / lane-manifest source of truth (OMN-10345) and the SSH +
@@ -28,6 +36,7 @@
 
 from __future__ import annotations
 
+import os
 from typing import Final
 
 from pydantic import BaseModel, ConfigDict, Field
@@ -46,11 +55,26 @@
 # lane's containers on that remote host.
 # ---------------------------------------------------------------------------
 
-# onex-allow-internal-ip OMN-13552 reason="default runtime host for the .201 lanes; overridden by --runtime-host / ONEX_DATA_FLOW_RUNTIME_HOST; not a shipping connection string"
-_DEFAULT_RUNTIME_HOST: Final[str] = "192.168.86.201"
+# Tailscale MagicDNS name for the .201 runtime host — resolves both on-LAN and
+# off-network, unlike a raw private IP (OMN-14531, mirrors OMN-14526).
+_DEFAULT_RUNTIME_HOST: Final[str] = "omninode-pc.tail75df5e.ts.net"
 _SSH_USER: Final[str] = "jonah"
 
 
+def _resolve_default_runtime_host() -> str:
+    """Resolve the default lane runtime host, honoring the env override.
+
+    ``ONEX_DATA_FLOW_RUNTIME_HOST`` overrides :data:`_DEFAULT_RUNTIME_HOST`
+    when set (mirrors node_database_sweep's ``_resolve_pg_runtime_host``, which
+    reads this same var, OMN-14526). A caller-supplied ``runtime_host=`` kwarg
+    to :func:`resolve_lane_target` still takes precedence over both.
+    """
+    return (
+        os.environ.get("ONEX_DATA_FLOW_RUNTIME_HOST", "").strip()
+        or _DEFAULT_RUNTIME_HOST
+    )
+
+
 class ModelLaneTarget(BaseModel):
     """Resolved probe endpoints for one runtime lane.
 
@@ -144,8 +168,9 @@ def resolve_lane_target(
 
     ``lane="local"`` keeps the legacy in-stack behavior (local docker/psql). Any
     of the .201 lanes (``dev`` / ``stability-test`` / ``prod`` / ``judge``)
-    resolve to that lane's container names on the runtime host (default
-    ``192.168.86.201``, overridable). An unknown lane raises
+    resolve to that lane's container names on the runtime host (default the
+    Tailscale MagicDNS host, overridable via ``runtime_host=`` or
+    ``ONEX_DATA_FLOW_RUNTIME_HOST``). An unknown lane raises
     :class:`LaneResolutionError` so the caller can fail loud instead of probing
     the wrong host.
     """
@@ -174,7 +199,7 @@ def resolve_lane_target(
         )
     redpanda_container, postgres_container = containers
 
-    host = (runtime_host or _DEFAULT_RUNTIME_HOST).strip()
+    host = (runtime_host or _resolve_default_runtime_host()).strip()
     return ModelLaneTarget(
         lane=normalized,
         runtime_host=host,

tests/test_data_flow_sweep_dispatch_resolution.py

@@ -152,3 +152,30 @@ def test_request_model_accepts_collect_and_flows(self) -> None:
         fields = set(DataFlowSweepRequest.model_fields.keys())
         # The skill_mapping surfaces exactly these payload fields.
         assert {"flows", "collect", "dry_run"} <= fields
+
+    def test_contract_declares_summary_output(self) -> None:
+        contract_path = (
+            Path(__file__).resolve().parents[1]
+            / "src"
+            / "omnimarket"
+            / "nodes"
+            / "node_data_flow_sweep"
+            / "contract.yaml"
+        )
+        contract = yaml.safe_load(contract_path.read_text(encoding="utf-8"))
+        assert "summary" in contract["outputs"]
+
+    def test_contract_publishes_result_topic(self) -> None:
+        contract_path = (
+            Path(__file__).resolve().parents[1]
+            / "src"
+            / "omnimarket"
+            / "nodes"
+            / "node_data_flow_sweep"
+            / "contract.yaml"
+        )
+        contract = yaml.safe_load(contract_path.read_text(encoding="utf-8"))
+        assert (
+            "onex.evt.omnimarket.data-flow-sweep-result.v1"
+            in contract["event_bus"]["publish_topics"]
+        )