#!/usr/bin/env bash
# test-wake-store-ack.sh — RED-FIRST invariant harness for W2 (EPIC #892):
# three-cursor durable store (store.sh, A2) + RECEIVED/CONSUMED ack-wrapper
# (ack.sh, A4).
#
# Each test asserts ONE CONVERGED-DESIGN invariant and is designed to go RED if
# that invariant regresses:
#   T1 three-cursor advancement                    (§1.2/§2.4)
#   T2 digest-coalesce-REPLACE vs actionable-APPEND (§1.2/§2.3)
#   T3 contiguous-prefix CONSUMED / gap rejection   (§1.2/§2.2)
#   T4 wake_id DELIVERY-dedup (never re-action)      (§2.2)
#   T5 atomic write-tmp+rename survives crash mid-write (§1.2)
#   T6 durability survives restart (retain until CONSUMED) (§2.3)
#   T7 ack local-write NEVER blocks on network       (§2.2)
#   T8 SOLE store-side allocator: enqueue (no --seq) allocates contiguous
#      observed_seq + anti-swallow fail-loud on explicit seq<=consumed (#908)
#   T9 burn-before-enqueue: a failed durable write must NOT advance the
#      observed_seq cursor (no burned seq / no interior gap)          (#908 arrow 1)
#   T10 concurrency: two concurrent enqueues get DISTINCT seqs (lock)  (#908)
#   T12 consume records the last-consumed observed_hash per (kind,id) into the
#      store-owned record (additive; monotonic last-seq wins; lazily created) (#932)
#   T13 #946 quarantine CLAMP: ordinary consume (store + ack wrapper) REFUSES to
#      advance past a quarantined seq; the refusal names the seq + the force flag
#   T14 #946 forced step-over: --force-past-quarantine advances LOUDLY, prunes the
#      set, and NEVER fabricates a consumed-hash row for the quarantined entry
#   T15 #946 quarantine-sync: full REPLACE semantics (sorted/deduped; empty input
#      CLEARS — the clamp self-heals once the gate is fixed; invalid input refused)
#   T16 #946 quarantine-audit: consumed-hashes rows provably false against the
#      dead-letter ledger are reported (exit 1) and removed only under --repair;
#      healed rows and the ledger itself are untouched
#   T17 #952 quarantine-audit clean sweep names BOTH unprovable residual
#      classes: evidence pruned (no evidence) AND evidence surviving with an
#      empty observed_hash (evidence unusable). Fixture is the VERBATIM live
#      specimen (mos-dt lane dead-letter seq 13, bench/malformed-locator-test)
#      — a nested .locators.* row with NO observed_hash key. Do NOT hand-build
#      a FLAT dead-letter fixture here: consumed-hashes rows are genuinely
#      flat, dead-letter rows are genuinely nested, and a flat fixture makes
#      the audit's CORRECT non-conviction look exactly like the defect under
#      hunt (this false-defect near-miss actually happened in the #951 review)
#
# Isolated: every test runs against a fresh WAKE_STATE_HOME temp dir.
set -uo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# #973: three-valued grep assertion helpers (has_match/count_lines); init saves real stderr for abort loudness.
# shellcheck disable=SC1091
if ! . "$SCRIPT_DIR/_wake-common.sh"; then
  echo "WAKE-ASSERT INIT ABORT: cannot source _wake-common.sh — suite ran ZERO wake assertions (#984)" >&2
  exit 97
fi
wake_assert_init
STORE="$SCRIPT_DIR/store.sh"
ACK="$SCRIPT_DIR/ack.sh"

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

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

# Failures are recorded to a marker FILE, not a shell var: each test runs in a
# subshell (for env isolation) and a subshell cannot mutate a parent variable,
# so a var-based counter would silently swallow failures (the exact bug this
# harness must never have).
FAILFILE="$TMP_ROOT/failures"
: >"$FAILFILE"
pass=0
fail_msg() {
  echo "  FAIL: $*" >&2
  echo "x" >>"$FAILFILE"
}
ok() { pass=$((pass + 1)); }

# jq_any FILE FILTER — true iff any JSONL record in FILE matches FILTER. Uses
# slurp (-s), NOT `jq -e select` (which reflects only the LAST input's value
# over a multi-line JSONL and would misreport presence).
jq_any() {
  local file="$1" filter="$2"
  [ -f "$file" ] || return 1
  [ "$(jq -s "[.[] | select($filter)] | length" "$file" 2>/dev/null || echo 0)" -gt 0 ]
}
# stream_any — same, but reads JSONL from stdin.
stream_any() {
  local filter="$1"
  [ "$(jq -s "[.[] | select($filter)] | length" 2>/dev/null || echo 0)" -gt 0 ]
}

# fresh_state NAME — echoes a fresh isolated state home for export.
fresh_state() {
  local d="$TMP_ROOT/$1"
  rm -rf "$d"
  mkdir -p "$d"
  printf '%s' "$d"
}

echo "== T1: three-cursor advancement =="
(
  WAKE_STATE_HOME="$(fresh_state t1)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  "$STORE" enqueue --seq 1 --class actionable --locators '{"repo":"r","issue":1}' >/dev/null
  "$STORE" enqueue --seq 2 --class actionable --locators '{"repo":"r","issue":2}' >/dev/null
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'observed_seq=2' || fail_msg "T1: observed_seq should be 2 after enqueue 1,2 [$cur]"
  echo "$cur" | has_match -q 'consumed_seq=0' || fail_msg "T1: consumed_seq must NOT advance on enqueue (only on CONSUMED ack) [$cur]"
  echo "$cur" | has_match -q 'pending_depth=2' || fail_msg "T1: pending depth should be 2 [$cur]"
  # consumed_seq advances only on CONSUMED.
  "$STORE" consume --upto 2 >/dev/null
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'consumed_seq=2' || fail_msg "T1: consumed_seq should be 2 after CONSUMED 2 [$cur]"
  echo "$cur" | has_match -q 'pending_depth=0' || fail_msg "T1: pending drained after CONSUMED [$cur]"
) && ok

