fix(wake): #932 stop reconciler re-enumerating already-CONSUMED detector state (#935)

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>

packages/mosaic/framework/tools/wake/_wake-common.sh

@@ -36,6 +36,12 @@ wake_state_dir() {
 #                    {observed_seq, locators, class, emit_ts, hmac}.
 #   ack-ledger.jsonl — local-write-only ack ledger (RECEIVED / CONSUMED).
 #   ack-sync.state   — background-sync bookkeeping (last shipped / outage flag).
+#   consumed-hashes.jsonl — #932 store-owned last-consumed record: one object per
+#                    (kind,id), {kind,id,observed_hash,observed_seq}, written at
+#                    consume-truncation. The reconciler's THIRD accounting source
+#                    (a consumed state matches neither the truncated inbox nor the
+#                    reconciler's own seen-ledger). ADDITIVE / lazily created;
+#                    older code ignores it (on-disk read-compat preserved).
 
 # _wake_tmp_glob DIR — the glob used for atomic-write temp files, so readers can
 # ignore in-flight/crashed writes. A crash leaves one of these; it is NEVER the

packages/mosaic/framework/tools/wake/manifest.txt

@@ -195,8 +195,35 @@
 #              tolerates on its observed.set-failure path. ON-DISK FORMAT UNCHANGED
 #              (read-compatible; a store written by older code reads identically). Only
 #              store.sh (+ test-wake-store-ack.sh T11) changed.
+#   0.6.10 #932 reconciler RE-ENUMERATION of already-CONSUMED detector-observed
+#              state (wake-pilot finding #7 — safe-but-noisy G2a alarm-hygiene).
+#              After consume-truncation a consumed state matched NO accounting
+#              record (inbox truncated; the reconciler's seen-ledger only covers
+#              its OWN enumerations; the detector hash-file is correctly
+#              DISTRUSTED) -> the reconciler treated it as UNACCOUNTED and
+#              re-enumerated it: one DUPLICATE orientation wake + one SPURIOUS
+#              rc=1 CRITICAL per detector-active window per cycle (functionally
+#              safe — no lost obligation — but cry-wolf erosion of real alarms at
+#              fleet scale). FIX: (1) store.sh records the last-consumed
+#              observed_hash per (kind,id) at consume-truncation into a NEW
+#              store-owned durable record consumed-hashes.jsonl (atomic write;
+#              ADDITIVE — existing on-disk format unchanged/read-compatible; #908
+#              allocator untouched); (2) reconcile.sh adds a THIRD accounting
+#              source alongside the inbox and its seen-ledger: a detector-observed
+#              state whose observed_hash MATCHES the store's recorded last-consumed
+#              hash is ACCOUNTED (not re-enumerated — no dup wake, no spurious
+#              CRITICAL). TRUST BOUNDARY: the 3rd check consults ONLY the
+#              store-written record (its existence implies the state was durably
+#              enqueued+consumed, so it structurally cannot exhibit the §5
+#              hash-advance-without-enqueue swallow signature); trusting DETECTOR
+#              hash-files STAYS REJECTED. G3 is NOT weakened: only states the store
+#              RECORDED as consumed are suppressed — a genuinely-unaccounted state
+#              (enqueued-but-unconsumed, still in the inbox, OR a real gap) still
+#              re-enumerates + alarms. Changed: store.sh, reconcile.sh,
+#              _wake-common.sh (doc), test-wake-reconcile.sh (R10/R11),
+#              test-wake-store-ack.sh (T12).
 component=wake
-version=0.6.9
+version=0.6.10
 
 # Watch-list schema this component consumes, and the INCLUSIVE range of
 # schema_version values it supports. A wake-watch-list.json whose schema_version

packages/mosaic/framework/tools/wake/reconcile.sh

@@ -47,10 +47,14 @@
 #     RETIRED; --allow-enumerate / WAKE_RECONCILE_ALLOW_ENUMERATE remain accepted
 #     as deprecated no-ops for caller compatibility. The G3 accounting invariants
 #     (0-UNACCOUNTED, vacuous-pass prevention) are unaffected and still enforced.
-#   * "accounted" is judged against observed_seq/inbox (§4/G3 wording) PLUS the
-#     reconciler's OWN durable reconciled-state ledger — NOT the detector's
-#     hash-file. Judging by the detector baseline would let a silent first-seen
-#     baseline swallow a pre-existing obligation (the exact hole G3 closes).
+#   * "accounted" is judged against THREE sources: observed_seq/inbox (§4/G3
+#     wording), the reconciler's OWN durable reconciled-state ledger, and (#932)
+#     the STORE's last-consumed record (consumed-hashes.jsonl, store-written at
+#     consume-truncation) — NOT the detector's hash-file. Judging by the detector
+#     baseline would let a silent first-seen baseline swallow a pre-existing
+#     obligation (the exact hole G3 closes). The 3rd source suppresses ONLY states
+#     the store RECORDED as consumed (structurally cannot be hash-advance-without-
+#     enqueue), so a truly-unaccounted state still re-enumerates (G3 teeth intact).
 #   * enumeration uses the non-coalescible fail-safe class `actionable` (§2.3)
 #     so two distinct pre-existing obligations can never coalesce into one.
 #
@@ -367,6 +371,35 @@ _inbox_has() {
   [ "${n:-0}" -gt 0 ]
 }
 
+# _store_consumed_has KIND ID HASH — #932 THIRD accounting source. True iff the
+# STORE's last-consumed record (consumed-hashes.jsonl, written by store.sh at
+# consume-truncation) records HASH as the last-consumed observed_hash for
+# (kind,id). After consume truncates the pending prefix, a consumed state matches
+# NEITHER _inbox_has (inbox truncated) NOR the reconciler's own seen-ledger (it
+# only covers the reconciler's OWN enumerations) — so without this check the
+# reconciler re-enumerates the just-consumed state: one DUPLICATE orientation
+# wake + one SPURIOUS rc=1 CRITICAL per detector-active window per cycle (#932).
+#
+# TRUST BOUNDARY (load-bearing — do NOT violate): this consults ONLY the
+# store-written record under STATE_DIR — NEVER a detector-owned hash-file. The
+# store record is trustworthy BY CONSTRUCTION: it is written only at CONSUME of a
+# durably-enqueued entry, so its existence implies the state WAS durably enqueued,
+# and it structurally cannot exhibit the §5 swallow-hole signature (hash-advance-
+# WITHOUT-enqueue). Trusting DETECTOR hash-files STAYS REJECTED (it reopens §5).
+#
+# G3 is NOT weakened: this suppresses ONLY states the store has RECORDED as
+# consumed. A genuinely-unaccounted state (durably enqueued but not yet consumed
+# — still in the inbox — OR a real gap) has no matching consumed record, so it is
+# still UNACCOUNTED and still re-enumerates + alarms.
+_store_consumed_has() {
+  local kind="$1" id="$2" h="$3" n
+  [ -f "$STATE_DIR/consumed-hashes.jsonl" ] || return 1
+  n="$(jq -s --arg k "$kind" --arg id "$id" --arg h "$h" \
+    '[ .[] | select(.kind==$k and .id==$id and .observed_hash==$h) ] | length' \
+    "$STATE_DIR/consumed-hashes.jsonl" 2>/dev/null || echo 0)"
+  [ "${n:-0}" -gt 0 ]
+}
+
 cmd_reconcile() {
   local enumerate=1 allow_enumerate=0
   while [ $# -gt 0 ]; do
@@ -444,6 +477,12 @@ cmd_reconcile() {
     # Or already reconciled to this exact state before (durable ledger)?
     elif [ -f "$ledger" ] && [ "$(tr -d '[:space:]' <"$ledger")" = "$curhash" ]; then
       accounted=1
+    # Or the STORE recorded this exact state as ALREADY CONSUMED (#932 — the THIRD
+    # accounting source; a consume-truncated state matches neither the inbox nor
+    # the seen-ledger, so without this it would spuriously re-enumerate). Consults
+    # ONLY the store-written record, never a detector hash-file (§5 stays closed).
+    elif _store_consumed_has "$kind" "$id" "$curhash"; then
+      accounted=1
     fi
 
     if [ "$accounted" -eq 1 ]; then

packages/mosaic/framework/tools/wake/store.sh

@@ -326,6 +326,51 @@ cmd_drain() {
   grep -v '^[[:space:]]*$' "$STATE_DIR/pending.jsonl" 2>/dev/null || true
 }
 
+# _record_last_consumed UPTO — #932: at consume-truncation, record the
+# last-consumed observed_hash per (kind,id) into a STORE-OWNED durable record
+# (consumed-hashes.jsonl). This is the record the reconciler consults as its
+# THIRD accounting source: after consume truncates the pending prefix, a consumed
+# state matches NEITHER the inbox (truncated) NOR the reconciler's seen-ledger
+# (which only covers the reconciler's OWN enumerations), so the reconciler used
+# to treat it as UNACCOUNTED and re-enumerate it — one duplicate orientation wake
+# + one spurious rc=1 CRITICAL per detector-active window per cycle (safe but
+# noisy; G2a alarm-hygiene at fleet scale).
+#
+# Called BEFORE the pending prefix is dropped (it reads the entries being
+# consumed). For each entry with observed_seq <= UPTO that carries full source
+# locators (kind,id,observed_hash), keep the HIGHEST-seq hash per (kind,id) as
+# that source's last-consumed state; merge with any prior record (monotonic —
+# consume only advances, so a newer consume's seq always wins its key).
+#
+# TRUST BOUNDARY (load-bearing): this record is written ONLY here, at CONSUME of a
+# durably-enqueued entry, so its existence implies the state WAS durably
+# enqueued+consumed. It therefore structurally cannot exhibit the §5 swallow-hole
+# signature (hash-advance-WITHOUT-enqueue). It is a STORE record — the detector's
+# own hash-file stays DISTRUSTED and is never consulted for accounting.
+#
+# ADDITIVE: a new file consumed-hashes.jsonl; the existing on-disk format is
+# unchanged and read-compatible (older code ignores this file). Atomic write.
+# Best-effort: a failure here degrades to the pre-#932 safe-but-noisy behaviour
+# (the reconciler re-enumerates the consumed state once — no lost obligation),
+# never a failed consume.
+_record_last_consumed() {
+  local upto="$1" existing new_records merged
+  [ -f "$STATE_DIR/pending.jsonl" ] || return 0
+  new_records="$(jq -c --argjson upto "$upto" '
+    select((.observed_seq // -1) <= $upto)
+    | select((.locators.kind // "") != "" and (.locators.id // "") != "" and (.locators.observed_hash // "") != "")
+    | {kind:.locators.kind, id:.locators.id, observed_hash:.locators.observed_hash, observed_seq:.observed_seq}
+  ' "$STATE_DIR/pending.jsonl" 2>/dev/null || true)"
+  [ -n "$new_records" ] || return 0
+  existing="$(cat "$STATE_DIR/consumed-hashes.jsonl" 2>/dev/null || true)"
+  merged="$(printf '%s\n%s\n' "$existing" "$new_records" | grep -v '^[[:space:]]*$' |
+    jq -s -c 'group_by([.kind, .id]) | .[] | max_by(.observed_seq)' 2>/dev/null || true)"
+  [ -n "$merged" ] || return 0
+  if ! printf '%s\n' "$merged" | grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/consumed-hashes.jsonl"; then
+    echo "store.sh consume: WARN — could not record last-consumed hashes (#932); the reconciler may re-enumerate the consumed state once (safe-but-noisy), no lost obligation." >&2
+  fi
+}
+
 cmd_consume() {
   local upto=''
   while [ $# -gt 0 ]; do
@@ -381,6 +426,12 @@ cmd_consume() {
     k=$((k + 1))
   done
 
+  # #932: record the last-consumed observed_hash per (kind,id) BEFORE the pending
+  # prefix is dropped (this reads the entries about to be truncated), so the
+  # reconciler can recognise an already-consumed state as ACCOUNTED instead of
+  # re-enumerating it. Additive, store-owned, best-effort (never fails consume).
+  _record_last_consumed "$upto"
+
   # Advance the consumer cursor and drop the now-consumed prefix from the
   # durable store (retain-until-CONSUMED is satisfied).
   jq -c "select(.observed_seq > $upto)" "$STATE_DIR/pending.jsonl" 2>/dev/null |

packages/mosaic/framework/tools/wake/test-wake-reconcile.sh

@@ -19,6 +19,10 @@
 #      observed_seq allocator, so co-feeding one store yields DISTINCT,
 #      contiguous seqs (no aliasing). Formerly the fail-closed dual-
 #      allocator refusal; #908 dissolves the hazard at the root.        (§4/G3, #908)
+#   R10 a CONSUMED detector state is ACCOUNTED via the store's last-consumed
+#      record — NOT re-enumerated (no dup wake, no spurious rc=1 CRITICAL). (#932)
+#   R11 pilot repro: consumed state SUPPRESSED while a distinct unconsumed/gap
+#      state STILL re-enumerates + flags (G3 teeth intact; no over-suppression). (#932)
 #
 # Uses FAKE/STUB sources only (no live network). Isolated per test.
 # shellcheck disable=SC2030,SC2031
@@ -295,6 +299,88 @@ EOF
   "$STORE" consume --upto 3 >/dev/null 2>&1 || fail_msg "R9: CONSUMED 3 must succeed over the gapless co-fed prefix"
 ) && ok
 
+echo "== R10: #932 — a CONSUMED detector state is ACCOUNTED via the store's last-consumed record (no dup wake, no spurious rc=1 CRITICAL) =="
+(
+  WAKE_STATE_HOME="$(fresh_state r10)"
+  export WAKE_STATE_HOME
+  unset WAKE_AGENT WAKE_RECONCILE_ALLOW_ENUMERATE
+  stub="$TMP_ROOT/r10stub"
+  mkdir -p "$stub"
+  make_stub "$stub"
+  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
+  printf 'V0\n' >"$stub/repo_r1"
+  wl="$TMP_ROOT/r10.json"
+  cat >"$wl" <<'EOF'
+{ "schema_version": 1,
+  "repos": [ { "id": "r1" } ],
+  "watches": [ { "lane": "L", "sources": [ { "kind": "repo", "id": "r1" } ] } ] }
+EOF
+  export WAKE_WATCH_LIST="$wl"
+  # Detector baselines (silent) then a real delta -> one pending inbox entry (seq 1).
+  "$DET" poll-once >/dev/null 2>&1 || fail_msg "R10: detector baseline failed"
+  printf 'V1\n' >"$stub/repo_r1"
+  "$DET" poll-once >/dev/null 2>&1 || fail_msg "R10: detector delta failed"
+  [ "$(depth)" = "1" ] || fail_msg "R10: detector delta should leave one pending entry, got $(depth)"
+  # CONSUME it: consumed_seq advances, the pending prefix is truncated. The store
+  # now records r1's last-consumed observed_hash. The source's CURRENT state is
+  # UNCHANGED (still V1) — exactly the pilot's just-consumed byte-match window.
+  "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "R10: CONSUMED 1 must succeed"
+  [ "$(depth)" = "0" ] || fail_msg "R10: consume must truncate the pending prefix (depth 0), got $(depth)"
+  # RE-RECONCILE. On BASE this RE-ENUMERATES the just-consumed state (inbox
+  # truncated + seen-ledger never covered a detector enqueue) -> UNACCOUNTED=1,
+  # rc=1 (spurious CRITICAL), depth 1 (duplicate wake). After #932 the store's
+  # last-consumed record ACCOUNTS it -> clean.
+  out="$("$RECON" reconcile 2>&1)"
+  rc=$?
+  [ "$rc" -eq 0 ] || fail_msg "R10: a consumed state must be ACCOUNTED (exit 0, no spurious rc=1 CRITICAL) [$out]"
+  echo "$out" | grep -qi 'UNACCOUNTED=0' || fail_msg "R10: a consumed state must be 0 unaccounted (no re-enumeration) [$out]"
+  [ "$(depth)" = "0" ] || fail_msg "R10: reconciler must NOT re-enumerate a consumed state (depth still 0 = no duplicate wake), got $(depth)"
+) && ok
+
+echo "== R11: #932 — pilot repro: consumed state SUPPRESSED while a distinct unconsumed/gap state STILL re-enumerates (G3 teeth intact; no blanket suppression) =="
+(
+  WAKE_STATE_HOME="$(fresh_state r11)"
+  export WAKE_STATE_HOME
+  unset WAKE_AGENT WAKE_RECONCILE_ALLOW_ENUMERATE
+  stub="$TMP_ROOT/r11stub"
+  mkdir -p "$stub"
+  make_stub "$stub"
+  export WAKE_DETECTOR_SOURCE_CMD="$stub/adapter.sh"
+  printf 'V0\n' >"$stub/repo_r1"
+  # r2 is a genuine unaccounted GAP: present from the start so the detector's
+  # FIRST-SEEN baseline is SILENT (never enqueues it), leaving it unaccounted and
+  # unconsumed — it has NO last-consumed record, so it MUST still re-enumerate.
+  printf 'GAP-STATE\n' >"$stub/repo_r2"
+  wl="$TMP_ROOT/r11.json"
+  cat >"$wl" <<'EOF'
+{ "schema_version": 1,
+  "repos": [ { "id": "r1" }, { "id": "r2" } ],
+  "watches": [ { "lane": "L", "sources": [ { "kind": "repo", "id": "r1" }, { "kind": "repo", "id": "r2" } ] } ] }
+EOF
+  export WAKE_WATCH_LIST="$wl"
+  # Baseline BOTH sources silently (no enqueue on first-seen). r2 stays at this
+  # state forever -> never enqueued = genuine gap.
+  "$DET" poll-once >/dev/null 2>&1 || fail_msg "R11: detector baseline failed"
+  # r1: a real delta -> enqueued (seq 1) -> CONSUMED (store records its hash).
+  printf 'V1\n' >"$stub/repo_r1"
+  "$DET" poll-once >/dev/null 2>&1 || fail_msg "R11: detector delta (r1) failed"
+  "$STORE" consume --upto 1 >/dev/null 2>&1 || fail_msg "R11: CONSUMED 1 must succeed"
+  [ "$(depth)" = "0" ] || fail_msg "R11: store should be empty before the reconcile, got depth $(depth)"
+  out="$("$RECON" reconcile 2>&1)"
+  rc=$?
+  # r2 is genuinely unaccounted -> the pass STILL FLAGS (G3 teeth intact).
+  [ "$rc" -ne 0 ] || fail_msg "R11: a genuine gap (r2) must still FLAG (non-zero) [$out]"
+  # After #932: ONLY r2 is unaccounted (r1 suppressed by the store record). On
+  # BASE both r1 and r2 re-enumerate (UNACCOUNTED=2) -> this assertion is red-first.
+  echo "$out" | grep -qi 'UNACCOUNTED=1' || fail_msg "R11: exactly ONE source (the gap r2) must be unaccounted; the consumed r1 must be suppressed [$out]"
+  # Prove precisely WHICH state re-enumerated: the gap r2 IS enumerated, the
+  # consumed r1 is NOT. (base re-enumerates r1 too -> the r1-absent assertion is
+  # red-first; the r2-present assertion holds both before and after = no over-suppression.)
+  en(){ "$STORE" drain | jq -s --arg id "$1" '[ .[] | select(.locators.id==$id) ] | length'; }
+  [ "$(en r2)" -ge 1 ] || fail_msg "R11: the genuine gap r2 MUST be re-enumerated into the store (no blanket suppression) [$out]"
+  [ "$(en r1)" -eq 0 ] || fail_msg "R11: the CONSUMED r1 must NOT be re-enumerated (its byte-matched hash is accounted by the store record) [$out]"
+) && ok
+
 echo
 if [ -s "$FAILFILE" ]; then
   echo "wake reconcile harness: FAILED ($(grep -c . "$FAILFILE") assertion(s))" >&2

packages/mosaic/framework/tools/wake/test-wake-store-ack.sh

@@ -17,6 +17,8 @@
 #   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)
 #
 # Isolated: every test runs against a fresh WAKE_STATE_HOME temp dir.
 set -uo pipefail
@@ -575,6 +577,34 @@ echo "== T11: final observed_seq cursor write FAILURE — fail loud + observed.s
   fi
 ) && 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 | grep -q 'consumed_seq=3' || fail_msg "T12: consumed_seq must be 3 after CONSUMED 3"
+) && ok
+
 echo
 if [ -s "$FAILFILE" ]; then
   echo "wake store/ack harness: FAILED ($(grep -c . "$FAILFILE") assertion(s))" >&2