#!/usr/bin/env bash
# test-wake-detector.sh — RED-FIRST invariant harness for W4 (EPIC #892):
# the per-host, single-instance, delta-gated DETECTOR daemon (detector.sh, A1).
#
# Each test asserts ONE CONVERGED-DESIGN invariant and is designed to go RED if
# that invariant regresses:
#   D1 no-change poll -> NO enqueue (delta-gated; 0-wasted)          (§1.1)
#   D2 a change -> EXACTLY ONE enqueue with a FRESH observed_seq      (§1.2/§2.4)
#   D3 revert A->B->A across polls -> caught (delta detected)         (§2.4)
#   D4 single-instance flock (2nd instance REFUSES)                   (§1.1)
#   D5 watch-list schema_version out of manifest range -> FAIL LOUD  (Gate B)
#   D6 source error/401/403/ambiguous-empty -> FAIL LOUD,
#      observed_seq NOT advanced (the G2a invariant)                  (§4/G2a)
#   D7 anchor-scoped hashing: edit OUTSIDE the anchor is a no-op;
#      edit INSIDE the anchor is a delta                              (§1.1 (a))
#   D8 MIGRATION-RESTART: post-migration first delta ALLOCATES a seq
#      > consumed via the single store-side allocator — never the
#      silent seq<=consumed no-op the old private counter caused      (#908 arrow 3)
#
# Uses FAKE/STUB sources only (no live network). Isolated: every test runs
# against a fresh WAKE_STATE_HOME temp dir.
#
# SC2030/SC2031 are DELIBERATELY disabled: each test runs in its own ( ) subshell
# and re-exports the per-test env (source-adapter path, watch-list path) so the
# environments are isolated and cannot leak between tests. shellcheck reads the
# re-export-per-subshell idiom as "a change that might be lost" — which is
# exactly the isolation we want, not a bug.
# shellcheck disable=SC2030,SC2031
set -uo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
DET="$SCRIPT_DIR/detector.sh"
STORE="$SCRIPT_DIR/store.sh"

command -v jq >/dev/null 2>&1 || {
  echo "SKIP: jq not available" >&2
  exit 0
}
command -v flock >/dev/null 2>&1 || {
  echo "SKIP: flock not available" >&2
  exit 0
}

TMP_ROOT="$(mktemp -d)"
trap 'rm -rf "$TMP_ROOT"' EXIT

# Failures recorded to a FILE (subshell-safe: a subshell cannot mutate a parent
# var, so a var counter would silently swallow failures — the exact anti-pattern
# this harness must never have; mirrors test-wake-store-ack.sh).
FAILFILE="$TMP_ROOT/failures"
: >"$FAILFILE"
pass=0
fail_msg() {
  echo "  FAIL: $*" >&2
  echo "x" >>"$FAILFILE"
}
ok() { pass=$((pass + 1)); }

fresh_state() {
  local d="$TMP_ROOT/$1"
  rm -rf "$d"
  mkdir -p "$d"
  printf '%s' "$d"
}

# depth — pending_depth reported by the store for the current namespace.
depth() { "$STORE" cursors | sed -n 's/pending_depth=//p'; }
# observed_seq is the STORE's single source of truth now (#908): the detector
# keeps no private counter, so `detector.sh cursors` reports the store cursors.
# This asserts the SAME invariant the old detector-private counter did (a delta
# advances observed_seq by exactly one; no-change/first-seen/fail-loud do not) —
# just read from the unified store allocator.
det_seq() { "$DET" cursors | sed -n 's/^observed_seq=//p'; }

# write_watchlist FILE VERSION — a minimal valid watch-list with one repo source
# ("r1") plus optionally a board_file with an anchor ("b1").
write_watchlist() {
  local file="$1" ver="$2"
  cat >"$file" <<EOF
{
  "schema_version": $ver,
  "repos": [{ "id": "r1", "remote": "example/repo", "class": "digest" }],
  "board_files": [{ "id": "b1", "path": "BOARD.md", "anchor": "## LANE-X" }],
  "watches": [
    { "lane": "lane-x", "sources": [
      { "kind": "repo", "id": "r1" },
      { "kind": "board_file", "id": "b1" }
    ] }
  ]
}
EOF
}