echo "== T2: digest coalesce-REPLACE vs actionable APPEND =="
(
  WAKE_STATE_HOME="$(fresh_state t2)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  "$STORE" enqueue --seq 1 --class digest --locators '{"head":"aaa"}' >/dev/null
  "$STORE" enqueue --seq 2 --class actionable --locators '{"issue":7}' >/dev/null
  "$STORE" enqueue --seq 3 --class digest --locators '{"head":"bbb"}' >/dev/null
  "$STORE" enqueue --seq 4 --class human --locators '{"from":"peer"}' >/dev/null
  out="$("$STORE" drain)"
  ndigest="$(printf '%s\n' "$out" | jq -c 'select(.class=="digest")' | count_lines . || true)"
  [ "$ndigest" = "1" ] || fail_msg "T2: digest must COALESCE to a single pending entry, got $ndigest"
  head="$(printf '%s\n' "$out" | jq -r 'select(.class=="digest") | .locators.head')"
  [ "$head" = "bbb" ] || fail_msg "T2: newest digest must REPLACE prior (expected bbb, got $head)"
  nactionable="$(printf '%s\n' "$out" | jq -c 'select(.class=="actionable")' | count_lines . || true)"
  nhuman="$(printf '%s\n' "$out" | jq -c 'select(.class=="human")' | count_lines . || true)"
  [ "$nactionable" = "1" ] || fail_msg "T2: actionable must APPEND (never replaced), got $nactionable"
  [ "$nhuman" = "1" ] || fail_msg "T2: human must APPEND / stay durable, got $nhuman"
  # Durability never bypassed: all classes present in the durable store.
  total="$(printf '%s\n' "$out" | count_lines . || true)"
  [ "$total" = "3" ] || fail_msg "T2: durable store should hold digest+actionable+human = 3, got $total"
) && ok

echo "== T3: contiguous-prefix CONSUMED (reject ack N while N-1 unconsumed) =="
(
  WAKE_STATE_HOME="$(fresh_state t3)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  # Gap: observe seq 1 and 3, but NOT 2.
  "$STORE" enqueue --seq 1 --class actionable --locators '{}' >/dev/null
  "$STORE" enqueue --seq 3 --class actionable --locators '{}' >/dev/null
  if "$STORE" consume --upto 3 >/dev/null 2>&1; then
    fail_msg "T3: CONSUMED 3 must be REJECTED while seq 2 is a gap (unconsumed)"
  fi
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'consumed_seq=0' || fail_msg "T3: rejected CONSUMED must NOT advance the cursor [$cur]"
  # Also reject via the ack wrapper.
  if "$ACK" consumed --upto 3 --no-sync >/dev/null 2>&1; then
    fail_msg "T3: ack.sh CONSUMED 3 must be REJECTED over a gap"
  fi
  # Fill the gap; now the contiguous prefix is ackable.
  "$STORE" enqueue --seq 2 --class actionable --locators '{}' >/dev/null
  "$ACK" consumed --upto 3 --no-sync >/dev/null 2>&1 || fail_msg "T3: CONSUMED 3 should succeed once gap at 2 is filled"
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'consumed_seq=3' || fail_msg "T3: consumed_seq should be 3 after contiguous prefix filled [$cur]"
  # Cumulative + can't regress: CONSUMED 2 after 3 is an idempotent no-op.
  "$ACK" consumed --upto 2 --no-sync >/dev/null 2>&1 || fail_msg "T3: cumulative CONSUMED 2 (<=3) should be an idempotent success"
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'consumed_seq=3' || fail_msg "T3: cursor must not regress below 3 [$cur]"
) && ok

echo "== T4: wake_id DELIVERY-dedup (dup delivery = re-RECEIVE, never re-action) =="
(
  WAKE_STATE_HOME="$(fresh_state t4)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  "$STORE" enqueue --seq 1 --class digest --locators '{}' >/dev/null
  r1="$("$ACK" received --wake-id WID-abc)"
  [ "$r1" = "RECEIVED" ] || fail_msg "T4: first delivery should be RECEIVED, got '$r1'"
  r2="$("$ACK" received --wake-id WID-abc)"
  [ "$r2" = "DUP" ] || fail_msg "T4: duplicate delivery of same wake_id should be DUP (re-RECEIVE), got '$r2'"
  # RECEIVED (delivery) must NEVER advance consumed_seq (delivery != consumption).
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'consumed_seq=0' || fail_msg "T4: RECEIVED must not advance consumed_seq [$cur]"
) && ok

echo "== T5: atomic write-tmp+rename survives crash mid-write =="
(
  WAKE_STATE_HOME="$(fresh_state t5)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  "$STORE" enqueue --seq 1 --class actionable --locators '{"issue":1}' >/dev/null
  STATE_DIR="$WAKE_STATE_HOME/default"
  before="$(cat "$STATE_DIR/pending.jsonl")"
  # Simulate a crash mid-write: a leftover atomic-write temp file with GARBAGE
  # that was never renamed over the live file.
  printf '{"observed_seq": 999, TRUNCATED-GARBAGE' >"$STATE_DIR/.wake.tmp.crash12"
  # The live store must be untouched and still valid JSON.
  after="$(cat "$STATE_DIR/pending.jsonl")"
  [ "$before" = "$after" ] || fail_msg "T5: live pending.jsonl changed by a crash temp file"
  drained="$("$STORE" drain)"
  printf '%s\n' "$drained" | jq -e . >/dev/null 2>&1 || fail_msg "T5: drain returned non-JSON — garbage temp corrupted the store"
  printf '%s\n' "$drained" | stream_any '.observed_seq==1' || fail_msg "T5: committed entry (seq 1) lost after simulated crash"
  printf '%s\n' "$drained" | has_match -q 999 && fail_msg "T5: uncommitted garbage (seq 999) leaked into the live store"
  # A subsequent real mutation must succeed DESPITE the stale temp present.
  "$STORE" enqueue --seq 2 --class actionable --locators '{"issue":2}' >/dev/null || fail_msg "T5: enqueue after crash temp failed"
  # #927: the enqueue HOT PATH must NOT reap tmp files — an unconditional delete
  # there clobbered a concurrent enqueue's LIVE in-flight tmp mid-write (spurious
  # "durable pending write FAILED"). So a FRESH stale tmp is intentionally left
  # untouched by an enqueue; reaping it on the hot path is exactly the bug.
  [ -e "$STATE_DIR/.wake.tmp.crash12" ] || fail_msg "T5: the enqueue hot path must NOT delete tmp files off its own write (that hot-path reap was the #927 clobber)"
  d2="$("$STORE" drain)"
  [ "$(printf '%s\n' "$d2" | has_match -c .)" = "2" ] || fail_msg "T5: store not healthy after crash+recovery (expected 2 entries)"
  # Bounded accumulation is preserved via a MAINTENANCE reap (store.sh init /
  # detector tick), age-scoped so it only removes DEMONSTRABLY-orphaned tmps
  # (older than any plausible in-flight write) and never a live one. Age the
  # crash temp into the past so it is unambiguously orphaned, then run the
  # maintenance path and confirm it is reaped.
  touch -d '1 hour ago' "$STATE_DIR/.wake.tmp.crash12" 2>/dev/null ||
    touch -t "$(date -d '1 hour ago' +%Y%m%d%H%M.%S 2>/dev/null || echo 197001010000)" "$STATE_DIR/.wake.tmp.crash12" 2>/dev/null || true
  "$STORE" init >/dev/null 2>&1 || fail_msg "T5: store.sh init (maintenance) failed"
  [ -e "$STATE_DIR/.wake.tmp.crash12" ] && fail_msg "T5: maintenance reap (store.sh init) must remove a demonstrably-orphaned stale temp (bounded accumulation)"
  d3="$("$STORE" drain)"
  [ "$(printf '%s\n' "$d3" | has_match -c .)" = "2" ] || fail_msg "T5: store not healthy after maintenance reap (expected 2 entries)"
  printf '%s\n' "$d3" | jq -e . >/dev/null 2>&1 || fail_msg "T5: drain returned non-JSON after maintenance reap"
) && ok

