stack/packages/mosaic/framework/tools/wake/detector.sh

#!/usr/bin/env bash
# detector.sh — A1 of the wake canon (EPIC #892, W4): the per-host,
# single-instance, delta-gated DETECTOR daemon.
#
# CONTRACT ANCHORS (docs/scratchpads/heartbeat-planning/CONVERGED-DESIGN.md):
#   §1.1  Placement — ONE detector instance PER HOST, serving that host's LOCAL
#         panes only (no cross-host waking). A long-lived SERVICE, NOT a timer:
#         retiring unconditional timers is the whole point. Generalizes the
#         proven delta-gated poller (gitea-pr-watch: poll -> hash -> deliver only
#         on delta; 0-wasted) and adds (a) repo-section/anchor-scoped hashing so
#         human-decision FILE edits are caught, not just API-visible state.
#   §1.2  observe -> enqueue into the W2 durable store (the store ALLOCATES the
#         observed_seq — single store-side allocator, #908).
#   §2.4  Cursor semantics: `observed_seq` is a monotonic int, AUTHORITATIVE, now
#         allocated SOLELY by store.sh enqueue (#908 unified the two former
#         allocators). The detector no longer keeps a private counter: it enqueues
#         and captures the store-assigned seq. Source SHAs are DESCRIPTORS, not the
#         cursor. A per-watch LAST-OBSERVED HASH is compared each poll so a revert
#         (A->B->A across polls) is caught as a delta.
#   §4/G2a FAIL-LOUD source semantics: a source/target failure OR a 401 / 403 /
#         privacy-404 / partial / ambiguous-empty response MUST fail loud, MUST
#         NOT advance the authoritative cursor (`observed_seq`), and MUST NOT be
#         silently treated as "no change". An error, or an empty response that
#         might mean "hidden", is NEVER "no change".
#   Gate B (#869 discipline): the watch-list `schema_version` is validated
#         against the component manifest's supported [schema_min, schema_max]
#         range; an out-of-range watch-list is REJECTED (fail loud), never
#         silently coerced.
#
# SCOPE: this tool OBSERVES sources and ENQUEUES deltas to the W2 store
# (store.sh, merged). It does NOT reimplement the store, ack, digest, or signer
# (W2/W3 — it only CALLS their public APIs). FN-oracle/reconciler (W5), off-host
# beacon (W6) and installer (W7) are explicitly OUT OF SCOPE.
#
# SOURCE ADAPTER CONTRACT (operator-agnostic firewall: the detector bakes in NO
# git/gitea/HTTP/file specifics — the operator supplies the adapter):
#   WAKE_DETECTOR_SOURCE_CMD is invoked as:  <cmd> <kind> <id>
#   with the source definition JSON on STDIN. It reports the CURRENT observed
#   state of that source:
#     exit 0 + NON-EMPTY stdout  = the raw source descriptor (bytes to hash).
#     exit 0 + EMPTY stdout      = AMBIGUOUS-EMPTY -> FAIL LOUD (an empty that
#                                  might mean "hidden/403'd" is never "no change").
#     exit non-zero              = source error (network failure, 401/403,
#                                  privacy-404, partial/truncated) -> FAIL LOUD.
#   The detector applies anchor-scoping (feature (a)) to the raw descriptor when
#   the source defines an `anchor`, then hashes the scoped region.
#
# Operator-agnostic: all state via XDG/env; no operator paths/names/secrets/hosts.
set -uo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=./_wake-common.sh disable=SC1091
. "$SCRIPT_DIR/_wake-common.sh"

STATE_DIR="$(wake_state_dir)"
STORE_SH="$SCRIPT_DIR/store.sh"
MANIFEST="$SCRIPT_DIR/manifest.txt"

# Detector-local state lives UNDER the store's STATE_DIR in its own subdir so it
# never collides with the store's cursor/inbox files.
DET_DIR="$STATE_DIR/detector"

_need_jq() {
  command -v jq >/dev/null 2>&1 || {
    echo "detector.sh: jq is required" >&2
    exit 3
  }
}

