feat(OMN-16769): read-only DLQ depth/arrival monitor with red-run alerting

.github/workflows/dlq-depth-monitor.yml

@@ -0,0 +1,171 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+#
+# dlq-depth-monitor.yml — OMN-16769
+#
+# Scheduled read-only sweep of every onex.dlq.* topic on the dev lane broker.
+# Measures retained depth and arrivals-in-window per topic, evaluates them
+# against the contract-pinned bounds in node_dlq_depth_evaluate_compute, and
+# EXITS NON-ZERO when a sink takes more than its declared bound allows. A red
+# run is the alert surface — there is no separate notification channel, and
+# none was added (OMN-16769 scopes the alert to "a failing scheduled-workflow
+# run (red is visible)").
+#
+# WHY THIS EXISTS: onex.dlq.omnibase-infra.quarantine.v1 reached ~8.88M records
+# with zero projection and zero alerting, and a total delegation outage
+# (OMN-16767) hid inside it silently until someone went looking by hand.
+#
+# READ-ONLY BY CONSTRUCTION, not by convention. The backing node's only broker
+# surface is ProtocolDlqAdminTransport — five methods, all reads — and the live
+# adapter backs it with a GROUP-LESS AIOKafkaConsumer that joins no consumer
+# group and commits nothing. There is no --apply flag here because there is
+# nothing this workflow could apply. That is a stronger guarantee than the
+# DRY-RUN defaults on the sibling sweeps (sync-revert-watchdog.yml,
+# evidence-autoclose-sweep.yml) this workflow is modeled on, which DO have a
+# live mutation path deliberately left unarmed.
+#
+# The kill switch (ONEX_DLQ_MONITOR_DISABLED) is honored by the node itself
+# unconditionally, independent of this workflow's config, so it is not (and
+# must not be) surfaced as an input here — set it as a repo/environment
+# variable to halt every run without touching this file.
+#
+# WINDOW SIZING — window_seconds=1800 deliberately EQUALS the 30-minute cron
+# interval. Unlike the lookback windows on the sibling sweeps (which
+# intentionally exceed their interval so a missed run never drops an event),
+# this measurement is a RATE, not a backlog scan: overlapping windows would
+# double-count arrivals across consecutive runs and inflate the apparent rate.
+# A missed run therefore leaves a genuine gap rather than being re-covered —
+# an acceptable trade for a monitor whose signal is "is this sink taking
+# traffic right now", and the standing depth column still reflects anything
+# that landed during a gap.
+#
+# No new scripts/** file: scripts/** is DEFAULT-DENY for new files (OMN-14475
+# canonical-form guard). The step below is a direct `uv run onex ...`
+# invocation of code that already lives under src/.
+
+name: DLQ Depth Monitor
+
+on:
+  schedule:
+    # Every 30 minutes; window_seconds below matches exactly (see WINDOW SIZING).
+    - cron: '*/30 * * * *'
+  workflow_dispatch:
+    inputs:
+      window_seconds:
+        description: 'Arrival-measurement window in seconds. Keep equal to the cron interval for scheduled runs.'
+        required: false
+        default: '1800'
+      topic_prefix:
+        description: 'Only topics with this prefix are probed.'
+        required: false
+        default: 'onex.dlq.'
+      suppress_alert_exit:
+        description: 'Set true for a characterization run: collect the histogram WITHOUT failing the run on a breach. Leave false so a breach goes red.'
+        required: false
+        default: 'false'
+        type: choice
+        options:
+          - 'false'
+          - 'true'
+
+env:
+  UV_VERSION: "0.8.3"
+  PYTHON_VERSION: "3.12"
+
+# Never let two probes overlap: a delayed run plus the next scheduled tick
+# would measure overlapping windows and report inflated arrival counts.
+concurrency:
+  group: dlq-depth-monitor
+  cancel-in-progress: false
+
+jobs:
+  dlq-depth-monitor:
+    name: DLQ Depth Monitor (read-only)
+    # Trusted runner required: needs LAN access to the dev lane broker
+    # (localhost:19092 on .201, same address baselines-scheduler.yml uses).
+    # OMNI_RUNNER_SELECTOR_V1 — trusted CI defaults to self-hosted omnibase-ci; public forks default to ubuntu-latest
+    runs-on: >-
+      ${{
+        (github.event_name == 'pull_request' &&
+         github.event.pull_request.head.repo.full_name != github.repository)
+        && fromJSON(vars.OMNI_PUBLIC_PR_RUNS_ON_JSON || '["ubuntu-latest"]')
+        || fromJSON(vars.OMNI_TRUSTED_CI_RUNS_ON_JSON || '["self-hosted","omnibase-ci"]')
+      }}
+    timeout-minutes: 15
+    permissions:
+      contents: read
+
+    steps:
+      - name: Checkout omnibase_infra
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          fetch-depth: 1
+
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: ${{ env.PYTHON_VERSION }}
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v7
+        with:
+          version: ${{ env.UV_VERSION }}
+          enable-cache: true
+          cache-dependency-glob: "uv.lock"
+
+      - name: Install dependencies
+        run: uv sync
+
+      - name: Probe DLQ depth (read-only; non-zero exit == alert)
+        env:
+          TOPIC_PREFIX: ${{ github.event.inputs.topic_prefix || 'onex.dlq.' }}
+          WINDOW_SECONDS: ${{ github.event.inputs.window_seconds || '1800' }}
+          SUPPRESS_ALERT_EXIT: ${{ github.event.inputs.suppress_alert_exit || 'false' }}
+          # Trusted runner: KAFKA_BOOTSTRAP_SERVERS comes from ~/.omnibase/.env,
+          # same sourcing pattern as baselines-scheduler.yml. The node applies
+          # NO default broker address — an unset value fails the run closed
+          # rather than silently probing the wrong lane and reporting a false
+          # clean bill of health.
+          KAFKA_BOOTSTRAP_SERVERS: ${{ secrets.KAFKA_BOOTSTRAP_SERVERS || 'localhost:19092' }}
+        run: |
+          set -euo pipefail
+          # On the self-hosted runner the shell env from ~/.omnibase/.env wins.
+          source ~/.omnibase/.env 2>/dev/null || true
+          if [ -z "${KAFKA_BOOTSTRAP_SERVERS:-}" ]; then
+            echo "::error::KAFKA_BOOTSTRAP_SERVERS is not set — refusing to probe an unspecified broker."
+            exit 1
+          fi
+          suppress_flag=()
+          if [ "$SUPPRESS_ALERT_EXIT" = "true" ]; then
+            suppress_flag=(--suppress-alert-exit)
+          fi
+          uv run onex skill dlq_depth_monitor \
+            --topic-prefix "$TOPIC_PREFIX" \
+            --window-seconds "$WINDOW_SECONDS" \
+            "${suppress_flag[@]}" \
+            | tee dlq-monitor-result.json
+
+      - name: Post summary
+        if: always()
+        run: |
+          {
+            echo "## DLQ Depth Monitor (OMN-16769, read-only)"
+            echo ""
+            echo "Probed every \`onex.dlq.*\` topic on the dev lane broker for retained"
+            echo "depth and arrivals-in-window. No writes of any kind were performed:"
+            echo "the node's only broker surface is read-only and its consumer is"
+            echo "group-less, so it commits nothing."
+            echo ""
+            if [ "${{ job.status }}" = "failure" ]; then
+              echo "**RED = at least one DLQ sink exceeded its declared arrival bound.**"
+              echo "The failing step's log names each offending topic with its arrival"
+              echo "count, its bound, and its retained depth."
+            else
+              echo "No DLQ sink exceeded its arrival bound this window."
+            fi
+            echo ""
+            echo "Standing depth is REPORTED but does not gate the run — see the"
+            echo "\`alert_policy\` block in node_dlq_depth_evaluate_compute/contract.yaml"
+            echo "for why a depth bound cannot be the primary signal on a topic"
+            echo "already holding ~8.88M records (OMN-16769 AC4)."
+          } >> "$GITHUB_STEP_SUMMARY"