echo "== T5b: _atomic_write is tmp+rename (killing the writer mid-write leaves the target intact) =="
(
  WAKE_STATE_HOME="$(fresh_state t5b)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  # shellcheck disable=SC1091
  . "$SCRIPT_DIR/_wake-common.sh"
  d="$WAKE_STATE_HOME/atomic"
  mkdir -p "$d"
  target="$d/file"
  printf 'GOOD\n' | _atomic_write "$target"
  # Feed a writer via a FIFO but NEVER send EOF, then kill it mid-write. With
  # true tmp+rename the partial bytes land in a temp file and the rename never
  # runs, so `target` keeps its committed content. A direct (non-atomic) write
  # would have streamed the garbage straight into `target`.
  fifo="$d/fifo"
  mkfifo "$fifo"
  ( _atomic_write "$target" <"$fifo" ) &
  wpid=$!
  exec 9>"$fifo"
  printf 'PARTIAL-GARBAGE-NO-EOF' >&9
  sleep 0.3
  pkill -9 -P "$wpid" 2>/dev/null || true
  kill -9 "$wpid" 2>/dev/null || true
  exec 9>&-
  wait "$wpid" 2>/dev/null || true
  got="$(cat "$target" 2>/dev/null)"
  [ "$got" = "GOOD" ] || fail_msg "T5b: target corrupted by a killed mid-write (got '$got') — write is not tmp+rename atomic"
) && ok

echo "== T6: durability survives restart (retain until CONSUMED) =="
(
  WAKE_STATE_HOME="$(fresh_state t6)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  # "Session 1": enqueue then vanish (no in-memory state carried).
  "$STORE" enqueue --seq 1 --class human --locators '{"from":"peer"}' >/dev/null
  "$STORE" enqueue --seq 2 --class actionable --locators '{"issue":9}' >/dev/null
  # "Session 2": a brand-new process (this subshell invocation of store.sh) must
  # see the persisted entries — nothing was held in memory.
  d="$("$STORE" drain)"
  [ "$(printf '%s\n' "$d" | has_match -c .)" = "2" ] || fail_msg "T6: entries not retained across restart (expected 2)"
  printf '%s\n' "$d" | stream_any '.class=="human"' || fail_msg "T6: a human message was lost across restart (durability bypassed)"
  # Retained UNTIL consumed: after CONSUMED they are released.
  "$STORE" consume --upto 2 >/dev/null
  d2="$("$STORE" drain)"
  n2="$(printf '%s\n' "$d2" | count_lines . || true)"
  [ "$n2" = "0" ] || fail_msg "T6: entries should be released only AFTER CONSUMED (still $n2 pending)"
) && ok

echo "== T7: ack local-write NEVER blocks on network =="
(
  WAKE_STATE_HOME="$(fresh_state t7)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  "$STORE" enqueue --seq 1 --class digest --locators '{}' >/dev/null
  # A network-shaped shim on PATH: if the SYNCHRONOUS ack path ever invoked the
  # network directly, this marker would appear before ack returns.
  bin="$TMP_ROOT/t7bin"
  mkdir -p "$bin"
  for netcmd in curl wget nc ssh; do
    cat >"$bin/$netcmd" <<EOF
#!/usr/bin/env bash
touch "$WAKE_STATE_HOME/NET_CALLED_SYNC"
exit 0
EOF
    chmod +x "$bin/$netcmd"
  done
  # Background sync deliberately SLOW: proves the ack does not wait for it.
  export WAKE_ACK_SYNC_CMD="sleep 3; touch '$WAKE_STATE_HOME/SYNC_DONE'"
  # Sub-second timing (#918): `date +%s` is SECOND-granularity, so a
  # legitimately fast (sub-second) ack that straddles a wall-clock second
  # boundary can misreport elapsed=2s and flake under CI load (observed on
  # Woodpecker pipeline 2057; 20/20 passes locally elsewhere). Use
  # `date +%s.%N` and compare in milliseconds via awk (bash `[ ]` can't do
  # float/ms math) against a threshold comfortably below the 3s background
  # sync — proving the ack did NOT wait on it — while tolerating a
  # legitimate sub-second-to-~1s ack across a boundary.
  start="$(date +%s.%N)"
  out="$(PATH="$bin:$PATH" "$ACK" consumed --upto 1 --wake-id WID-1)"
  end="$(date +%s.%N)"
  elapsed_ms="$(awk -v s="$start" -v e="$end" 'BEGIN { printf "%d", (e - s) * 1000 }')"
  [ "$elapsed_ms" -lt 1500 ] || fail_msg "T7: ack path blocked ${elapsed_ms}ms — must return without waiting on the background sync"
  echo "$out" | has_match -q 'CONSUMED 1' || fail_msg "T7: CONSUMED not reported [$out]"
  # Local write is durable IMMEDIATELY (no network needed to record the ack).
  STATE_DIR="$WAKE_STATE_HOME/default"
  jq_any "$STATE_DIR/ack-ledger.jsonl" '.type=="CONSUMED" and .upto==1' ||
    fail_msg "T7: CONSUMED not written to the local ledger synchronously"
  consumed_now="$("$STORE" cursors | sed -n 's/consumed_seq=//p')"
  [ "$consumed_now" = "1" ] || fail_msg "T7: consumed_seq not advanced by the local ack write (got '$consumed_now')"
  # The synchronous path must NOT have touched the network.
  [ -e "$WAKE_STATE_HOME/NET_CALLED_SYNC" ] && fail_msg "T7: a network command was invoked on the SYNCHRONOUS ack path"
  # And the slow sync must still be in flight (proves it was truly backgrounded).
  [ -e "$WAKE_STATE_HOME/SYNC_DONE" ] && fail_msg "T7: background sync finished synchronously — it was not detached"
  true
) && ok

