#!/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 -> assign observed_seq -> enqueue into the W2 durable store.
#   §2.4  Cursor semantics: `observed_seq` is a detector-local monotonic int
#         assigned AT OBSERVATION and is AUTHORITATIVE; 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 detector-local observed_seq
                                         counter and the store cursors.

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"
}

# --- detector-local observed_seq (§2.4, authoritative monotonic int) --------

# _next_observed_seq — atomically increment + return the detector-local counter.
# Assigned AT OBSERVATION (only on a real delta). A single global monotonic
# sequence across ALL watches (source SHAs are descriptors, NOT the cursor).
_next_observed_seq() {
  mkdir -p "$DET_DIR"
  local cur nxt
  cur="$(_wake_read_int "$DET_DIR/observed_seq_counter" 0)"
  nxt=$((cur + 1))
  printf '%s' "$nxt" | _atomic_write "$DET_DIR/observed_seq_counter"
  printf '%s' "$nxt"
}

# --- 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
  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"
  raw="$("$WAKE_DETECTOR_SOURCE_CMD" "$kind" "$id" <"$deftmp" 2>/dev/null)"
  rc=$?
  rm -f "$deftmp"
  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

  # --- 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: assign the next observed_seq AT OBSERVATION and enqueue --------
  # (a revert A->B->A lands here because $h != the stored $last).
  local seq locators emit_ts
  seq="$(_next_observed_seq)"
  emit_ts="$(date +%s)"
  # NB: `def` is a reserved word in jq — the source-definition arg is `$sdef`.
  locators="$(jq -cn \
    --arg kind "$kind" \
    --arg id "$id" \
    --argjson sdef "$def" \
    --arg hash "$h" \
    '{kind:$kind, id:$id, observed_hash:$hash}
       + ( $sdef | {repo, path, anchor, remote, branches} | with_entries(select(.value != null)) )')"

  local args=(enqueue --seq "$seq" --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")
  if ! "$STORE_SH" "${args[@]}" >/dev/null; then
    echo "detector.sh: store enqueue FAILED for '$kind/$id' seq $seq" >&2
    return 1
  fi

  # Only AFTER a durable enqueue do we advance the per-watch last-observed hash.
  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"
  mkdir -p "$DET_DIR"

  # 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 failed=0 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() {
  local counter
  counter="$(_wake_read_int "$DET_DIR/observed_seq_counter" 0)"
  printf 'detector_observed_seq=%s\n' "$counter"
  "$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 "$@"