docker/runners/runner-image.lock.json

@@ -1,12 +1,12 @@
 {
   "base_image_digest": "sha256:3ba65aa20f86a0fad9df2b2c259c613df006b2e6d0bfcc8a146afb8c525a9751",
   "gh_version": "2.67.0",
-  "identity_digest": "c896b6c2d2ea576b7056c2d57f0e4d0f",
+  "identity_digest": "47224bdefe93869e4dd05501c506eb18",
   "image_version": 7,
   "kubectl_version": "1.32.1",
   "python_version": "3.12",
   "runner_version": "2.336.0",
-  "shared_env_digest": "af3a810f5ded3d0f4612a34c",
+  "shared_env_digest": "ac02631ce31562210f83ee2f",
   "shared_env_install_args": "--frozen --all-extras --all-groups --no-install-project",
   "uv_version": "0.6.14"
 }

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "omnibase_infra"
-version = "0.38.11"
+version = "0.38.12"
 description = "ONEX Infrastructure - Service integration and database infrastructure tools"
 authors = [{ name = "OmniNode.ai", email = "contact@omninode.ai" }]
 license = {text = "MIT"}
@@ -597,6 +597,8 @@ node_decision_store_effect = "omnibase_infra.nodes.node_decision_store_effect"
 node_decision_store_query_compute = "omnibase_infra.nodes.node_decision_store_query_compute"
 node_delta_bundle_effect = "omnibase_infra.nodes.node_delta_bundle_effect"
 node_delta_metrics_effect = "omnibase_infra.nodes.node_delta_metrics_effect"