echo "== T8: SOLE store-side allocator — enqueue (no --seq) allocates contiguous seqs + anti-swallow fail-loud (#908) =="
(
  WAKE_STATE_HOME="$(fresh_state t8)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  # No --seq: the STORE allocates and PRINTS the assigned observed_seq.
  s1="$("$STORE" enqueue --class actionable --locators '{"n":1}')"
  s2="$("$STORE" enqueue --class actionable --locators '{"n":2}')"
  [ "$s1" = "1" ] || fail_msg "T8: first store-allocated observed_seq must be 1, got '$s1'"
  [ "$s2" = "2" ] || fail_msg "T8: second store-allocated observed_seq must be 2 (contiguous), got '$s2'"
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'observed_seq=2' || fail_msg "T8: observed_seq cursor should be 2 [$cur]"
  echo "$cur" | has_match -q 'pending_depth=2' || fail_msg "T8: both allocations must be durably enqueued [$cur]"
  # ANTI-SWALLOW: consume the prefix, then a LEGACY explicit --seq inside the
  # consumed prefix must FAIL LOUD (never the old silent seq<=consumed no-op).
  "$STORE" consume --upto 2 >/dev/null
  before_depth="$("$STORE" cursors | sed -n 's/pending_depth=//p')"
  if "$STORE" enqueue --seq 1 --class actionable --locators '{}' >/dev/null 2>&1; then
    fail_msg "T8: explicit --seq 1 <= consumed_seq 2 must FAIL LOUD (anti-swallow), not be a silent no-op"
  fi
  err="$("$STORE" enqueue --seq 1 --class actionable --locators '{}' 2>&1 || true)"
  echo "$err" | has_match -qi 'anti-swallow' || fail_msg "T8: the refusal must name the anti-swallow guarantee [$err]"
  after_depth="$("$STORE" cursors | sed -n 's/pending_depth=//p')"
  [ "$before_depth" = "$after_depth" ] || fail_msg "T8: a refused enqueue must not mutate the store (depth $before_depth->$after_depth)"
  # And the NEXT real allocation is > consumed (2) — never inside the prefix.
  s3="$("$STORE" enqueue --class actionable --locators '{"n":3}')"
  [ "$s3" = "3" ] || fail_msg "T8: post-consume allocation must be 3 (>consumed 2), got '$s3'"
) && ok

echo "== T9: burn-before-enqueue — a failed durable write must NOT advance observed_seq (no burned seq / no gap) (#908 arrow 1) =="
(
  WAKE_STATE_HOME="$(fresh_state t9)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  STATE_DIR="$WAKE_STATE_HOME/default"
  # One successful allocation: observed_seq=1, and the enqueue lock now exists.
  s1="$("$STORE" enqueue --class actionable --locators '{"n":1}')"
  [ "$s1" = "1" ] || fail_msg "T9: baseline allocation should be 1, got '$s1'"
  obs_before="$("$STORE" cursors | sed -n 's/^observed_seq=//p')"
  [ "$obs_before" = "1" ] || fail_msg "T9: observed_seq should be 1 before the forced failure, got '$obs_before'"
  # Force the durable PENDING write to FAIL via the MOUNT-FREE, privilege-invariant
  # test-only fault seam (#934): WAKE_TEST_FAULT=pending makes store.sh's
  # _atomic_write to pending.jsonl (the durable write's exact target — see
  # _wake-common.sh) report the commit as failed, WITHOUT any unshare/bind-mount.
  # It exercises the SAME #908 arrow-1 fail-loud path the old EBUSY-on-mountpoint
  # trigger did, but RUNS UNPRIVILEGED, so it EXECUTES (never skips) in the real
  # non-privileged CI runner that denies mount-in-userns. observed_seq is left an
  # ordinary writable file, untouched by the seam, so a regression that commits the
  # cursor before/independent of the durable write is still genuinely caught below.
  # The seam is scoped to THIS single enqueue via a command-prefix env assignment,
  # so it never leaks to the cursors/consume/enqueue calls that follow.
  WAKE_TEST_FAULT=pending "$STORE" enqueue --class actionable --locators '{"n":2}' >/dev/null 2>&1
  rc=$?
  [ "$rc" -ne 0 ] || fail_msg "T9: an enqueue whose durable write fails must EXIT NON-ZERO (fail loud)"
  # THE ARROW-#1 ASSERTION: the observed_seq cursor did NOT advance — no seq was
  # burned by a write that never reached the store. (RED if the cursor bumps
  # before/independent of the durable write.)
  obs_after="$("$STORE" cursors | sed -n 's/^observed_seq=//p')"
  [ "$obs_after" = "1" ] || fail_msg "T9: a failed enqueue must NOT advance observed_seq (burned seq -> permanent gap), got '$obs_after'"
  # No interior gap was created: CONSUMED 1 (the only real obligation) still holds.
  "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "T9: CONSUMED 1 must remain valid — no burned-seq gap"
  # And a fresh allocation resumes cleanly at 2 (not 3 — nothing was burned).
  s="$("$STORE" enqueue --class actionable --locators '{"n":"2b"}')"
  [ "$s" = "2" ] || fail_msg "T9: post-failure allocation must resume at 2 (no burned seq), got '$s'"
) && ok

