jason.woltje
28f022d9c0
Co-authored-by: jason.woltje <jason@diversecanvas.com> Co-committed-by: jason.woltje <jason@diversecanvas.com>
320 lines
10 KiB
Bash
Executable File
320 lines
10 KiB
Bash
Executable File
#!/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 "$@"
|