+node_dlq_depth_evaluate_compute = "omnibase_infra.nodes.node_dlq_depth_evaluate_compute"
+node_dlq_depth_monitor_effect = "omnibase_infra.nodes.node_dlq_depth_monitor_effect"
 node_dlq_replay_effect = "omnibase_infra.nodes.node_dlq_replay_effect"
 node_emit_daemon_runtime = "omnibase_infra.nodes.node_emit_daemon_runtime"
 node_event_bus_wiring_effect = "omnibase_infra.nodes.node_event_bus_wiring_effect"

scripts/check-env-reads.sh

@@ -72,6 +72,15 @@ APPROVED_INFIX_PATTERNS=(
     # (ONEX_SYNC_REVERT_WATCHDOG_DISABLED) directly for the same reason.
     # Declares required_secrets per the OMN-14951 gap-2 check below.
     "/nodes/node_sync_revert_watchdog_effect/handlers/handler_sync_revert_watchdog.py"
+    # OMN-16769: the DLQ depth monitor is the same dispatch shape as the two
+    # entries above (RuntimeLocal's single-shot compute path, no
+    # config-prefetch/overlay seam reachable from a node's own handler) — it
+    # reads its broker address (KAFKA_BOOTSTRAP_SERVERS) and kill switch
+    # (ONEX_DLQ_MONITOR_DISABLED) directly for the same reason. It applies NO
+    # default for the broker address: an unset value fails the run closed
+    # rather than silently probing the wrong lane, which would report a false
+    # clean bill of health.
+    "/nodes/node_dlq_depth_monitor_effect/handlers/handler_dlq_depth_monitor.py"
 )
 
 ENV_READ_PATTERNS='os\.environ\[|os\.environ\.get|os\.getenv|from os import environ|from os import getenv'

scripts/ci/infra-node-allowlist.txt

@@ -280,3 +280,7 @@ src/omnibase_infra/nodes/node_evidence_autoclose_sweep_effect/handlers/__init__.
 src/omnibase_infra/nodes/node_evidence_autoclose_sweep_effect/handlers/handler_evidence_autoclose_sweep.py  # OMN-16106: infra-owned governed-Done-flip sweep. Shells out to `uv run onex skill dod_verify` verbatim, which dispatches through the omnimarket co-install into THIS venv (see omnimarket_drift_guard) — the sweep itself must run from the same infra locality, so it is not a portable omnimarket business-domain node.
 src/omnibase_infra/nodes/node_sync_revert_watchdog_effect/handlers/__init__.py  # OMN-16536: infra-owned Linear-sync-revert detect+correct sweep, same class as node_evidence_autoclose_sweep_effect above — a scheduled CI job calling the Linear GraphQL API directly with a trusted-runner-bound LINEAR_API_KEY secret, not a portable omnimarket business-domain node.
 src/omnibase_infra/nodes/node_sync_revert_watchdog_effect/handlers/handler_sync_revert_watchdog.py  # OMN-16536: infra-owned Linear-sync-revert detect+correct sweep, same class as node_evidence_autoclose_sweep_effect above — a scheduled CI job calling the Linear GraphQL API directly with a trusted-runner-bound LINEAR_API_KEY secret, not a portable omnimarket business-domain node.