# A stub SOURCE ADAPTER whose per-source output is read from files under
# $STUB_DIR/<kind>_<id>, and whose exit code is read from
# $STUB_DIR/<kind>_<id>.rc (default 0). No network. The detector invokes it as
# `<cmd> <kind> <id>` with the source def on stdin (ignored here).
make_stub() {
  local dir="$1"
  cat >"$dir/adapter.sh" <<EOF
#!/usr/bin/env bash
set -u
kind="\$1"; id="\$2"
base="$dir/\${kind}_\${id}"
rc=0
[ -f "\$base.rc" ] && rc="\$(cat "\$base.rc")"
[ -f "\$base" ] && cat "\$base"
exit "\$rc"
EOF
  chmod +x "$dir/adapter.sh"
}

echo "== D1: no-change poll -> NO enqueue (delta-gated) =="
(
  WAKE_STATE_HOME="$(fresh_state d1)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d1stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  wl="$TMP_ROOT/d1.json"
  write_watchlist "$wl" 1
  export WAKE_WATCH_LIST="$wl"
  printf 'SHA-AAA\n' >"$stub/repo_r1"
  printf '## LANE-X\ndecision: hold\n' >"$stub/board_file_b1"
  # Pass 1: first-seen -> baseline, NO wake.
  "$DET" poll-once || fail_msg "D1: baseline pass should succeed"
  [ "$(depth)" = "0" ] || fail_msg "D1: first-seen must baseline silently (deliver-on-new), got depth $(depth)"
  # Pass 2: identical -> STILL no enqueue.
  "$DET" poll-once || fail_msg "D1: unchanged pass should succeed"
  [ "$(depth)" = "0" ] || fail_msg "D1: no-change poll must NOT enqueue, got depth $(depth)"
  [ "$(det_seq)" = "0" ] || fail_msg "D1: observed_seq must not advance without a delta, got $(det_seq)"
) && ok

echo "== D2: a change -> EXACTLY ONE enqueue with a fresh observed_seq =="
(
  WAKE_STATE_HOME="$(fresh_state d2)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d2stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  wl="$TMP_ROOT/d2.json"
  write_watchlist "$wl" 1
  export WAKE_WATCH_LIST="$wl"
  printf 'SHA-AAA\n' >"$stub/repo_r1"
  printf '## LANE-X\ndecision: hold\n' >"$stub/board_file_b1"
  "$DET" poll-once >/dev/null || fail_msg "D2: baseline pass failed" # baseline both
  [ "$(depth)" = "0" ] || fail_msg "D2: baseline should not enqueue"
  # Change ONLY the repo source.
  printf 'SHA-BBB\n' >"$stub/repo_r1"
  "$DET" poll-once >/dev/null || fail_msg "D2: change pass failed"
  [ "$(depth)" = "1" ] || fail_msg "D2: a single change must yield EXACTLY ONE enqueue, got depth $(depth)"
  [ "$(det_seq)" = "1" ] || fail_msg "D2: observed_seq must advance to 1 on the first delta, got $(det_seq)"
  # The enqueued entry carries a fresh observed_seq and the source locator.
  entry="$("$DET" cursors >/dev/null; "$STORE" drain)"
  echo "$entry" | jq -e 'select(.observed_seq==1 and .locators.kind=="repo" and .locators.id=="r1")' >/dev/null \
    || fail_msg "D2: enqueued entry must have observed_seq=1 and repo/r1 locator [$entry]"
  # Poll again with NO further change -> no second enqueue (still exactly one).
  "$DET" poll-once >/dev/null || fail_msg "D2: post-change no-op pass failed"
  [ "$(depth)" = "1" ] || fail_msg "D2: no new change must NOT add a second enqueue, got depth $(depth)"
  [ "$(det_seq)" = "1" ] || fail_msg "D2: observed_seq must stay 1 with no new delta, got $(det_seq)"
) && ok

echo "== D3: revert A->B->A across polls is caught (delta detected) =="
(
  WAKE_STATE_HOME="$(fresh_state d3)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d3stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  wl="$TMP_ROOT/d3.json"
  write_watchlist "$wl" 1
  export WAKE_WATCH_LIST="$wl"
  printf '## LANE-X\nx\n' >"$stub/board_file_b1"
  # Use only the repo source for a clean A->B->A count.
  printf 'STATE-A\n' >"$stub/repo_r1"
  "$DET" poll-once >/dev/null || fail_msg "D3: baseline A failed" # baseline A
  printf 'STATE-B\n' >"$stub/repo_r1"
  "$DET" poll-once >/dev/null || fail_msg "D3: A->B failed" # delta 1
  printf 'STATE-A\n' >"$stub/repo_r1"
  "$DET" poll-once >/dev/null || fail_msg "D3: B->A failed" # delta 2 (the revert)
  # Two deltas total: A->B and the revert B->A. Neither is swallowed as "no change".
  # (board_file b1 never changed, so it contributes 0.)
  [ "$(det_seq)" = "2" ] || fail_msg "D3: revert A->B->A must produce TWO deltas (observed_seq=2), got $(det_seq)"
) && ok