echo "== T10: concurrency — two concurrent enqueues get DISTINCT seqs (enqueue lock) (#908) =="
if command -v flock >/dev/null 2>&1; then
  (
    WAKE_STATE_HOME="$(fresh_state t10)"
    export WAKE_STATE_HOME
    unset WAKE_AGENT
    "$STORE" init >/dev/null 2>&1 || true
    o1="$TMP_ROOT/t10.o1"
    o2="$TMP_ROOT/t10.o2"
    # --- deterministic start barrier (#923 de-flake; NOT a sleep) --------------
    # A plain `cmd & cmd & wait` gives no guarantee the two enqueues are ever
    # truly in-flight together — the scheduler can run one to completion before
    # the other is even forked, so the flock contention this test exists to
    # prove is never really exercised. Force REAL concurrency: both children
    # block on a shared start gate (an flock'd file — no sleep, no timing
    # guess) until BOTH have confirmed they are launched and waiting; only then
    # does the parent release the gate, so both race into store.sh's own
    # enqueue lock at (as close to) the same instant as the scheduler allows.
    gate="$TMP_ROOT/t10.gate"
    r1="$TMP_ROOT/t10.ready1"
    r2="$TMP_ROOT/t10.ready2"
    rm -f "$r1" "$r2"
    exec 6>"$gate"
    flock -x 6 # gate CLOSED: a shared-locker below blocks here until released.

    (
      : >"$r1" # "child 1 is launched and about to block on the gate"
      exec 7<"$gate"
      flock -s 7 # blocks until the parent drops its exclusive lock
      flock -u 7
      exec 7<&-
      "$STORE" enqueue --class actionable --locators '{"c":1}'
    ) >"$o1" 2>/dev/null &
    p1=$!
    (
      : >"$r2"
      exec 7<"$gate"
      flock -s 7
      flock -u 7
      exec 7<&-
      "$STORE" enqueue --class actionable --locators '{"c":2}'
    ) >"$o2" 2>/dev/null &
    p2=$!

    # Busy-poll (NOT a sleep — no timing assumption) until BOTH children have
    # confirmed they are launched and blocked on the gate, so the contention
    # window below is genuine rather than incidental.
    spins=0
    while [ ! -e "$r1" ] || [ ! -e "$r2" ]; do
      spins=$((spins + 1))
      if [ "$spins" -gt 2000000 ]; then
        fail_msg "T10: start barrier never saw both children launch — cannot construct genuine concurrency"
        break
      fi
    done

    # Release the gate: both children's pending shared-lock acquisitions are
    # granted together, so they race into store.sh's OWN enqueue lock for real.
    flock -u 6
    exec 6>&-

    wait "$p1"; wait "$p2"
    a="$(tr -d '[:space:]' <"$o1")"
    b="$(tr -d '[:space:]' <"$o2")"
    [ -n "$a" ] && [ -n "$b" ] || fail_msg "T10: both concurrent enqueues must print an allocated seq (got '$a','$b')"
    [ "$a" != "$b" ] || fail_msg "T10: two concurrent enqueues must get DISTINCT seqs, both got '$a' (aliasing / lost write without the lock)"
    cur="$("$STORE" cursors)"
    echo "$cur" | has_match -q 'observed_seq=2' || fail_msg "T10: observed_seq must reach 2 after two enqueues [$cur]"
    echo "$cur" | has_match -q 'pending_depth=2' || fail_msg "T10: both entries must be durably stored (no lost write) [$cur]"
    # The two allocated seqs are exactly {1,2} (distinct, contiguous, gapless).
    lo="$a"; hi="$b"; [ "$a" -gt "$b" ] && { lo="$b"; hi="$a"; }
    { [ "$lo" = "1" ] && [ "$hi" = "2" ]; } || fail_msg "T10: concurrent seqs must be {1,2}, got {$lo,$hi}"
  ) && ok
else
  echo "  SKIP: flock not available (concurrency guarantee needs flock; matches detector single-instance SKIP)"
  ok
fi

echo "== T11: final observed_seq cursor write FAILURE — fail loud + observed.set/cursor stay CONSISTENT (#917 def-in-depth) =="
(
  WAKE_STATE_HOME="$(fresh_state t11)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  STATE_DIR="$WAKE_STATE_HOME/default"
  # Baseline: one clean allocation. observed_seq cursor=1, observed.set={1}, the
  # cursor FILE now exists (so the faulted enqueue below exercises the cursor-WRITE
  # commit, not a first-time seed via _wake_init_dir).
  s1="$("$STORE" enqueue --class actionable --locators '{"n":1}')"
  [ "$s1" = "1" ] || fail_msg "T11: baseline allocation should be 1, got '$s1'"
  obs_before="$("$STORE" cursors | sed -n 's/^observed_seq=//p')"
  [ "$obs_before" = "1" ] || fail_msg "T11: observed_seq should be 1 before the forced failure, got '$obs_before'"

  # Force ONLY the FINAL observed_seq cursor _atomic_write to fail, AFTER the
  # pending + observed.set writes have already succeeded, via the MOUNT-FREE,
  # privilege-invariant test-only fault seam (#934): WAKE_TEST_FAULT=cursor fails
  # ONLY the _atomic_write whose target basename is observed_seq (the CURSOR).
  # pending.jsonl and observed.set stay ordinary writable files, so their writes
  # SUCCEED and only the cursor commit fails — exactly the #917 window (cursor-write
  # failure after a good observed.set write). Because the baseline enqueue already
  # created observed_seq, _wake_init_dir does NOT rewrite it, so within this faulted
  # enqueue the ONLY _atomic_write to observed_seq is the final cursor commit; the
  # observed.set rollback write (a different basename) still succeeds. No
  # unshare/bind-mount: this RUNS UNPRIVILEGED, so it EXECUTES (never skips) in the
  # real non-privileged CI runner that denies mount-in-userns. The seam is scoped to
  # THIS single enqueue via a command-prefix env assignment, so it never leaks to
  # the cursors/consume calls that assert the outcome.
  errfile="$TMP_ROOT/t11.err"
  : >"$errfile"
  WAKE_TEST_FAULT=cursor "$STORE" enqueue --class actionable --locators '{"n":2}' >/dev/null 2>"$errfile"
  rc=$?

  # (1) FAIL LOUD: the cursor-write failure must EXIT NON-ZERO, never a silent
  # success (RED against pre-#917 code, whose ungated final cursor write swallows
  # the _atomic_write failure and returns 0).
  [ "$rc" -ne 0 ] || fail_msg "T11: an enqueue whose FINAL cursor write fails must EXIT NON-ZERO (fail loud), got rc=$rc"
  # The loud diagnostic names the cursor write (not a generic error).
  has_match -qi 'cursor' "$errfile" || fail_msg "T11: the failure diagnostic must name the observed_seq cursor write [$(cat "$errfile")]"

  # (2) The cursor did NOT advance — the allocation is NOT committed.
  obs_after="$("$STORE" cursors | sed -n 's/^observed_seq=//p')"
  [ "$obs_after" = "1" ] || fail_msg "T11: a failed cursor write must NOT leave the cursor advanced, got '$obs_after'"

  # (3) THE #917 CONSISTENCY ASSERTION: observed.set and the cursor must NEVER be
  # left cross-file inconsistent. observed.set must carry NO seq greater than the
  # cursor — either the observed.set advance was rolled back (neither advances) or
  # both advanced together; there is no state where observed.set is ahead of the
  # cursor. RED against pre-#917 code: observed.set={1,2} while the cursor lags at
  # 1, so seq 2 is stranded above the cursor.
  stranded="$(awk -v c="$obs_after" 'NF && $1+0 > c' "$STATE_DIR/observed.set" 2>/dev/null || true)"
  [ -z "$stranded" ] || fail_msg "T11: observed.set left CROSS-FILE INCONSISTENT with the cursor — seq(s) above cursor $obs_after: [$stranded]"

  # (4) The consumed prefix is intact — no interior gap corrupts future consume.
  "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "T11: CONSUMED 1 must remain valid after the failed cursor write (no gap)"
) && ok