# _hash — sha256 of stdin, first field only (portable across sha256sum/shasum/
# openssl). Used for the per-watch last-observed hash (delta gate).
_hash() {
  if command -v sha256sum >/dev/null 2>&1; then
    sha256sum | awk '{print $1}'
  elif command -v shasum >/dev/null 2>&1; then
    shasum -a 256 | awk '{print $1}'
  elif command -v openssl >/dev/null 2>&1; then
    openssl dgst -sha256 | awk '{print $NF}'
  else
    echo "detector.sh: no sha256 tool (sha256sum/shasum/openssl) available" >&2
    exit 3
  fi
}

usage() {
  cat >&2 <<'EOF'
Usage: detector.sh <command> [options]

Commands:
  poll-once                              Run EXACTLY ONE delta-gated polling pass
                                         over every source in the watch-list.
                                         Enqueues to the W2 store on a delta only.
                                         Exits non-zero if ANY source failed loud
                                         (G2a) — the failing source does NOT
                                         advance observed_seq and is NOT treated
                                         as "no change".
  run [--once]                           Long-lived per-host SINGLE-INSTANCE
                                         service: acquire a non-blocking flock
                                         (a 2nd instance REFUSES), then loop
                                         poll-once every WAKE_DETECTOR_INTERVAL
                                         seconds. --once does a single guarded
                                         cycle then exits (still flock-guarded).
  validate                               Load + validate the watch-list and its
                                         schema_version against the manifest;
                                         exit 0 if usable, non-zero (loud) if not.
  cursors                                Print the store cursors (observed_seq is
                                         the store's single source of truth, #908;
                                         the detector keeps no private counter).

Environment:
  WAKE_WATCH_LIST          Path to the operator watch-list JSON (required).
  WAKE_DETECTOR_SOURCE_CMD Source adapter command (see header contract; required
                           for poll-once/run).
  WAKE_DETECTOR_INTERVAL   run-loop poll interval seconds (default 30).
  WAKE_DETECTOR_LOCK       flock path (default <state>/detector/detector.lock).
  WAKE_STATE_HOME/WAKE_AGENT  store namespace (see store.sh).
EOF
}

# --- watch-list load + Gate B schema-version validation --------------------

_manifest_val() {
  # _manifest_val KEY — echo VALUE for KEY=VALUE in the manifest (blank if none).
  local key="$1"
  [ -f "$MANIFEST" ] || return 0
  sed -n "s/^${key}=//p" "$MANIFEST" | head -n1 | tr -d '[:space:]'
}

# _load_watchlist — validate the watch-list path + JSON + schema_version range.
# FAIL LOUD (non-zero) on anything unusable. Echoes the validated JSON on stdout.
_load_watchlist() {
  _need_jq
  local wl="${WAKE_WATCH_LIST:-}"
  if [ -z "$wl" ]; then
    echo "detector.sh: WAKE_WATCH_LIST is not set (no watch-list to observe)" >&2
    return 2
  fi
  if [ ! -f "$wl" ]; then
    echo "detector.sh: watch-list not found: $wl" >&2
    return 2
  fi
  local json
  if ! json="$(jq -e . "$wl" 2>/dev/null)"; then
    echo "detector.sh: watch-list is not valid JSON: $wl" >&2
    return 2
  fi

  # Required shape.
  if ! printf '%s' "$json" | jq -e 'has("schema_version") and has("watches")' >/dev/null 2>&1; then
    echo "detector.sh: watch-list missing required 'schema_version' or 'watches'" >&2
    return 2
  fi

  # Gate B: schema_version MUST be an integer inside the manifest's supported
  # [schema_min, schema_max] range. Out of range => REJECT (never coerce).
  local ver smin smax
  ver="$(printf '%s' "$json" | jq -r '.schema_version')"
  smin="$(_manifest_val schema_min)"
  smax="$(_manifest_val schema_max)"
  case "$ver" in
  '' | *[!0-9]*)
    echo "detector.sh: watch-list schema_version must be an integer (got '$ver')" >&2
    return 2
    ;;
  esac
  if [ -z "$smin" ] || [ -z "$smax" ]; then
    echo "detector.sh: manifest is missing schema_min/schema_max (cannot validate Gate B)" >&2
    return 2
  fi
  if [ "$ver" -lt "$smin" ] || [ "$ver" -gt "$smax" ]; then
    echo "detector.sh: FAIL LOUD (Gate B) — watch-list schema_version $ver is OUTSIDE the supported range [$smin, $smax]; refusing to run against an incompatible watch-list" >&2
    return 2
  fi

  printf '%s' "$json"
}