echo "== D4: single-instance flock (2nd instance refuses) =="
(
  WAKE_STATE_HOME="$(fresh_state d4)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d4stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  wl="$TMP_ROOT/d4.json"
  write_watchlist "$wl" 1
  export WAKE_WATCH_LIST="$wl"
  printf 'SHA-AAA\n' >"$stub/repo_r1"
  printf '## LANE-X\nx\n' >"$stub/board_file_b1"
  lock="$WAKE_STATE_HOME/d4.lock"
  export WAKE_DETECTOR_LOCK="$lock"
  export WAKE_DETECTOR_INTERVAL=60
  # First long-lived instance: loops (interval 60), holds the flock.
  "$DET" run >/dev/null 2>&1 &
  runpid=$!
  # Wait until it has acquired the lock (ready pid marker written post-flock).
  for _ in $(seq 1 50); do
    [ -f "$lock.pid" ] && break
    sleep 0.1
  done
  [ -f "$lock.pid" ] || fail_msg "D4: first instance never signalled lock acquisition"
  # Second instance MUST refuse (non-zero) because the flock is held.
  if "$DET" run --once >/dev/null 2>&1; then
    fail_msg "D4: second instance must REFUSE while the flock is held (per-host single-instance)"
  fi
  kill "$runpid" 2>/dev/null || true
  wait "$runpid" 2>/dev/null || true
  # After the holder exits, a fresh instance may acquire the lock. The kernel's
  # fd/flock release can lag process reaping slightly, so allow a bounded wait
  # (the assertion is that the lock IS released eventually, not instantly).
  reacquired=0
  for _ in $(seq 1 30); do
    if "$DET" run --once >/dev/null 2>&1; then
      reacquired=1
      break
    fi
    sleep 0.1
  done
  [ "$reacquired" = "1" ] || fail_msg "D4: a new instance should acquire the lock once the holder is gone"
) && ok

echo "== D5: watch-list schema_version out of manifest range -> FAIL LOUD =="
(
  WAKE_STATE_HOME="$(fresh_state d5)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d5stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  printf 'SHA-AAA\n' >"$stub/repo_r1"
  printf '## LANE-X\nx\n' >"$stub/board_file_b1"
  # schema_version 999 is far outside the manifest's supported range.
  wl="$TMP_ROOT/d5.json"
  write_watchlist "$wl" 999
  export WAKE_WATCH_LIST="$wl"
  err="$("$DET" poll-once 2>&1)"
  rc=$?
  [ "$rc" -ne 0 ] || fail_msg "D5: an out-of-range schema_version must FAIL LOUD (non-zero exit)"
  echo "$err" | grep -qi 'schema_version' || fail_msg "D5: the failure must name schema_version [$err]"
  echo "$err" | grep -qi 'range' || fail_msg "D5: the failure must state it is out of the supported range [$err]"
  # And nothing was enqueued / no cursor advance under a rejected watch-list.
  [ "$(depth)" = "0" ] || fail_msg "D5: a rejected watch-list must not enqueue, got depth $(depth)"
  [ "$(det_seq)" = "0" ] || fail_msg "D5: a rejected watch-list must not advance observed_seq, got $(det_seq)"
  # In-range still works (guards against a validator that rejects everything).
  write_watchlist "$wl" 1
  "$DET" poll-once >/dev/null 2>&1 || fail_msg "D5: an in-range schema_version must be accepted"
) && ok