echo "== T12: #932 — consume records the last-consumed observed_hash per (kind,id) into the store-owned record (additive; monotonic; keyed by kind,id) =="
(
  WAKE_STATE_HOME="$(fresh_state t12)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  rec="$WAKE_STATE_HOME/default/consumed-hashes.jsonl"
  # Two source states for the SAME (kind,id): the store must record only the
  # LAST-consumed (highest-seq) hash for that key; plus a distinct (kind,id).
  s1="$("$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"r1","observed_hash":"HASH-A"}')"
  s2="$("$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"r1","observed_hash":"HASH-B"}')"
  s3="$("$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"r2","observed_hash":"HASH-C"}')"
  [ "$s1 $s2 $s3" = "1 2 3" ] || fail_msg "T12: contiguous allocation expected 1 2 3, got '$s1 $s2 $s3'"
  # BEFORE consume there is no record (additive / lazily created).
  [ ! -f "$rec" ] || fail_msg "T12: the last-consumed record must not exist before any consume"
  "$STORE" consume --upto 3 >/dev/null 2>&1 || fail_msg "T12: CONSUMED 3 must succeed over the gapless prefix"
  # AFTER consume the store-owned record exists and holds ONE entry per (kind,id).
  [ -f "$rec" ] || fail_msg "T12: consume must write the store-owned last-consumed record ($rec)"
  nkeys="$(jq -s '[ .[] | {kind,id} ] | unique | length' "$rec" 2>/dev/null || echo 0)"
  [ "$nkeys" = "2" ] || fail_msg "T12: the record must hold exactly one entry per (kind,id) — 2 keys, got $nkeys"
  # repo/r1 must record the LAST-consumed hash (HASH-B at seq 2), never HASH-A.
  r1h="$(jq -sr '[ .[] | select(.kind=="repo" and .id=="r1") ] | .[0].observed_hash' "$rec" 2>/dev/null)"
  [ "$r1h" = "HASH-B" ] || fail_msg "T12: repo/r1 must record the LAST-consumed hash HASH-B (seq 2), got '$r1h'"
  r2h="$(jq -sr '[ .[] | select(.kind=="repo" and .id=="r2") ] | .[0].observed_hash' "$rec" 2>/dev/null)"
  [ "$r2h" = "HASH-C" ] || fail_msg "T12: repo/r2 must record HASH-C, got '$r2h'"
  # ADDITIVE: existing on-disk state files are unchanged/consistent post-consume.
  "$STORE" cursors | has_match -q 'consumed_seq=3' || fail_msg "T12: consumed_seq must be 3 after CONSUMED 3"
) && ok

echo "== T13: #946 — ordinary consume REFUSES to advance past a quarantined seq (store + ack paths) =="
(
  WAKE_STATE_HOME="$(fresh_state t13)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"a","observed_hash":"HA"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"b","observed_hash":"HB"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"c","observed_hash":"HC"}' >/dev/null
  printf '2\n' | "$STORE" quarantine-sync || fail_msg "T13: quarantine-sync must accept a valid seq list"
  # A quarantined seq was dead-lettered at render and NEVER delivered in any
  # digest; the ordinary path must REFUSE to record it consumed (#946: a force
  # flag the ordinary path can bypass is decoration).
  if "$STORE" consume --upto 3 >/dev/null 2>&1; then
    fail_msg "T13: ordinary consume --upto 3 must be REFUSED while seq 2 is quarantined"
  fi
  err="$("$STORE" consume --upto 3 2>&1 >/dev/null || true)"
  echo "$err" | has_match -q 'quarantined seq(s): 2' || fail_msg "T13: the refusal must NAME the quarantined seq [$err]"
  echo "$err" | has_match -q -- '--force-past-quarantine' || fail_msg "T13: the refusal must NAME the force flag [$err]"
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'consumed_seq=0' || fail_msg "T13: a refused consume must NOT advance the cursor [$cur]"
  # BELOW the quarantined seq the ordinary path is unaffected.
  "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "T13: consume --upto 1 (below the quarantined seq) must succeed"
  # The ack wrapper propagates the refusal — no ordinary-path bypass exists.
  if "$ACK" consumed --upto 3 --no-sync >/dev/null 2>&1; then
    fail_msg "T13: ack.sh consumed --upto 3 must be REFUSED while seq 2 is quarantined (ordinary-path bypass)"
  fi
  cur="$("$STORE" cursors)"
  echo "$cur" | has_match -q 'consumed_seq=1' || fail_msg "T13: cursor must still be 1 after the refused ack [$cur]"
) && ok

