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

#!/usr/bin/env bash
# store.sh — A2 of the wake canon (EPIC #892, W2): the THREE-CURSOR durable
# queue (store + drain lib).
#
# CONTRACT ANCHORS (docs/scratchpads/heartbeat-planning/CONVERGED-DESIGN.md):
#   §1.2  three-cursor data-flow: observe -> durable store -> coalesce -> drain.
#   §2.3  classes & durability: ALL classes durable; coalescing is optional.
#   §2.4  cursor semantics: observed_seq is authoritative; SHAs are descriptors.
#
# Three cursors:
#   observed_seq  (detector-owned; this lib RECORDS/stores it) — monotonic int
#                 assigned at observation. Authoritative. Source SHAs are
#                 descriptors, NOT the cursor (§2.4).
#   pending-inbox (durable) — append store; retains until CONSUMED; survives
#                 pane death / park / restart.
#   consumed_seq  (consumer-owned) — advances ONLY on a CONSUMED ack, over a
#                 CONTIGUOUS gapless prefix <=N. NEVER on sender exit-0 / paste.
#
# Coalesce (§1.2/§2.3): class=digest REPLACES the pending cumulative-state entry
# (newest subsumes prior). Non-coalescible classes (actionable, human, ...)
# APPEND. ALL classes are stored durably — durability is NEVER bypassed.
# Absent class => treated as `actionable` (fail-safe).
#
# HMAC: entries carry an `hmac` field left as an UNSIGNED PLACEHOLDER ("").
# Signing is W3/A5, explicitly NOT W2.
#
# Every state mutation is atomic write-tmp+rename (see _wake-common.sh).
#
# Operator-agnostic: state via XDG/env only; no operator paths/names/secrets.
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)"

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

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

Commands:
  init                                   Create/repair the XDG state layout.
  enqueue --seq N --class C [opts]       Record an observation into the durable
                                         pending-inbox (coalesce or append).
      --seq N            observed_seq (detector-assigned monotonic int). Required.
      --class C          digest|actionable|human|terminal-log|reaction.
                         Absent/empty => actionable (fail-safe).
      --locators JSON    JSON value for the entry's locators (default {}).
      --emit-ts TS       epoch seconds (default: now).
  drain [--require-idle-cmd CMD]         Emit the deliverable set (pending-inbox,
                                         coalesced) as JSONL. WHAT to deliver;
                                         the actual paste is out of scope.
                                         If --require-idle-cmd is given and it
                                         exits non-zero, emit nothing (not idle).
  consume --upto N                       Advance consumed_seq over the contiguous
                                         gapless prefix <=N; drop consumed
                                         entries. Rejects a gap (cannot ack N
                                         while N-1 is unconsumed). Cumulative &
                                         idempotent.
  cursors                                Print observed_seq / consumed_seq / depth.

Environment:
  WAKE_STATE_HOME  override base (default ${XDG_STATE_HOME:-$HOME/.local/state}/mosaic/wake)
  WAKE_AGENT       per-agent queue namespace (default: default)
EOF
}

cmd_init() {
  _wake_init_dir "$STATE_DIR" || {
    echo "store.sh: failed to init $STATE_DIR" >&2
    exit 1
  }
  echo "$STATE_DIR"
}

cmd_enqueue() {
  _need_jq
  local seq='' class='' locators='{}' emit_ts=''
  while [ $# -gt 0 ]; do
    case "$1" in
    --seq)
      seq="${2:-}"
      shift 2
      ;;
    --class)
      class="${2:-}"
      shift 2
      ;;
    --locators)
      locators="${2:-}"
      shift 2
      ;;
    --emit-ts)
      emit_ts="${2:-}"
      shift 2
      ;;
    *)
      echo "store.sh enqueue: unknown option '$1'" >&2
      exit 2
      ;;
    esac
  done

  case "$seq" in
  '' | *[!0-9]*)
    echo "store.sh enqueue: --seq must be a non-negative integer" >&2
    exit 2
    ;;
  esac

  # Absent class => actionable (fail-safe, §2.3).
  [ -n "$class" ] || class="actionable"
  case "$class" in
  digest | actionable | human | terminal-log | reaction) : ;;
  *)
    echo "store.sh enqueue: unknown class '$class'" >&2
    exit 2
    ;;
  esac

  [ -n "$emit_ts" ] || emit_ts="$(date +%s)"
  case "$emit_ts" in
  *[!0-9]*)
    echo "store.sh enqueue: --emit-ts must be epoch seconds" >&2
    exit 2
    ;;
  esac

  # Validate locators is well-formed JSON.
  if ! printf '%s' "$locators" | jq -e . >/dev/null 2>&1; then
    echo "store.sh enqueue: --locators must be valid JSON" >&2
    exit 2
  fi

  _wake_init_dir "$STATE_DIR"

  local consumed
  consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"

  # An observation already inside the consumed prefix is idempotently ignored
  # (§1.2: cursor never regresses; a re-observed already-consumed seq is a no-op).
  if [ "$seq" -le "$consumed" ]; then
    return 0
  fi

  # Build the durable entry. `hmac` is an UNSIGNED PLACEHOLDER (W3/A5 signs it).
  local entry
  entry="$(jq -cn \
    --argjson seq "$seq" \
    --arg class "$class" \
    --argjson locators "$locators" \
    --argjson emit_ts "$emit_ts" \
    '{observed_seq:$seq, locators:$locators, class:$class, emit_ts:$emit_ts, hmac:""}')" || {
    echo "store.sh enqueue: failed to build entry" >&2
    exit 1
  }

  # --- durable store write (coalesce vs append) -----------------------------
  # digest: REPLACE the single pending cumulative-state entry (drop any existing
  #         digest lines, then append the newest). Newest subsumes prior.
  # others: APPEND (never replaced).
  # In BOTH cases the entry lands in the durable store.
  local new_pending
  if [ "$class" = "digest" ]; then
    new_pending="$(jq -c 'select(.class != "digest")' "$STATE_DIR/pending.jsonl" 2>/dev/null || true)"
    new_pending="$(printf '%s\n%s\n' "$new_pending" "$entry" | grep -v '^[[:space:]]*$' || true)"
  else
    new_pending="$(cat "$STATE_DIR/pending.jsonl" 2>/dev/null; printf '%s\n' "$entry")"
    new_pending="$(printf '%s' "$new_pending" | grep -v '^[[:space:]]*$' || true)"
  fi
  printf '%s\n' "$new_pending" | grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/pending.jsonl"

  # --- observed.set: record the seq in the live window (gap-detector SoT) ----
  {
    cat "$STATE_DIR/observed.set" 2>/dev/null
    printf '%s\n' "$seq"
  } | grep -v '^[[:space:]]*$' | sort -n | uniq | _atomic_write "$STATE_DIR/observed.set"

  # --- observed_seq cursor: max monotonic int -------------------------------
  local observed
  observed="$(_wake_read_int "$STATE_DIR/observed_seq" 0)"
  if [ "$seq" -gt "$observed" ]; then
    printf '%s' "$seq" | _atomic_write "$STATE_DIR/observed_seq"
  fi
}