# --- observed_seq: the store is the SOLE allocator (#908) -------------------
#
# The detector NO LONGER keeps a private observed_seq counter. That private
# counter was the shared root of three defects (burn-before-enqueue, W5
# dual-allocator aliasing, and the migration-restart silent-swallow); see the PR
# for #908. observed_seq is now allocated EXCLUSIVELY by store.sh enqueue (which
# reads its own cursor and returns the assigned seq). The detector calls enqueue
# WITHOUT --seq and captures the store-returned seq for logging/locators — it
# never allocates a seq itself, so the seam is gone.

# --- per-watch last-observed hash (§2.4 revert/ABA-at-rest detection) --------

_hash_file() {
  # _hash_file KIND ID — deterministic per-source hash-state filename.
  local kind="$1" id="$2" key
  key="$(printf '%s\037%s' "$kind" "$id" | _hash)"
  printf '%s/watch-%s.hash' "$DET_DIR" "$key"
}

# --- anchor-scoped hashing (§1.1 feature (a)) -------------------------------

# _scope_anchor ANCHOR  (raw content on stdin) — extract ONLY the region of the
# content delimited by ANCHOR: from the first line CONTAINING the anchor marker
# up to (but excluding) the next markdown-style heading line, or EOF. This lets
# a file edit OUTSIDE the lane's anchored section not wake the lane, while an
# edit INSIDE it is caught — human-decision FILE edits, not just API state.
# An empty scope (anchor not present) is surfaced to the caller as empty output.
_scope_anchor() {
  local anchor="$1"
  awk -v a="$anchor" '
    index($0, a) > 0 && !inzone { inzone=1; print; next }
    inzone && /^#/ { exit }
    inzone { print }
  '
}

# --- one poll of one source -------------------------------------------------