echo "== D6: source error / ambiguous-empty -> FAIL LOUD, observed_seq NOT advanced (G2a) =="
(
  WAKE_STATE_HOME="$(fresh_state d6)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d6stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  wl="$TMP_ROOT/d6.json"
  write_watchlist "$wl" 1
  export WAKE_WATCH_LIST="$wl"
  printf 'SHA-AAA\n' >"$stub/repo_r1"
  printf '## LANE-X\nx\n' >"$stub/board_file_b1"
  "$DET" poll-once >/dev/null || fail_msg "D6: baseline failed" # baseline
  seq_before="$(det_seq)"

  # (i) A 403-style source error: adapter exits non-zero. MUST fail loud AND NOT
  # be treated as "no change" AND NOT advance observed_seq.
  printf '3\n' >"$stub/repo_r1.rc" # non-zero exit (privacy/403/partial class)
  printf 'FORBIDDEN\n' >"$stub/repo_r1"
  err="$("$DET" poll-once 2>&1)"
  rc=$?
  [ "$rc" -ne 0 ] || fail_msg "D6: a source error must FAIL LOUD (non-zero exit)"
  echo "$err" | grep -qi 'FAIL LOUD' || fail_msg "D6: the source error must be loud [$err]"
  [ "$(det_seq)" = "$seq_before" ] || fail_msg "D6: a source error must NOT advance observed_seq (got $(det_seq), was $seq_before)"
  [ "$(depth)" = "0" ] || fail_msg "D6: a source error must NOT enqueue, got depth $(depth)"

  # (ii) Ambiguous-empty: adapter exits 0 but with EMPTY output. An empty that
  # might mean "hidden" is never "no change" -> FAIL LOUD, no advance.
  rm -f "$stub/repo_r1.rc"
  : >"$stub/repo_r1" # empty, exit 0
  err="$("$DET" poll-once 2>&1)"
  rc=$?
  [ "$rc" -ne 0 ] || fail_msg "D6: an ambiguous-empty response must FAIL LOUD (non-zero exit)"
  echo "$err" | grep -qi 'AMBIGUOUS-EMPTY' || fail_msg "D6: ambiguous-empty must be named in the loud failure [$err]"
  [ "$(det_seq)" = "$seq_before" ] || fail_msg "D6: ambiguous-empty must NOT advance observed_seq (got $(det_seq))"
  [ "$(depth)" = "0" ] || fail_msg "D6: ambiguous-empty must NOT enqueue, got depth $(depth)"

  # (iii) Recovery proof: once the source recovers with a REAL new value, the
  # (un-swallowed) change is delivered — the error never masked it as "seen".
  rm -f "$stub/repo_r1.rc"
  printf 'SHA-RECOVERED\n' >"$stub/repo_r1"
  "$DET" poll-once >/dev/null 2>&1 || true
  [ "$(depth)" -ge 1 ] || fail_msg "D6: after recovery the real change must be delivered (not masked by the prior error)"
  [ "$(det_seq)" -gt "$seq_before" ] || fail_msg "D6: observed_seq must advance only now, on the real post-recovery delta"
) && ok

echo "== D7: anchor-scoped hashing (edit outside anchor = no-op; inside = delta) =="
(
  WAKE_STATE_HOME="$(fresh_state d7)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d7stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  wl="$TMP_ROOT/d7.json"
  write_watchlist "$wl" 1
  export WAKE_WATCH_LIST="$wl"
  # Keep the repo source static so only the anchored board_file is exercised.
  printf 'SHA-STATIC\n' >"$stub/repo_r1"
  printf '## LANE-A\nalpha\n## LANE-X\ndecision: hold\n## LANE-Z\nzulu\n' >"$stub/board_file_b1"
  "$DET" poll-once >/dev/null || fail_msg "D7: baseline failed" # baseline
  d0="$(det_seq)"
  # Edit OUTSIDE the "## LANE-X" anchor (LANE-A / LANE-Z): must be a NO-OP.
  printf '## LANE-A\nALPHA-CHANGED\n## LANE-X\ndecision: hold\n## LANE-Z\nZULU-CHANGED\n' >"$stub/board_file_b1"
  "$DET" poll-once >/dev/null || fail_msg "D7: out-of-anchor pass failed"
  [ "$(det_seq)" = "$d0" ] || fail_msg "D7: an edit OUTSIDE the anchor must NOT wake (anchor-scoped), got seq $(det_seq)"
  [ "$(depth)" = "0" ] || fail_msg "D7: out-of-anchor edit must NOT enqueue, got depth $(depth)"
  # Edit INSIDE the "## LANE-X" anchor: must be a DELTA.
  printf '## LANE-A\nALPHA-CHANGED\n## LANE-X\ndecision: GO\n## LANE-Z\nZULU-CHANGED\n' >"$stub/board_file_b1"
  "$DET" poll-once >/dev/null || fail_msg "D7: in-anchor pass failed"
  [ "$(det_seq)" -gt "$d0" ] || fail_msg "D7: an edit INSIDE the anchor MUST wake (human-decision file edit caught)"
  [ "$(depth)" -ge 1 ] || fail_msg "D7: in-anchor edit must enqueue, got depth $(depth)"
) && ok