echo "== T14: #946 — FORCED step-over is LOUD, prunes the set, and NEVER fabricates a consumed-hash row for the quarantined entry =="
(
  WAKE_STATE_HOME="$(fresh_state t14)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  rec="$WAKE_STATE_HOME/default/consumed-hashes.jsonl"
  qf="$WAKE_STATE_HOME/default/quarantined.set"
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"a","observed_hash":"HA"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"b","observed_hash":"HB"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"c","observed_hash":"HC"}' >/dev/null
  printf '2\n' | "$STORE" quarantine-sync || fail_msg "T14: quarantine-sync failed"
  errf="$TMP_ROOT/t14.err"
  out="$("$STORE" consume --upto 3 --force-past-quarantine 2>"$errf")"
  rc=$?
  [ "$rc" -eq 0 ] || fail_msg "T14: forced consume must succeed (rc=$rc) [$(cat "$errf")]"
  [ "$out" = "3" ] || fail_msg "T14: forced consume must print the new cursor 3, got '$out'"
  has_match -q 'FORCED PAST QUARANTINE' "$errf" || fail_msg "T14: the forced path must be LOUD on stderr [$(cat "$errf")]"
  has_match -q 'seq 2' "$errf" || fail_msg "T14: the forced-path diagnostic must name the stepped-over seq 2 [$(cat "$errf")]"
  "$STORE" cursors | has_match -q 'consumed_seq=3' || fail_msg "T14: forced consume must advance the cursor to 3"
  # NO FALSE WITNESS: the quarantined entry (repo/b) was NEVER delivered, so no
  # consumed-hash row may exist for it — even on the forced path (the reconciler
  # re-enumerating it once is safe-but-noisy; a false witness silences it
  # forever). Its delivered siblings' rows must exist.
  jq_any "$rec" '.kind=="repo" and .id=="a" and .observed_hash=="HA"' || fail_msg "T14: the delivered sibling repo/a must have its consumed-hash row"
  jq_any "$rec" '.kind=="repo" and .id=="c" and .observed_hash=="HC"' || fail_msg "T14: the delivered sibling repo/c must have its consumed-hash row"
  jq_any "$rec" '.kind=="repo" and .id=="b"' && fail_msg "T14: the quarantined entry repo/b must have NO consumed-hash row (a row would witness a delivery that never happened)"
  # The stepped-over seq is PRUNED from the set (it is consumed now; a stale
  # entry would re-refuse forever).
  has_match -qxF '2' "$qf" 2>/dev/null && fail_msg "T14: seq 2 must be PRUNED from quarantined.set after the forced step-over"
  # The ack wrapper's force flag passes through, stays LOUD on stderr, and
  # still reports a CLEAN cursor line on stdout.
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"d","observed_hash":"HD"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"e","observed_hash":"HE"}' >/dev/null
  printf '5\n' | "$STORE" quarantine-sync || fail_msg "T14: quarantine-sync (2nd) failed"
  errf2="$TMP_ROOT/t14b.err"
  out2="$("$ACK" consumed --upto 5 --no-sync --force-past-quarantine 2>"$errf2")"
  rc2=$?
  [ "$rc2" -eq 0 ] || fail_msg "T14: forced ack must succeed (rc=$rc2) [$(cat "$errf2")]"
  echo "$out2" | has_match -q '^CONSUMED 5$' || fail_msg "T14: forced ack must report a CLEAN cursor line 'CONSUMED 5', got '$out2'"
  has_match -q 'FORCED PAST QUARANTINE' "$errf2" || fail_msg "T14: the forced-path loudness must survive the ack wrapper (stderr) [$(cat "$errf2")]"
  jq_any "$rec" '.kind=="repo" and .id=="e"' && fail_msg "T14: the quarantined repo/e must have NO consumed-hash row via the forced ack path either"
  true
) && ok

echo "== T15: #946 — quarantine-sync is a full REPLACE (sorted, deduped; empty input CLEARS; invalid input REFUSED) =="
(
  WAKE_STATE_HOME="$(fresh_state t15)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  qf="$WAKE_STATE_HOME/default/quarantined.set"
  printf '3\n1\n3\n' | "$STORE" quarantine-sync || fail_msg "T15: sync of a valid list must succeed"
  [ "$(cat "$qf" 2>/dev/null)" = "$(printf '1\n3')" ] || fail_msg "T15: set must be sorted+deduped {1,3}, got [$(cat "$qf" 2>/dev/null)]"
  printf '2\n' | "$STORE" quarantine-sync || fail_msg "T15: re-sync must succeed"
  [ "$(cat "$qf" 2>/dev/null)" = "2" ] || fail_msg "T15: sync must REPLACE, not merge — expected {2}, got [$(cat "$qf" 2>/dev/null)]"
  # Empty input CLEARS the set: the set is re-DERIVED per authoritative render,
  # never accumulated, so a fixed locator gate self-heals the clamp.
  : | "$STORE" quarantine-sync || fail_msg "T15: empty sync (clear) must succeed"
  [ ! -s "$qf" ] || fail_msg "T15: empty sync must CLEAR the set, got [$(cat "$qf")]"
  # Invalid input is refused loudly and must not corrupt the set.
  printf '1\n' | "$STORE" quarantine-sync || fail_msg "T15: re-seed failed"
  if printf 'abc\n' | "$STORE" quarantine-sync >/dev/null 2>&1; then
    fail_msg "T15: a non-integer line must be REFUSED"
  fi
  [ "$(cat "$qf" 2>/dev/null)" = "1" ] || fail_msg "T15: a refused sync must leave the set untouched, got [$(cat "$qf" 2>/dev/null)]"
  # End-to-end: a cleared set stops clamping (the #944 recovery case).
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"x","observed_hash":"H1"}' >/dev/null
  if "$STORE" consume --upto 1 >/dev/null 2>&1; then
    fail_msg "T15: consume --upto 1 must be refused while seq 1 is quarantined"
  fi
  : | "$STORE" quarantine-sync || fail_msg "T15: clear failed"
  "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "T15: after the set is cleared (gate fixed), the ordinary consume must succeed — the clamp must self-heal"
) && ok