# _poll_source KIND ID DEF_JSON CLASS — returns:
#   0  processed (delta enqueued OR no-op no-change OR silent first-seen baseline)
#   1  FAIL LOUD (source error / ambiguous-empty) — cursor + hash left UNTOUCHED
_poll_source() {
  local kind="$1" id="$2" def="$3" class="$4"
  local anchor
  anchor="$(printf '%s' "$def" | jq -r '.anchor // empty')"

  # --- observe via the operator source adapter (fail-loud contract) ---------
  # The source definition is handed to the adapter on STDIN via a temp FILE
  # (not a pipe): an adapter that ignores stdin must not take SIGPIPE and read
  # back as a spurious "source error" under `pipefail`.
  local raw rc deftmp metatmp
  mkdir -p "$DET_DIR"
  # _wake_tmp_prefix is provided by _wake-common.sh (sourced above).
  # shellcheck disable=SC2154
  deftmp="$(mktemp "$DET_DIR/${_wake_tmp_prefix}defXXXXXX")" || return 1
  printf '%s' "$def" >"$deftmp"
  # fd 3 is the OUT-OF-BAND snapshot-metadata channel (#940): the adapter MAY
  # write one JSON object {"snapshot_sha": "<git commit sha>", "snapshot_ts":
  # <epoch>} there. It must stay out of stdout because EVERYTHING on stdout is
  # hashed by the delta gate — an in-band tip-commit SHA would advance
  # observed_hash on every unrelated push. Adapters that never write fd 3 leave
  # the file empty: byte-identical legacy behavior.
  metatmp="$(mktemp "$DET_DIR/${_wake_tmp_prefix}metaXXXXXX")" || {
    rm -f "$deftmp"
    return 1
  }
  raw="$("$WAKE_DETECTOR_SOURCE_CMD" "$kind" "$id" <"$deftmp" 3>"$metatmp" 2>/dev/null)"
  rc=$?
  rm -f "$deftmp"
  # Slurp + remove the metadata file NOW so every return path below is clean;
  # it is parsed only after the fail-loud gates (metadata from a FAILED
  # observation is meaningless and must not be trusted or diagnosed).
  local rawmeta=""
  [ -s "$metatmp" ] && rawmeta="$(cat "$metatmp")"
  rm -f "$metatmp"
  if [ "$rc" -ne 0 ]; then
    echo "detector.sh: FAIL LOUD (G2a) — source '$kind/$id' errored (adapter exit $rc: network/401/403/privacy-404/partial). observed_seq NOT advanced; NOT treated as 'no change'." >&2
    return 1
  fi
  if [ -z "$raw" ]; then
    echo "detector.sh: FAIL LOUD (G2a) — source '$kind/$id' returned AMBIGUOUS-EMPTY (empty-that-might-mean-hidden is never 'no change'). observed_seq NOT advanced." >&2
    return 1
  fi

  # --- snapshot metadata (fd 3, #940) — ADVISORY, validated, never gating ----
  # Malformed metadata is dropped with a LOUD diagnostic but cannot fail the
  # poll or suppress the wake: the obligation never depends on optional dating.
  local snap_sha="" snap_ts="" snap_json=""
  if [ -n "$rawmeta" ]; then
    if snap_json="$(jq -ce '.' <<<"$rawmeta" 2>/dev/null)"; then
      snap_sha="$(jq -r 'if (.snapshot_sha|type) == "string" then .snapshot_sha else "" end' <<<"$snap_json")"
      snap_ts="$(jq -r 'if (.snapshot_ts|type) == "number" then (.snapshot_ts|floor|tostring) else "" end' <<<"$snap_json")"
      if [ -n "$snap_sha" ] && ! printf '%s' "$snap_sha" | grep -Eq '^[0-9a-f]{7,64}$'; then
        echo "detector.sh: source '$kind/$id' snapshot_sha rejected (not a 7-64 char lowercase-hex git sha) — snapshot metadata DROPPED, poll continues (#940)." >&2
        snap_sha=""
        snap_ts=""
      fi
      # A ts must be a sane positive epoch BEFORE any arithmetic touches it: a
      # negative or absurdly large value would make the shell integer comparison
      # below error out and silently KEEP the bad ts — validate first, compare after.
      if [ -n "$snap_ts" ] && ! printf '%s' "$snap_ts" | grep -Eq '^[0-9]{1,12}$'; then
        echo "detector.sh: source '$kind/$id' snapshot_ts rejected (not a sane positive epoch) — snapshot_ts DROPPED, poll continues (#940)." >&2
        snap_ts=""
      fi
      # ts is only meaningful anchored to a revision the consumer can re-verify —
      # a bare number with no sha behind it is the weakest possible attestation,
      # so ts requires a valid sha (#940 review §2).
      if [ -n "$snap_ts" ] && [ -z "$snap_sha" ]; then
        echo "detector.sh: source '$kind/$id' snapshot_ts without a valid snapshot_sha — snapshot_ts DROPPED (unverifiable dating), poll continues (#940)." >&2
        snap_ts=""
      fi
      # A FUTURE ts renders a stale snapshot fresher-than-fresh (negative age) —
      # wrong in the reassuring direction, the exact failure class #940 fixes.
      # Cross-host NTP skew of a few seconds is the NORMAL case, so allow a small
      # slack; beyond it, drop the ts (the sha stays: independently verifiable).
      if [ -n "$snap_ts" ]; then
        # The OPERATOR's knob gets the same discipline as the adapter's ts: it
        # is interpolated into $((...)) under set -u, so a non-numeric value
        # ('300s', '5m', 'abc') would be FATAL to the poll — the one thing this
        # block must never be. Resolve it ONCE, validate, fall back loudly.
        local now_s slack slack_ok
        slack="${WAKE_SNAPSHOT_TS_FUTURE_SLACK:-300}"
        # NOT grep: grep is LINE-oriented, so ^...$ anchors bind per line and a
        # multi-line value ($'300\n8') passes the regex yet is FATAL in $((...)).
        # The case pattern matches the WHOLE string, newlines included. (snap_ts
        # is immune: jq's number type-check above cannot emit an embedded newline.)
        case "$slack" in
        '' | *[!0-9]*) slack_ok=1 ;;
        *) [ "${#slack}" -le 9 ] && slack_ok=0 || slack_ok=1 ;;
        esac
        if [ "$slack_ok" -ne 0 ]; then
          echo "detector.sh: WAKE_SNAPSHOT_TS_FUTURE_SLACK='$slack' is not a plain non-negative integer of at most 9 digits (seconds) — falling back to 300, poll continues (#940)." >&2
          slack=300
        fi
        # Shape validation is not radix validation: bash reads a leading zero as
        # OCTAL, so '08'/'09' pass the shape check yet are FATAL in $((...)), and
        # '0300' silently means 192. Force base-10 so the knob means what the
        # operator wrote (safe: the case pattern above guarantees pure digits).
        slack=$((10#$slack))
        now_s="$(date +%s)"
        if [ "$snap_ts" -gt $((now_s + slack)) ]; then
          echo "detector.sh: source '$kind/$id' snapshot_ts is beyond the ${slack}s future-skew allowance (ts=$snap_ts now=$now_s) — snapshot_ts DROPPED, poll continues (#940)." >&2
          snap_ts=""
        fi
      fi
    else
      echo "detector.sh: source '$kind/$id' wrote UNPARSEABLE snapshot metadata on fd 3 — DROPPED, poll continues (#940)." >&2
    fi
  fi

  # --- anchor-scope (feature (a)) then hash ---------------------------------
  local scoped
  if [ -n "$anchor" ]; then
    scoped="$(printf '%s' "$raw" | _scope_anchor "$anchor")"
    if [ -z "$scoped" ]; then
      echo "detector.sh: FAIL LOUD (G2a) — source '$kind/$id' anchor '$anchor' not present in the observed content (ambiguous: cannot tell 'section removed' from 'hidden'). observed_seq NOT advanced." >&2
      return 1
    fi
  else
    scoped="$raw"
  fi

  local h hf last
  h="$(printf '%s' "$scoped" | _hash)"
  hf="$(_hash_file "$kind" "$id")"
  mkdir -p "$DET_DIR"

  if [ ! -f "$hf" ]; then
    # First-seen: establish the baseline WITHOUT a wake (generalizes the
    # proven deliver-on-NEW poller; startup parity is the reconciler's job,
    # W5, out of scope). Flagged as a deliberate choice in the PR body.
    printf '%s' "$h" | _atomic_write "$hf"
    return 0
  fi

  last="$(tr -d '[:space:]' <"$hf")"
  if [ "$h" = "$last" ]; then
    # DELTA GATE: unchanged -> NO enqueue (0-wasted). Cursor untouched.
    return 0
  fi

  # --- DELTA: enqueue and let the STORE allocate observed_seq (#908) ---------
  # (a revert A->B->A lands here because $h != the stored $last). The store is
  # the sole allocator: we call enqueue WITHOUT --seq and capture the seq it
  # assigns (for logging/locators/emit). We do NOT allocate a seq ourselves.
  local locators emit_ts
  emit_ts="$(date +%s)"
  # NB: `def` is a reserved word in jq — the source-definition arg is `$sdef`.
  # snapshot_sha/snapshot_ts (#940) join only when the adapter attested them.
  locators="$(jq -cn \
    --arg kind "$kind" \
    --arg id "$id" \
    --argjson sdef "$def" \
    --arg hash "$h" \
    --arg ssha "$snap_sha" \
    --arg sts "$snap_ts" \
    '{kind:$kind, id:$id, observed_hash:$hash}
       + (if $ssha != "" then {snapshot_sha: $ssha} else {} end)
       + (if $sts != "" then {snapshot_ts: ($sts|tonumber)} else {} end)
       + ( $sdef | {repo, path, anchor, remote, branches} | with_entries(select(.value != null)) )')"

  local args=(enqueue --locators "$locators" --emit-ts "$emit_ts")
  # Pass class through only when the source declares one; absent => the store
  # defaults to `actionable` (fail-safe, §2.3). Never guess a coalescible class.
  [ -n "$class" ] && args+=(--class "$class")
  local seq
  if ! seq="$("$STORE_SH" "${args[@]}")"; then
    echo "detector.sh: store enqueue FAILED for '$kind/$id' (observed_seq NOT advanced; hash NOT advanced)" >&2
    return 1
  fi

  # Only AFTER a durable enqueue do we advance the per-watch last-observed hash.
  # (store-allocated seq $seq is available for logging/emit if needed.)
  : "$seq"
  printf '%s' "$h" | _atomic_write "$hf"
  return 0
}

# --- one full pass over the watch-list -------------------------------------

cmd_poll_once() {
  local json
  json="$(_load_watchlist)" || exit $?

  if [ -z "${WAKE_DETECTOR_SOURCE_CMD:-}" ]; then
    echo "detector.sh: WAKE_DETECTOR_SOURCE_CMD is not set (no adapter to observe sources)" >&2
    exit 2
  fi

  _wake_init_dir "$STATE_DIR"
  # Detector-run maintenance stale-tmp reap (#927). Stale-tmp cleanup was moved
  # OFF the per-enqueue hot path (it clobbered concurrent in-flight writes); the
  # detector poll tick is a natural once-per-pass maintenance point that bounds
  # crash-left tmp accumulation on the long-running co-feed daemon. Age-scoped,
  # so it can never delete a live in-flight enqueue write (its own or a
  # concurrent reconciler's).
  _wake_clean_stale_tmp "$STATE_DIR"
  mkdir -p "$DET_DIR"

  local failed=0

  # #958 preimage provenance: check the OPERATOR-SIDE preimage definition (the
  # source adapter file, the watch-list, operator-declared extras) BEFORE
  # observing any source. A changed preimage re-baselines EVERY source at once;
  # running the check first means its first-class cause line is enqueued at a
  # LOWER observed_seq than the N per-source deltas it explains, so the digest
  # shows the cause, not just the flood. An infrastructure failure of the check
  # is LOUD and marks this pass failed (G2a discipline — never read as "no
  # change"), but source observation still proceeds: provenance must not be
  # able to starve wake delivery.
  if ! "$SCRIPT_DIR/preimage.sh" check --enqueue; then
    echo "detector.sh: FAIL LOUD — preimage provenance check failed (see preimage.sh above); source observation continues but this pass exits non-zero." >&2
    failed=1
  fi

  # Iterate the DECLARED source-coverage inventory (§4/G3): only sources listed
  # in watches[].sources[] are polled. An omitted source is not observed (and so
  # cannot make anything pass vacuously); a referenced-but-undefined source is a
  # malformed watch-list -> FAIL LOUD.
  local pairs
  pairs="$(printf '%s' "$json" | jq -r '
    [ .watches[].sources[] | "\(.kind)\t\(.id)" ] | unique | .[]')"

  if [ -z "$pairs" ]; then
    echo "detector.sh: watch-list declares no sources under watches[].sources[]" >&2
    exit 2
  fi

  local kind id def class
  while IFS=$'\t' read -r kind id; do
    [ -n "$kind" ] || continue
    # Resolve the source definition from its top-level collection by id.
    local coll
    case "$kind" in
    repo) coll="repos" ;;
    board_file) coll="board_files" ;;
    lane_anchor) coll="lane_anchors" ;;
    *)
      echo "detector.sh: FAIL LOUD — unknown source kind '$kind' in watch-list" >&2
      failed=1
      continue
      ;;
    esac
    def="$(printf '%s' "$json" | jq -c --arg c "$coll" --arg id "$id" \
      '(.[$c] // []) | map(select(.id == $id)) | .[0] // empty')"
    if [ -z "$def" ]; then
      echo "detector.sh: FAIL LOUD (G3 parity) — watch references '$kind/$id' but no such entry is declared in '$coll'" >&2
      failed=1
      continue
    fi
    class="$(printf '%s' "$def" | jq -r '.class // empty')"
    if ! _poll_source "$kind" "$id" "$def" "$class"; then
      failed=1
    fi
  done <<EOF