cmd_drain() {
  local idle_cmd=''
  while [ $# -gt 0 ]; do
    case "$1" in
    --require-idle-cmd)
      idle_cmd="${2:-}"
      shift 2
      ;;
    *)
      echo "store.sh drain: unknown option '$1'" >&2
      exit 2
      ;;
    esac
  done

  # Drain returns the deliverable set ONLY when the pane is idle-at-prompt. The
  # actual idle detection + paste is out of W2 scope (send-message.sh); the
  # caller supplies an idle predicate. No predicate => emit unconditionally.
  if [ -n "$idle_cmd" ]; then
    if ! sh -c "$idle_cmd" >/dev/null 2>&1; then
      return 0 # not idle-at-prompt: deliver nothing this cycle.
    fi
  fi

  [ -f "$STATE_DIR/pending.jsonl" ] || return 0
  grep -v '^[[:space:]]*$' "$STATE_DIR/pending.jsonl" 2>/dev/null || true
}

cmd_consume() {
  local upto=''
  while [ $# -gt 0 ]; do
    case "$1" in
    --upto)
      upto="${2:-}"
      shift 2
      ;;
    *)
      echo "store.sh consume: unknown option '$1'" >&2
      exit 2
      ;;
    esac
  done
  case "$upto" in
  '' | *[!0-9]*)
    echo "store.sh consume: --upto must be a non-negative integer" >&2
    exit 2
    ;;
  esac

  _need_jq
  _wake_init_dir "$STATE_DIR"

  local consumed observed
  consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"
  observed="$(_wake_read_int "$STATE_DIR/observed_seq" 0)"

  # Cumulative & idempotent: CONSUMED N implies all <=N. N already covered is a
  # no-op success (the sender checks cursor-advanced-past-X, not per-wake).
  if [ "$upto" -le "$consumed" ]; then
    echo "$consumed"
    return 0
  fi

  # Never consume beyond what was observed.
  if [ "$upto" -gt "$observed" ]; then
    echo "store.sh consume: cannot CONSUMED $upto beyond observed_seq $observed" >&2
    exit 1
  fi

  # CONTIGUOUS gapless prefix: every seq in (consumed, upto] must have been
  # observed. A missing interior seq is a GAP => reject (cannot ack N while N-1
  # is unconsumed). observed.set is coalescing-immune, so a subsumed digest seq
  # still counts as observed and does not read as a gap.
  local k
  k=$((consumed + 1))
  while [ "$k" -le "$upto" ]; do
    if ! grep -qxF "$k" "$STATE_DIR/observed.set" 2>/dev/null; then
      echo "store.sh consume: gap at seq $k — cannot CONSUMED $upto while $k is unobserved/unconsumed" >&2
      exit 1
    fi
    k=$((k + 1))
  done

  # 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 |
    grep -v '^[[:space:]]*$' | _atomic_write "$STATE_DIR/pending.jsonl" || true
  awk -v c="$upto" 'NF && $1+0 > c' "$STATE_DIR/observed.set" 2>/dev/null |
    _atomic_write "$STATE_DIR/observed.set" || true
  printf '%s' "$upto" | _atomic_write "$STATE_DIR/consumed_seq"
  echo "$upto"
}

cmd_cursors() {
  _wake_init_dir "$STATE_DIR"
  local observed consumed depth
  observed="$(_wake_read_int "$STATE_DIR/observed_seq" 0)"
  consumed="$(_wake_read_int "$STATE_DIR/consumed_seq" 0)"
  depth="$(grep -cv '^[[:space:]]*$' "$STATE_DIR/pending.jsonl" 2>/dev/null || echo 0)"
  printf 'observed_seq=%s\nconsumed_seq=%s\npending_depth=%s\n' "$observed" "$consumed" "$depth"
}

main() {
  [ $# -ge 1 ] || {
    usage
    exit 2
  }
  local cmd="$1"
  shift
  case "$cmd" in
  init) cmd_init "$@" ;;
  enqueue) cmd_enqueue "$@" ;;
  drain) cmd_drain "$@" ;;
  consume) cmd_consume "$@" ;;
  cursors) cmd_cursors "$@" ;;
  -h | --help | help) usage ;;
  *)
    echo "store.sh: unknown command '$cmd'" >&2
    usage
    exit 2
    ;;
  esac
}

main "$@"