echo "== T16: #946 — quarantine-audit: a consumed-hash row matching a dead-letter entry on (kind,id,seq,hash) at/below consumed_seq is PROVABLY FALSE; --repair removes ONLY those rows =="
(
  WAKE_STATE_HOME="$(fresh_state t16)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  STATE_DIR="$WAKE_STATE_HOME/default"
  rec="$STATE_DIR/consumed-hashes.jsonl"
  dl="$STATE_DIR/dead-letter.jsonl"
  # Rebuild the historical false-witness state via the REAL flow the defect
  # used: X@1 was quarantined (dead-lettered) yet consumed under pre-#946 code;
  # Y@2 is clean; Z re-emitted (dead-lettered at seq 3, healed by seq 4 winning
  # the per-key max_by merge — Finding A's live-canary shape).
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"X","observed_hash":"HX"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"Y","observed_hash":"HY"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"Z","observed_hash":"HZ-OLD"}' >/dev/null
  "$STORE" enqueue --class actionable --locators '{"kind":"repo","id":"Z","observed_hash":"HZ-NEW"}' >/dev/null
  {
    printf '%s\n' '{"observed_seq":1,"locators":{"kind":"repo","id":"X","observed_hash":"HX"},"class":"actionable","emit_ts":1,"hmac":""}'
    printf '%s\n' '{"observed_seq":3,"locators":{"kind":"repo","id":"Z","observed_hash":"HZ-OLD"},"class":"actionable","emit_ts":1,"hmac":""}'
  } >"$dl"
  # Pre-#946-shaped consume: NO quarantined.set exists, so this consume writes
  # the false witness for X@1 exactly as the live defect did.
  "$STORE" consume --upto 4 >/dev/null 2>&1 || fail_msg "T16: baseline consume failed"
  jq_any "$rec" '.id=="X" and .observed_seq==1' || fail_msg "T16: fixture broken — the false X@1 row was not written"
  # REPORT: exactly the X row is provably false; non-zero exit signals findings.
  rep="$TMP_ROOT/t16.rep"
  if "$STORE" quarantine-audit >"$rep" 2>&1; then
    fail_msg "T16: report-mode audit must exit NON-ZERO when false rows exist"
  fi
  has_match -q 'FALSE WITNESS' "$rep" || fail_msg "T16: the audit must name the false row loudly [$(cat "$rep")]"
  has_match -q '"id":"X"' "$rep" || fail_msg "T16: the audit must identify the false row (repo/X@1) [$(cat "$rep")]"
  has_match -q '"id":"Y"' "$rep" && fail_msg "T16: the clean row repo/Y must NOT be flagged"
  has_match -q '"id":"Z"' "$rep" && fail_msg "T16: the HEALED row repo/Z@4 must NOT be flagged (its dead-letter evidence is seq 3 with a different hash)"
  # Report mode modifies nothing.
  jq_any "$rec" '.id=="X"' || fail_msg "T16: report mode must not modify the record"
  # REPAIR: exactly the false row is removed; the dead-letter LEDGER is history
  # and must never be modified.
  "$STORE" quarantine-audit --repair >"$TMP_ROOT/t16.fix" 2>&1 || fail_msg "T16: --repair must succeed [$(cat "$TMP_ROOT/t16.fix")]"
  jq_any "$rec" '.id=="X"' && fail_msg "T16: --repair must REMOVE the provably-false X row"
  jq_any "$rec" '.id=="Y" and .observed_hash=="HY"' || fail_msg "T16: --repair must keep the clean Y row"
  jq_any "$rec" '.id=="Z" and .observed_hash=="HZ-NEW" and .observed_seq==4' || fail_msg "T16: --repair must keep the healed Z@4 row"
  [ "$(has_match -c . "$dl")" = "2" ] || fail_msg "T16: the dead-letter LEDGER must be untouched by --repair"
  # Clean re-audit: OK, exit 0.
  "$STORE" quarantine-audit >"$TMP_ROOT/t16.ok" 2>&1 || fail_msg "T16: a clean audit must exit 0 [$(cat "$TMP_ROOT/t16.ok")]"
  has_match -qi 'OK' "$TMP_ROOT/t16.ok" || fail_msg "T16: a clean audit must say OK [$(cat "$TMP_ROOT/t16.ok")]"
) && ok

echo "== T17: #952 — the clean-sweep message names BOTH unprovable residual classes; a surviving-but-empty-hash dead-letter row is correctly NOT convicted =="
(
  WAKE_STATE_HOME="$(fresh_state t17)"
  export WAKE_STATE_HOME
  unset WAKE_AGENT
  STATE_DIR="$WAKE_STATE_HOME/default"
  rec="$STATE_DIR/consumed-hashes.jsonl"
  dl="$STATE_DIR/dead-letter.jsonl"
  # Drive the REAL flow so the consumed-hashes row is produced by the actual
  # writer (_record_last_consumed), not hand-built: 12 fillers, then the bench
  # key lands at observed_seq 13 — the same seq as the live specimen below.
  i=1
  while [ "$i" -le 12 ]; do
    "$STORE" enqueue --class actionable --locators "{\"kind\":\"filler\",\"id\":\"f$i\",\"observed_hash\":\"HF$i\"}" >/dev/null || fail_msg "T17: filler enqueue $i failed"
    i=$((i + 1))
  done
  "$STORE" enqueue --class actionable --locators '{"kind":"bench","id":"malformed-locator-test","observed_hash":"H-REAL-CONSUMED"}' >/dev/null
  "$STORE" consume --upto 13 >/dev/null 2>&1 || fail_msg "T17: baseline consume failed"
  jq_any "$rec" '.kind=="bench" and .id=="malformed-locator-test" and .observed_seq==13 and .observed_hash=="H-REAL-CONSUMED"' ||
    fail_msg "T17: fixture broken — the real writer did not record the bench@13 row"
  # VERBATIM live specimen: mos-dt lane dead-letter.jsonl line 1 (2026-07-26).
  # Copied byte-for-byte per the #952 fixture constraint — nested .locators.*
  # with NO observed_hash key, so the audit's extraction yields "".
  printf '%s\n' '{"observed_seq":13,"locators":{"kind":"bench","id":"malformed-locator-test","note":"deliberately non-conformant"},"class":"actionable","emit_ts":1785055610,"hmac":""}' >"$dl"
  # Guard the near-miss trap: the specimen must be genuinely NESTED (no
  # top-level kind) and must extract an EMPTY hash via the audit's own paths.
  jq -e '(has("kind") | not) and .locators.kind=="bench" and ((.locators.observed_hash // "") == "")' "$dl" >/dev/null ||
    fail_msg "T17: specimen fixture is not the genuine nested shape — rebuild it from the live lane, not the schema"
  # Evidence SURVIVES (kind/id/seq all match the row) but extracts hash "";
  # _record_last_consumed never writes an empty-hash row, so the four-field
  # match can never fire: this is unprovable class 2, NOT a false witness.
  rep="$TMP_ROOT/t17.rep"
  "$STORE" quarantine-audit >"$rep" 2>&1 || fail_msg "T17: the audit must exit 0 — nothing here is provable [$(cat "$rep")]"
  has_match -q 'FALSE WITNESS' "$rep" && fail_msg "T17: the empty-hash evidence must NOT convict (the predicate is correct and must not change) [$(cat "$rep")]"
  # The wording under test (#952): BOTH residual classes, named.
  has_match -qi 'pruned' "$rep" || fail_msg "T17: clean sweep must name residual class 1 — evidence pruned away [$(cat "$rep")]"
  has_match -qi 'empty observed_hash' "$rep" || fail_msg "T17: clean sweep must name residual class 2 — surviving evidence with an empty observed_hash [$(cat "$rep")]"
  # --repair on a clean sweep removes nothing: the unprovable row survives.
  "$STORE" quarantine-audit --repair >"$TMP_ROOT/t17.fix" 2>&1 || fail_msg "T17: --repair on a clean sweep must exit 0 [$(cat "$TMP_ROOT/t17.fix")]"
  jq_any "$rec" '.kind=="bench" and .observed_seq==13 and .observed_hash=="H-REAL-CONSUMED"' ||
    fail_msg "T17: --repair must NOT remove the unprovable bench@13 row — the audit only removes what the ledger can convict"
  [ "$(count_lines . "$dl")" = "1" ] || fail_msg "T17: the dead-letter LEDGER must be untouched"
) && ok

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