+src/omnibase_infra/nodes/node_dlq_depth_evaluate_compute/handlers/__init__.py  # OMN-16769: infra-owned DLQ observability. Reads Kafka broker offsets on the lane it is deployed beside via a trusted-runner-bound KAFKA_BOOTSTRAP_SERVERS — platform observability of the infra event bus itself (same class as node_gateway_link_health_projection_compute above), not a portable omnimarket business-domain node.
+src/omnibase_infra/nodes/node_dlq_depth_evaluate_compute/handlers/handler_dlq_depth_evaluate.py  # OMN-16769: infra-owned DLQ observability. Reads Kafka broker offsets on the lane it is deployed beside via a trusted-runner-bound KAFKA_BOOTSTRAP_SERVERS — platform observability of the infra event bus itself (same class as node_gateway_link_health_projection_compute above), not a portable omnimarket business-domain node.
+src/omnibase_infra/nodes/node_dlq_depth_monitor_effect/handlers/__init__.py  # OMN-16769: infra-owned DLQ observability. Reads Kafka broker offsets on the lane it is deployed beside via a trusted-runner-bound KAFKA_BOOTSTRAP_SERVERS — platform observability of the infra event bus itself (same class as node_gateway_link_health_projection_compute above), not a portable omnimarket business-domain node.
+src/omnibase_infra/nodes/node_dlq_depth_monitor_effect/handlers/handler_dlq_depth_monitor.py  # OMN-16769: infra-owned DLQ observability. Reads Kafka broker offsets on the lane it is deployed beside via a trusted-runner-bound KAFKA_BOOTSTRAP_SERVERS — platform observability of the infra event bus itself (same class as node_gateway_link_health_projection_compute above), not a portable omnimarket business-domain node.

scripts/pull-all.sh