$pairs
EOF

  [ "$failed" -eq 0 ] || exit 1
}

# --- long-lived single-instance service ------------------------------------

cmd_run() {
  local once=0
  while [ $# -gt 0 ]; do
    case "$1" in
    --once)
      once=1
      shift
      ;;
    *)
      echo "detector.sh run: unknown option '$1'" >&2
      exit 2
      ;;
    esac
  done

  mkdir -p "$DET_DIR"
  local lock="${WAKE_DETECTOR_LOCK:-$DET_DIR/detector.lock}"
  mkdir -p "$(dirname "$lock")"

  # PER-HOST SINGLE-INSTANCE (§1.1): a non-blocking exclusive flock. A second
  # instance CANNOT acquire it and REFUSES loudly rather than double-observing.
  exec 9>"$lock"
  if ! flock -n 9; then
    echo "detector.sh: FAIL LOUD — another detector instance already holds $lock; refusing (per-host single-instance)." >&2
    exit 1
  fi
  # Ready marker so a supervisor/test can observe the lock is held before racing.
  printf '%s\n' "$$" >"$lock.pid" 2>/dev/null || true

  local interval="${WAKE_DETECTOR_INTERVAL:-30}"
  case "$interval" in
  '' | *[!0-9]*) interval=30 ;;
  esac

  while :; do
    # A single failing source must not kill the long-lived service; poll-once
    # already failed loud on stderr for it. The loop keeps serving healthy ones.
    cmd_poll_once || true
    # W6 SEAM (§1.3, off-host dead-man beacon): emit ONE monotonic liveness
    # beacon per poll cycle. This is the ONLY W6 call site in the detector — a
    # single, clean, opt-in seam. It is INERT unless the operator wires a beacon
    # sink (WAKE_BEACON_SINK_CMD); wiring + install-validating that target is W7's
    # job, not the detector's. Liveness is SPLIT from work-triggering, so a beacon
    # emit failure NEVER kills the detector loop — but it is loud (not silent),
    # and the off-host monitor's absence check (beacon.sh check) is the real
    # safety net regardless of what this dying host does.
    if [ -n "${WAKE_BEACON_SINK_CMD:-}" ]; then
      "$SCRIPT_DIR/beacon.sh" emit >/dev/null || \
        echo "detector.sh: WARN — off-host liveness beacon emit failed (see beacon.sh); the off-host absence check remains the authoritative dead-man." >&2
    fi
    [ "$once" -eq 1 ] && break
    sleep "$interval"
  done
}

cmd_cursors() {
  # observed_seq has a SINGLE source of truth now (#908): the STORE cursor. The
  # detector keeps no private counter, so cursors simply reports the store's
  # authoritative cursors (observed_seq / consumed_seq / pending_depth).
  "$STORE_SH" cursors 2>/dev/null || true
}

cmd_validate() {
  local json
  json="$(_load_watchlist)" || exit $?
  local n
  n="$(printf '%s' "$json" | jq -r '[.watches[].sources[]] | length')"
  echo "detector.sh: watch-list OK ($n declared source reference(s))"
}

main() {
  [ $# -ge 1 ] || {
    usage
    exit 2
  }
  local cmd="$1"
  shift
  case "$cmd" in
  poll-once) cmd_poll_once "$@" ;;
  run) cmd_run "$@" ;;
  validate) cmd_validate "$@" ;;
  cursors) cmd_cursors "$@" ;;
  -h | --help | help) usage ;;
  *)
    echo "detector.sh: unknown command '$cmd'" >&2
    usage
    exit 2
    ;;
  esac
}

main "$@"