echo "== D8: MIGRATION-RESTART — post-migration first delta ALLOCATES > consumed, never a silent no-op (#908, arrow #3) =="
(
  WAKE_STATE_HOME="$(fresh_state d8)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  stub="$TMP_ROOT/d8stub"
  mkdir -p "$stub"
  make_stub "$stub"
  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
  # A watch-list with a SINGLE repo source (clean seq accounting).
  wl="$TMP_ROOT/d8.json"
  cat >"$wl" <<'EOF'
{ "schema_version": 1,
  "repos": [ { "id": "r1", "class": "actionable" } ],
  "watches": [ { "lane": "L", "sources": [ { "kind": "repo", "id": "r1" } ] } ] }
EOF
  export WAKE_WATCH_LIST="$wl"

  # --- simulate adopt -> migrate -> restart --------------------------------
  # §5 migration seeds the STORE cursors (consumed_seq=N, observed_seq=N) so the
  # already-consumed prefix is preserved. It seeds NO detector-private counter —
  # because after #908 the detector HAS none; the store cursor is the single SoT.
  # This is the exact pilot state that used to silently swallow the first delta.
  sd="$WAKE_STATE_HOME/default"
  mkdir -p "$sd"
  printf '5' >"$sd/consumed_seq"
  printf '5' >"$sd/observed_seq"
  : >"$sd/observed.set" # prefix <=5 fully consumed; live window empty
  : >"$sd/pending.jsonl"

  # "Restart" the detector: fresh detector-local state (no hash yet, and — the
  # killer precondition — NO private observed_seq counter exists).
  printf 'STATE-A\n' >"$stub/repo_r1"
  "$DET" poll-once >/dev/null 2>&1 || fail_msg "D8: post-restart baseline pass failed"
  # First-seen baselines silently (deliver-on-new); the store cursor is untouched.
  [ "$(depth)" = "0" ] || fail_msg "D8: baseline must not enqueue, got depth $(depth)"
  [ "$(det_seq)" = "5" ] || fail_msg "D8: migration-seeded observed_seq must be preserved at 5, got $(det_seq)"

  # --- the real, un-consumed obligation: ONE delta -------------------------
  printf 'STATE-B\n' >"$stub/repo_r1"
  "$DET" poll-once >/dev/null 2>&1 || fail_msg "D8: post-migration delta pass failed"
  # THE KILLER ASSERTION: the store (sole allocator) allocates 6 = observed(5)+1,
  # which is > consumed(5). Under the OLD private-counter detector this delta got
  # seq 1 (counter restarted at 0) which is <= consumed 5 and was SILENTLY
  # swallowed by store enqueue's seq<=consumed no-op (depth stayed 0). With the
  # unified allocator it CANNOT be swallowed: it enqueues and WOULD wake.
  [ "$(det_seq)" = "6" ] || fail_msg "D8: post-migration first delta must ALLOCATE observed_seq 6 (>consumed 5), got $(det_seq) — a restart-at-0 private counter would give <=5 and be swallowed"
  [ "$(depth)" = "1" ] || fail_msg "D8: post-migration delta must ENQUEUE a real obligation (depth 1), NOT be a silent no-op, got depth $(depth)"
  # And it is genuinely deliverable (a locator-bearing pending entry the consumer
  # would wake on), not swallowed.
  entry="$("$STORE" drain)"
  echo "$entry" | jq -e 'select(.observed_seq==6 and .locators.kind=="repo" and .locators.id=="r1")' >/dev/null \
    || fail_msg "D8: the enqueued post-migration obligation must be a real deliverable at observed_seq=6 [$entry]"
  # Contiguous-prefix contract still holds: CONSUMED 6 succeeds (6 is observed,
  # 5 is the consumed prefix — no interior gap).
  "$STORE" consume --upto 6 >/dev/null 2>&1 || fail_msg "D8: CONSUMED 6 must succeed over the contiguous prefix after the delta"
) && ok

echo
if [ -s "$FAILFILE" ]; then
  echo "wake detector harness: FAILED ($(grep -c . "$FAILFILE") assertion(s))" >&2
  exit 1
fi
echo "wake detector harness: all invariants passed ($pass groups)"