@@ -149,13 +149,34 @@ _run_bounded() {
   local bound="$1"
   shift
 
-  local marker cmd_pid watchdog_pid rc waited
+  local marker pgid_marker cmd_pid cmd_pgid watchdog_pid rc waited
   marker=$(mktemp)
+  pgid_marker=""
 
-  set -m
-  "$@" &
-  cmd_pid=$!
-  set +m
+  if command -v setsid >/dev/null 2>&1; then
+    if declare -F "$1" >/dev/null 2>&1; then
+      export -f "$1"
+    fi
+    pgid_marker=$(mktemp)
+    setsid bash -c '
+      pgid_file="$1"
+      shift
+      printf "%s\n" "$$" > "$pgid_file"
+      "$@"
+    ' pull-all-bounded "$pgid_marker" "$@" &
+    cmd_pid=$!
+    for _ in {1..50}; do
+      [[ -s "$pgid_marker" ]] && break
+      sleep 0.1
+    done
+    cmd_pgid=$(cat "$pgid_marker" 2>/dev/null || printf "%s" "$cmd_pid")
+  else
+    set -m
+    "$@" &
+    cmd_pid=$!
+    cmd_pgid="$cmd_pid"
+    set +m
+  fi
 
   (
     waited=0
@@ -165,9 +186,11 @@ _run_bounded() {
       kill -0 "$cmd_pid" 2>/dev/null || exit 0
     done
     echo "timeout" > "$marker"
-    kill -TERM -"$cmd_pid" 2>/dev/null || kill -TERM "$cmd_pid" 2>/dev/null || true
+    _kill_descendants_leaf_first TERM "$cmd_pid"
+    kill -TERM -"$cmd_pgid" 2>/dev/null || kill -TERM "$cmd_pid" 2>/dev/null || true
     sleep 5
-    kill -KILL -"$cmd_pid" 2>/dev/null || kill -KILL "$cmd_pid" 2>/dev/null || true
+    _kill_descendants_leaf_first KILL "$cmd_pid"
+    kill -KILL -"$cmd_pgid" 2>/dev/null || kill -KILL "$cmd_pid" 2>/dev/null || true
   ) &
   watchdog_pid=$!
 
@@ -179,14 +202,29 @@ _run_bounded() {
 
   # Sweep the group unconditionally: a command can exit while leaving a
   # backgrounded descendant behind, which is the same orphan class.
-  kill -KILL -"$cmd_pid" 2>/dev/null || true
+  _kill_descendants_leaf_first KILL "$cmd_pid"
+  kill -KILL -"$cmd_pgid" 2>/dev/null || true
 
   if [[ -s "$marker" ]]; then
     rc=124
   fi
   rm -f "$marker"
+  [[ -n "$pgid_marker" ]] && rm -f "$pgid_marker"
   return "$rc"
 }
+
+_kill_descendants_leaf_first() {
+  local signal="$1"
+  local parent="$2"
+  local child
+
+  command -v pgrep >/dev/null 2>&1 || return 0
+  while IFS= read -r child; do
+    [[ -n "$child" ]] || continue
+    _kill_descendants_leaf_first "$signal" "$child"
+    kill "-$signal" "$child" 2>/dev/null || true
+  done < <(pgrep -P "$parent" 2>/dev/null || true)
+}
 # === End stage tracking ===
 
 # === Pre-pull validation: detect bare repo corruption (OMN-7600) ===
@@ -577,7 +615,7 @@ fi
 # is stable regardless of which directory it is computed against.
 #
 # The hash is computed against RELATIVE paths so a repo-side and cache-side
-# computation of the same plugin tree yield the same hash (shasum emits
+# computation of the same plugin tree yield the same hash (the hasher emits
 # `hash  path` — absolute paths would otherwise break comparability).
 #
 # `-exec ... +` NOT `-exec ... \;` (OMN-15590). The per-file form spawned ONE
@@ -590,12 +628,21 @@ fi
 # the same tree in 7.9s.
 _plugin_content_hash() {
   local root="$1"
+  local hasher=""
+  if command -v shasum >/dev/null 2>&1; then
+    hasher="shasum"
+  elif command -v sha1sum >/dev/null 2>&1; then
+    hasher="sha1sum"
+  else
+    echo "ERROR: neither shasum nor sha1sum is available for plugin content hashing" >&2
+    return 127
+  fi
   ( cd "${root}" && find . -type f \
       ! -name "*.pyc" \
       ! -path "*/__pycache__/*" \
       ! -name ".deployed-commit" \
       ! -name ".content-hash" \
-      -exec shasum {} + 2>/dev/null | sort | shasum | cut -d' ' -f1 )
+      -exec "$hasher" {} + 2>/dev/null | sort | "$hasher" | cut -d' ' -f1 )
 }
 
 STAGE_PLUGIN_CACHE="SKIPPED"  # no cache and/or no omniclaude clone -- nothing to refresh

src/omnibase_infra/cli/skill_mapping.yaml

@@ -212,6 +212,28 @@ skills:
       - {name: human-comment-window-seconds, payload_field: human_comment_window_seconds, arg_type: integer}
       - {name: linear-timeout-seconds, payload_field: linear_timeout_seconds, arg_type: integer}
       - {name: watchdog-comment-marker, payload_field: watchdog_comment_marker, arg_type: string}
+  # OMN-16769: read-only DLQ depth/arrival-rate monitor. Node-backed in
+  # omnibase_infra itself (same locality precedent as
+  # node_evidence_autoclose_sweep_effect / node_sync_revert_watchdog_effect).
+  # There is NO --apply-style flag here because there is nothing to apply: the
+  # node's only broker surface is read-only by construction. With no flags an
+  # alerting run exits non-zero and the scheduled workflow goes RED — that red
+  # run IS the alert surface. Pass --suppress-alert-exit for a characterization
+  # run that wants the histogram without gating. The flag is phrased negatively
+  # because `onex skill` booleans are PRESENCE-ONLY (cli_skill.py sets True on
+  # sight and rejects unknown `--no-*` tokens), so the safe value must be the
+  # default that needs no flag. Thresholds are contract-pinned; the CLI exposes them only
+  # so a one-off run can widen them, never so a schedule can silently weaken
+  # them (the workflow passes none of them).
+  - skill_name: dlq_depth_monitor
+    node_name: node_dlq_depth_monitor_effect
+    result_model: omnibase_infra.nodes.node_dlq_depth_monitor_effect.models.model_dlq_depth_monitor_result.ModelDlqDepthMonitorResult
+    args:
+      - {name: topic-prefix, payload_field: topic_prefix, arg_type: string}
+      - {name: window-seconds, payload_field: window_seconds, arg_type: integer}
+      - {name: default-max-arrivals-per-window, payload_field: default_max_arrivals_per_window, arg_type: integer}
+      - {name: max-retained-depth, payload_field: max_retained_depth, arg_type: integer}
+      - {name: suppress-alert-exit, payload_field: suppress_alert_exit, arg_type: boolean, default: false}
   - skill_name: duplication_sweep
     node_name: node_duplication_sweep
     result_model: omnimarket.nodes.node_duplication_sweep.handlers.handler_duplication_sweep.DuplicationSweepResult

src/omnibase_infra/nodes/node_dlq_depth_evaluate_compute/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT