diff --git a/bin/gstack-brain-enqueue b/bin/gstack-brain-enqueue index ffc09c11e..815eff31b 100755 --- a/bin/gstack-brain-enqueue +++ b/bin/gstack-brain-enqueue @@ -1,13 +1,13 @@ #!/usr/bin/env bash -# gstack-brain-enqueue — atomically append a path to the GBrain sync queue. +# gstack-brain-enqueue — write a path record into the GBrain sync spool. # # Usage: # gstack-brain-enqueue # # Called by writer scripts (gstack-learnings-log, gstack-timeline-log, etc.) # after their local write. Fire-and-forget; failures are silent (never blocks -# the writer). Queue is drained by `gstack-brain-sync --once` invoked from the -# preamble at skill START and END boundaries. +# the writer). The spool is drained by `gstack-brain-sync --once` invoked from +# the preamble at skill START and END boundaries. # # No-op when: # - artifacts_sync_mode is off (the default) @@ -18,8 +18,12 @@ # GSTACK_HOME — override ~/.gstack state directory (aligns with writers). # Tests use GSTACK_HOME=/tmp/test-$$ for isolation. # -# Concurrency: POSIX append is atomic up to PIPE_BUF (~4KB Linux, 512 BSD). -# Queue lines are ~200 bytes, safe under concurrent callers. +# Concurrency: maildir-style spool — one FILE per record under +# .brain-queue.d/, created via tmp-file + atomic rename. Writer and drainer +# never share an inode, so there is no append/rewrite race by construction +# (the legacy single-file .brain-queue.jsonl append could race the drain's +# rewrite). Filenames are --.json, so a sorted listing is +# chronological. # No `-e` — writer shims rely on this never failing loudly. set -uo pipefail @@ -28,7 +32,7 @@ FILE="${1:-}" [ -z "$FILE" ] && exit 0 GSTACK_HOME="${GSTACK_HOME:-$HOME/.gstack}" -QUEUE="$GSTACK_HOME/.brain-queue.jsonl" +SPOOL="$GSTACK_HOME/.brain-queue.d" SKIP_FILE="$GSTACK_HOME/.brain-skip.txt" # Fast exits: no git repo, no sync. @@ -50,6 +54,11 @@ fi ESC_FILE=$(printf '%s' "$FILE" | sed 's/\\/\\\\/g; s/"/\\"/g') TS=$(date -u +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo "") -printf '{"file":"%s","ts":"%s"}\n' "$ESC_FILE" "$TS" >> "$QUEUE" 2>/dev/null +# One spool file per record: tmp write + atomic rename. Any failure exits 0 +# silently (fire-and-forget contract), cleaning up the tmp file. +mkdir -p "$SPOOL" 2>/dev/null || exit 0 +TMP="$SPOOL/.tmp-$$-$RANDOM" +printf '{"file":"%s","ts":"%s"}\n' "$ESC_FILE" "$TS" > "$TMP" 2>/dev/null || { rm -f "$TMP" 2>/dev/null; exit 0; } +mv -f "$TMP" "$SPOOL/$(date +%s)-$$-$RANDOM.json" 2>/dev/null || rm -f "$TMP" 2>/dev/null exit 0 diff --git a/bin/gstack-brain-sync b/bin/gstack-brain-sync index 1a9c7b5ff..52b757877 100755 --- a/bin/gstack-brain-sync +++ b/bin/gstack-brain-sync @@ -20,6 +20,13 @@ set -uo pipefail GSTACK_HOME="${GSTACK_HOME:-$HOME/.gstack}" +# Maildir-style spool: one FILE per record, --.json. +# Writers (gstack-brain-enqueue, --discover-new) create records via tmp-file +# + atomic rename; the drain deletes exactly the files it snapshotted. No +# shared inode between writer and drainer → no append/rewrite race. +QUEUE_DIR="$GSTACK_HOME/.brain-queue.d" +# Legacy single-file queue: kept ONLY for migration. Pre-spool writers +# appended lines here; migrate_legacy_queue converts them to spool files. QUEUE="$GSTACK_HOME/.brain-queue.jsonl" ALLOWLIST="$GSTACK_HOME/.brain-allowlist" PRIVACY_MAP="$GSTACK_HOME/.brain-privacy-map.json" @@ -120,7 +127,84 @@ sys.exit(0) " } -# Compute matched allowlisted, privacy-filtered path set from queue. +# True (0) if the spool holds at least one record file. +spool_has_records() { + local f + for f in "$QUEUE_DIR"/*.json; do + [ -e "$f" ] && return 0 + done + return 1 +} + +# Convert one legacy queue file's lines into spool record files (tmp + +# os.replace, one file per line). Reads the file TWICE before unlinking: a +# pre-rename writer can still append through its already-open fd after our +# rename, and those appends land in the renamed file — the second pass +# catches them (the tail race the shared-file design could never close). +# Unparseable lines migrate as-is; finalize_queue keeps + warns on them. +convert_legacy_file() { + local legacy="$1" + python3 - "$legacy" "$QUEUE_DIR" <<'PYEOF' 2>/dev/null || true +import os, sys, time + +legacy, spool = sys.argv[1:3] + +def read_lines(path): + try: + with open(path) as f: + return [l.rstrip("\r\n") for l in f if l.strip()] + except (FileNotFoundError, OSError): + return [] + +seq = 0 +def write_spool(line): + global seq + seq += 1 + tmp = os.path.join(spool, f".tmp-{os.getpid()}-m{seq}") + with open(tmp, "w") as f: + f.write(line + "\n") + os.replace(tmp, os.path.join(spool, f"{int(time.time())}-{os.getpid()}-m{seq}.json")) + +written = set() +for _pass in (1, 2): # second read closes the pre-rename-fd tail race + for line in read_lines(legacy): + if line not in written: # identical duplicates collapse, as the old rewrite did + write_spool(line) + written.add(line) +os.unlink(legacy) +PYEOF +} + +# Legacy migration (transition window only). If the single-file queue holds +# records, atomically rename it aside and convert each line to a spool file. +# A concurrent OLD writer that recreates a fresh legacy file after the rename +# simply gets migrated on the NEXT drain — nothing is lost, only deferred one +# boundary. Runs inside the run lock, before the drain reads the spool. +migrate_legacy_queue() { + local migrating="$QUEUE.migrating" + # Crash leftover: a prior migration renamed but died before unlink. Some of + # its lines may already exist as spool files — re-converting duplicates is + # safe (at-least-once; the drain dedups paths per snapshot and downstream + # content-hash dedup absorbs re-syncs). Losing the file would not be. If + # the conversion itself fails, the file stays for the next run (never rm a + # non-empty .migrating file outside convert_legacy_file's own unlink). + if [ -f "$migrating" ]; then + if [ -s "$migrating" ]; then + mkdir -p "$QUEUE_DIR" 2>/dev/null || return 0 + convert_legacy_file "$migrating" + else + rm -f "$migrating" 2>/dev/null || true + fi + fi + if [ -s "$QUEUE" ]; then + mkdir -p "$QUEUE_DIR" 2>/dev/null || return 0 + mv -f "$QUEUE" "$migrating" 2>/dev/null || return 0 + convert_legacy_file "$migrating" + fi + return 0 +} + +# Compute matched allowlisted, privacy-filtered path set from the spool. # Output: newline-delimited relative paths that should be staged. # # #2549: every non-staged queue entry is CLASSIFIED, never silently discarded. @@ -132,13 +216,20 @@ sys.exit(0) # attempt, no allowlist glob, not on disk) and are removed WITH a counted # status — the old behavior truncated the whole queue and reported every one # of these, including privacy holds, as "no allowlisted changes". +# +# Spool snapshot ($3): the sorted list of spool record filenames read here is +# written to the snapshot manifest, one filename per line. finalize_queue +# deletes exactly the manifest's files and never touches records created +# after this listing — a concurrent enqueue is a separate file by +# construction, so it simply rides to the next drain. compute_paths_to_stage() { local mode="$1" local class_file="${2:-}" - python3 - "$GSTACK_HOME" "$QUEUE" "$ALLOWLIST" "$PRIVACY_MAP" "$SKIP_FILE" "$mode" "$class_file" <<'PYEOF' + local snapshot_file="${3:-}" + python3 - "$GSTACK_HOME" "$QUEUE_DIR" "$ALLOWLIST" "$PRIVACY_MAP" "$SKIP_FILE" "$mode" "$class_file" "$snapshot_file" <<'PYEOF' import sys, json, os, fnmatch, glob -gstack_home, queue, allowlist_path, privacy_path, skip_path, mode, class_file = sys.argv[1:8] +gstack_home, spool_dir, allowlist_path, privacy_path, skip_path, mode, class_file, snapshot_file = sys.argv[1:9] def load_lines(path): try: @@ -164,23 +255,38 @@ privacy_map = load_privacy_map(privacy_path) # discover_new — otherwise an explicitly-skipped file gets committed. skip_lines = {s.replace(os.sep, "/") for s in load_lines(skip_path)} -# Read queue; collect unique file paths. -queue_paths = set() +# Snapshot the spool: sorted (= chronological, filenames are epoch-first) +# list of record files at read time. Records that appear after this listing +# belong to the NEXT drain. Files we cannot read stay OUT of the manifest so +# finalize never deletes a record this drain didn't actually consume. try: - with open(queue) as f: - for line in f: - line = line.strip() - if not line: - continue - try: - obj = json.loads(line) - p = obj.get("file") - if isinstance(p, str): - queue_paths.add(p) - except json.JSONDecodeError: - continue -except FileNotFoundError: - pass + snapshot = sorted(n for n in os.listdir(spool_dir) if n.endswith(".json")) +except (FileNotFoundError, NotADirectoryError): + snapshot = [] + +queue_paths = set() +consumed = [] +for name in snapshot: + try: + with open(os.path.join(spool_dir, name)) as f: + line = f.readline().strip() + except OSError: + continue + consumed.append(name) + if not line: + continue + try: + obj = json.loads(line) + p = obj.get("file") + if isinstance(p, str): + queue_paths.add(p) + except json.JSONDecodeError: + continue # unparseable record: finalize keeps + warns + +if snapshot_file: + with open(snapshot_file, "w") as f: + for name in consumed: + f.write(name + "\n") def path_matches_any(path, globs): for pattern in globs: @@ -241,22 +347,26 @@ for p in final: PYEOF } -# #2549: surgical queue rewrite — replaces every whole-queue truncation -# (`: > "$QUEUE"`). Re-reads the LIVE queue at rewrite time (a writer may have -# enqueued while we were staging/pushing — those entries must survive; the old -# truncation destroyed them) and keeps every line whose file is either -# retained (privacy/mode-held) or not part of this drain at all. Atomic -# tmp+mv in the same directory. Dropped-path detail goes to a 0600 sidecar so -# the status line can stay content-free (counts only). -rewrite_queue() { - local paths_file="$1" # staged (drained) paths, one per line - local class_file="$2" # classification JSON from compute_paths_to_stage - # Fail-open by design (a failed rewrite self-corrects next run: re-stage → +# Finalize the drain: delete exactly the spool record files this drain +# consumed (per the snapshot manifest), keeping retained (privacy/mode-held) +# and unparseable records queued. The predecessor (a shared-file queue +# rewrite) had a lockless-append race between its live re-read and the +# os.replace; with one file per record that race class is structurally gone — +# a concurrent enqueue is a separate file the snapshot never listed, so +# finalize cannot touch it. Crash semantics are at-least-once: a drain that +# dies before finalize leaves its spool files in place and the next run +# re-drains them; downstream content-hash dedup absorbs the duplicates. +# Dropped-path detail goes to a 0600 sidecar so the status line can stay +# content-free (counts only). +finalize_queue() { + local snapshot_file="$1" # spool filenames this drain consumed, one per line + local class_file="$2" # classification JSON from compute_paths_to_stage + # Fail-open by design (a failed finalize self-corrects next run: re-stage → # nothing-to-commit), but say so — a silent failure here would let the # subsequent "ok/idle" status claim a drain that did not happen. - python3 - "$QUEUE" "$paths_file" "$class_file" "$GSTACK_HOME/.brain-sync-drops.json" <<'PYEOF' || echo "BRAIN_SYNC: warning: queue rewrite failed — entries retained; next run re-drains" >&2 + python3 - "$QUEUE_DIR" "$snapshot_file" "$class_file" "$GSTACK_HOME/.brain-sync-drops.json" <<'PYEOF' || echo "BRAIN_SYNC: warning: queue finalize failed — entries retained; next run re-drains" >&2 import json, os, sys, time -queue, paths_file, class_file, drops_file = sys.argv[1:5] +spool_dir, snapshot_file, class_file, drops_file = sys.argv[1:5] def lines(path): try: @@ -265,7 +375,6 @@ def lines(path): except FileNotFoundError: return [] -staged = set(lines(paths_file)) try: with open(class_file) as f: classified = json.load(f) @@ -275,36 +384,31 @@ retained = set(classified.get("retained", [])) dropped = set() for group in (classified.get("dropped", {}) or {}).values(): dropped.update(group) -processed = staged | dropped -kept = [] -seen_lines = set() unparseable = 0 -# LIVE re-read narrows (not fully closes) the concurrent-append window: the -# lockless enqueue can still land on the old inode between this read and the -# os.replace below. Vastly better than the old whole-queue truncation. -for line in lines(queue): - if line in seen_lines: - continue # identical duplicate lines collapse on rewrite +for name in lines(snapshot_file): + full = os.path.join(spool_dir, name) try: - p = json.loads(line).get("file") + with open(full) as f: + rec = f.readline().strip() + except OSError: + continue # unreadable now: leave it for the next drain + p = None + try: + p = json.loads(rec).get("file") except Exception: - unparseable += 1 - kept.append(line) # unparseable line: keep, never destroy - seen_lines.add(line) + pass + if not isinstance(p, str): + unparseable += 1 # keep — never destroy what we can't read continue - if not isinstance(p, str) or p in retained or p not in processed: - kept.append(line) - seen_lines.add(line) + if p in retained: + continue # stays queued: syncs under a higher mode + try: + os.unlink(full) # staged or dropped: fully processed + except FileNotFoundError: + pass if unparseable: - import sys as _sys - print(f"BRAIN_SYNC: {unparseable} unparseable queue line(s) held (inspect {queue})", file=_sys.stderr) - -tmp = queue + ".tmp." + str(os.getpid()) -with open(tmp, "w") as f: - for l in kept: - f.write(l + "\n") -os.replace(tmp, queue) + print(f"BRAIN_SYNC: {unparseable} unparseable spool record(s) held (inspect {spool_dir})", file=sys.stderr) if dropped: fd = os.open(drops_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) @@ -374,6 +478,10 @@ subcmd_once() { # the lock removal. trap 'rm -rf "$lock_dir" 2>/dev/null || true' EXIT INT TERM + # Convert any legacy single-file queue lines into spool records before the + # drain reads the spool (transition window for pre-spool writers). + migrate_legacy_queue + local mode mode=$("$CONFIG_BIN" get artifacts_sync_mode 2>/dev/null || echo off) @@ -449,27 +557,31 @@ subcmd_once() { fi # Empty-queue fast path: this is the steady state at every skill boundary. - # Skipping compute/rewrite here is safe — with zero queue lines there is - # nothing to classify, retain, or drop, and a concurrent append after this - # check simply waits for the next boundary. (The detector above already ran: - # its whole point is re-pushing stranded commits when the queue is empty.) - # The lock-release trap installed at acquisition covers this exit. - if [ ! -s "$QUEUE" ]; then + # Skipping compute/finalize here is safe — with zero spool records there is + # nothing to classify, retain, or drop, and a record created after this + # check simply waits for the next boundary. The legacy file is checked too: + # an OLD writer may have recreated it after the migration above (it gets + # migrated next run, but the depth is honest now). (The detector above + # already ran: its whole point is re-pushing stranded commits when the + # queue is empty.) The lock-release trap installed at acquisition covers + # this exit. + if ! spool_has_records && [ ! -s "$QUEUE" ]; then write_status "idle" "queue empty" exit 0 fi - local paths_file class_file + local paths_file class_file snapshot_file paths_file=$(mktemp /tmp/brain-sync-paths.XXXXXX) || { rm -rf "$lock_dir" 2>/dev/null; write_status "error" "mktemp failed"; exit 1; } class_file=$(mktemp /tmp/brain-sync-class.XXXXXX) || { rm -f "$paths_file"; rm -rf "$lock_dir" 2>/dev/null; write_status "error" "mktemp failed"; exit 1; } + snapshot_file=$(mktemp /tmp/brain-sync-snapshot.XXXXXX) || { rm -f "$paths_file" "$class_file"; rm -rf "$lock_dir" 2>/dev/null; write_status "error" "mktemp failed"; exit 1; } # Single trap covers all: lock cleanup AND tempfile cleanup. - trap 'rm -f "$paths_file" "$class_file" 2>/dev/null; rm -rf "$lock_dir" 2>/dev/null || true' EXIT INT TERM + trap 'rm -f "$paths_file" "$class_file" "$snapshot_file" 2>/dev/null; rm -rf "$lock_dir" 2>/dev/null || true' EXIT INT TERM - compute_paths_to_stage "$mode" "$class_file" > "$paths_file" + compute_paths_to_stage "$mode" "$class_file" "$snapshot_file" > "$paths_file" if [ ! -s "$paths_file" ]; then - # Nothing stageable. Rewrite the queue (retained entries + concurrent - # appends survive; classified drops removed) instead of truncating it. - rewrite_queue "$paths_file" "$class_file" + # Nothing stageable. Finalize the snapshot (retained entries survive; + # classified drops removed; records created after the snapshot untouched). + finalize_queue "$snapshot_file" "$class_file" local summary summary=$(queue_summary "$class_file") write_status "idle" "no stageable changes${summary:+ ($summary)}" @@ -520,8 +632,8 @@ subcmd_once() { git -C "$GSTACK_HOME" -c user.email="gstack@localhost" -c user.name="gstack-brain-sync" \ commit -q -m "$msg" 2>/dev/null || { # Nothing to commit (e.g. all files already committed). The drained - # paths leave the queue; retained + concurrent entries survive (#2549). - rewrite_queue "$paths_file" "$class_file" + # records leave the spool; retained + post-snapshot records survive. + finalize_queue "$snapshot_file" "$class_file" write_status "idle" "queue drained but no new changes to commit" exit 0 } @@ -535,10 +647,10 @@ subcmd_once() { hint=$(remote_auth_hint) write_status "push_failed" "push failed: auth error; commit retained locally, will retry next run. fix: $hint" echo "BRAIN_SYNC: push failed: auth. fix: $hint" >&2 - # Drained paths leave the queue — they live in the local commit, which + # Drained records leave the spool — they live in the local commit, which # the run-start detector re-pushes next time (#2549). Retained + - # concurrent entries survive the rewrite. - rewrite_queue "$paths_file" "$class_file" + # post-snapshot records survive the finalize. + finalize_queue "$snapshot_file" "$class_file" exit 0 fi @@ -552,7 +664,7 @@ subcmd_once() { if git -C "$GSTACK_HOME" merge --no-edit "origin/$branch" >/dev/null 2>&1; then if GSTACK_HOME="$GSTACK_HOME" _receipted_git closed brain-sync "$push_host" curated-memory-git-push "artifacts_sync_mode!=off" \ bash -c 'git -C "$1" push origin HEAD 2>/dev/null' _ "$GSTACK_HOME"; then - rewrite_queue "$paths_file" "$class_file" + finalize_queue "$snapshot_file" "$class_file" date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE" write_status "ok" "pushed $n file(s) after rebase" exit 0 @@ -561,12 +673,12 @@ subcmd_once() { fi # Commit exists locally; the run-start detector re-pushes it next time. write_status "push_failed" "push failed: $(printf '%s' "$push_err" | head -1); commit retained locally, will retry next run" - rewrite_queue "$paths_file" "$class_file" + finalize_queue "$snapshot_file" "$class_file" exit 0 } - # Success: drained paths leave the queue (retained + concurrent survive). - rewrite_queue "$paths_file" "$class_file" + # Success: drained records leave the spool (retained + post-snapshot survive). + finalize_queue "$snapshot_file" "$class_file" date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE" write_status "ok" "pushed $n file(s)" exit 0 @@ -578,9 +690,13 @@ subcmd_status() { else echo '{"status":"unknown","message":"no status file yet"}' fi - # Supplemental info (not in status file). - local queue_depth=0 - [ -f "$QUEUE" ] && queue_depth=$(wc -l < "$QUEUE" | tr -d ' ') + # Supplemental info (not in status file). Depth = spool record files plus + # any not-yet-migrated legacy queue lines (transition window). + local queue_depth spool_depth legacy_depth + spool_depth=$(ls "$QUEUE_DIR"/*.json 2>/dev/null | wc -l | tr -d ' ') + legacy_depth=0 + [ -f "$QUEUE" ] && legacy_depth=$(wc -l < "$QUEUE" | tr -d ' ') + queue_depth=$(( spool_depth + legacy_depth )) local last_push="never" [ -f "$LAST_PUSH_FILE" ] && last_push=$(cat "$LAST_PUSH_FILE" 2>/dev/null || echo never) local mode @@ -611,13 +727,22 @@ subcmd_drop_queue() { echo "Refusing: --drop-queue discards pending syncs. Pass --yes to confirm." >&2 exit 1 fi - if [ ! -f "$QUEUE" ]; then + # Remove spool record files, then truncate any legacy queue remnant. + local n=0 f + for f in "$QUEUE_DIR"/*.json; do + [ -e "$f" ] || continue + rm -f "$f" 2>/dev/null && n=$(( n + 1 )) + done + if [ -f "$QUEUE" ]; then + local legacy_n + legacy_n=$(wc -l < "$QUEUE" | tr -d ' ') + n=$(( n + legacy_n )) + : > "$QUEUE" + fi + if [ "$n" -eq 0 ]; then echo "queue already empty" exit 0 fi - local n - n=$(wc -l < "$QUEUE" | tr -d ' ') - : > "$QUEUE" echo "dropped $n queue entries" } @@ -627,11 +752,11 @@ subcmd_discover_new() { fi # Walk allowlist globs; enqueue any file where mtime+size differs from cursor. python3 - "$GSTACK_HOME" "$ALLOWLIST" "$DISCOVER_CURSOR" <<'PYEOF' 2>/dev/null || true -import sys, os, json, fnmatch +import sys, os, json, fnmatch, time from datetime import datetime, timezone gstack_home, allowlist_path, cursor_path = sys.argv[1:4] -queue_path = os.path.join(gstack_home, ".brain-queue.jsonl") +spool_dir = os.path.join(gstack_home, ".brain-queue.d") skip_path = os.path.join(gstack_home, ".brain-skip.txt") def load_lines(path): @@ -689,34 +814,36 @@ for root, dirs, files in os.walk(gstack_home): if cursor.get(rel) != key: to_enqueue.append((rel, key)) -# Append to the queue directly. The previous implementation shelled out to +# Write spool records directly. The previous implementation shelled out to # gstack-brain-enqueue once per file, but Windows Python cannot exec a # bash-shebang script (the spawn fails with a fork error), so discovery # enqueued nothing on Windows even after the path-match fix above. -# Writing the queue line here is platform-agnostic; the drain step +# Writing the record here is platform-agnostic; the drain step # (compute_paths_to_stage) still re-applies the skip-list + privacy filters. if to_enqueue: ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + written = [] try: - # One atomic append per record (O_APPEND, each line < PIPE_BUF), matching - # gstack-brain-enqueue's concurrency contract so a writer-shim append - # running in parallel can't interleave mid-record. Buffered text writes - # don't guarantee that. Compact separators match the shim's JSON shape. - fd = os.open(queue_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644) - try: - for rel, key in to_enqueue: - rec = json.dumps({"file": rel, "ts": ts}, separators=(",", ":")) - os.write(fd, (rec + "\n").encode("utf-8")) - finally: - os.close(fd) + # One spool FILE per record (tmp write + atomic os.replace), matching + # gstack-brain-enqueue's maildir contract: writers and the drain never + # share an inode, so a parallel writer or drain can't race this. + # Compact separators match the shim's JSON shape. + os.makedirs(spool_dir, exist_ok=True) + for i, (rel, key) in enumerate(to_enqueue): + rec = json.dumps({"file": rel, "ts": ts}, separators=(",", ":")) + tmp = os.path.join(spool_dir, f".tmp-{os.getpid()}-d{i}") + with open(tmp, "w") as f: + f.write(rec + "\n") + os.replace(tmp, os.path.join(spool_dir, f"{int(time.time())}-{os.getpid()}-d{i}.json")) + written.append((rel, key)) except OSError: - # Queue write failed (disk full, AV file lock). Leave the cursor - # unadvanced so these files are retried on the next discover instead of - # being silently recorded as synced (which loses the change until the - # file next changes). - to_enqueue = [] + # Spool write failed (disk full, AV file lock). Leave the cursor + # unadvanced for unwritten records so they are retried on the next + # discover instead of being silently recorded as synced (which loses + # the change until the file next changes). + pass # Advance the cursor only for records actually written. - for rel, key in to_enqueue: + for rel, key in written: new_cursor[rel] = key save_cursor(cursor_path, new_cursor) diff --git a/bin/gstack-brain-uninstall b/bin/gstack-brain-uninstall index e170b11dd..72841aca9 100755 --- a/bin/gstack-brain-uninstall +++ b/bin/gstack-brain-uninstall @@ -19,7 +19,8 @@ # .gitattributes — merge driver declarations # .brain-allowlist — sync path list # .brain-privacy-map.json — sync privacy classifier -# .brain-queue.jsonl — pending queue +# .brain-queue.d/ — pending spool (one file per record) +# .brain-queue.jsonl — legacy pending queue (pre-spool) # .brain-discover-cursor — discover-new cursor # .brain-last-push — timestamp marker # .brain-skip.txt — user-maintained skip list @@ -118,7 +119,9 @@ rm -f "$GSTACK_HOME/.gitignore" 2>/dev/null || true rm -f "$GSTACK_HOME/.gitattributes" 2>/dev/null || true rm -f "$GSTACK_HOME/.brain-allowlist" 2>/dev/null || true rm -f "$GSTACK_HOME/.brain-privacy-map.json" 2>/dev/null || true +rm -rf "$GSTACK_HOME/.brain-queue.d" 2>/dev/null || true rm -f "$GSTACK_HOME/.brain-queue.jsonl" 2>/dev/null || true +rm -f "$GSTACK_HOME/.brain-queue.jsonl.migrating" 2>/dev/null || true rm -f "$GSTACK_HOME/.brain-discover-cursor" 2>/dev/null || true rm -f "$GSTACK_HOME/.brain-last-push" 2>/dev/null || true rm -f "$GSTACK_HOME/.brain-last-pull" 2>/dev/null || true diff --git a/test/brain-sync.test.ts b/test/brain-sync.test.ts index 4f22f933d..bc42bd6fc 100644 --- a/test/brain-sync.test.ts +++ b/test/brain-sync.test.ts @@ -51,6 +51,28 @@ function git(args: string[], cwd?: string) { return { stdout: res.stdout || '', stderr: res.stderr || '', status: res.status ?? -1 }; } +// ---- spool helpers (maildir-style queue: one FILE per record) ---- +// Writers create --.json under .brain-queue.d/ via tmp + +// atomic rename; the drain deletes exactly the files it snapshotted. The +// legacy single-file .brain-queue.jsonl exists only as a migration source. +const spoolDir = () => path.join(tmpHome, '.brain-queue.d'); +const spoolFiles = () => + fs.existsSync(spoolDir()) + ? fs.readdirSync(spoolDir()).filter((f) => f.endsWith('.json')).sort() + : []; +const spoolText = () => + spoolFiles() + .map((f) => fs.readFileSync(path.join(spoolDir(), f), 'utf-8')) + .join(''); +let spoolSeq = 0; +function seedSpool(record: string): string { + fs.mkdirSync(spoolDir(), { recursive: true }); + spoolSeq += 1; + const name = `${Math.floor(Date.now() / 1000)}-${process.pid}-t${spoolSeq}.json`; + fs.writeFileSync(path.join(spoolDir(), name), record.endsWith('\n') ? record : record + '\n'); + return name; +} + beforeEach(() => { tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), 'brain-sync-home-')); bareRemote = fs.mkdtempSync(path.join(os.tmpdir(), 'brain-sync-remote-')); @@ -130,6 +152,7 @@ describe('gstack-brain-enqueue', () => { test('no-op when feature not initialized', () => { const r = run(['gstack-brain-enqueue', 'projects/foo/learnings.jsonl']); expect(r.status).toBe(0); + expect(fs.existsSync(spoolDir())).toBe(false); expect(fs.existsSync(path.join(tmpHome, '.brain-queue.jsonl'))).toBe(false); }); @@ -137,18 +160,22 @@ describe('gstack-brain-enqueue', () => { fs.mkdirSync(path.join(tmpHome, '.git'), { recursive: true }); const r = run(['gstack-brain-enqueue', 'projects/foo/learnings.jsonl']); expect(r.status).toBe(0); - expect(fs.existsSync(path.join(tmpHome, '.brain-queue.jsonl'))).toBe(false); + expect(fs.existsSync(spoolDir())).toBe(false); }); - test('enqueues when mode is full and .git exists', () => { + test('enqueues one spool file when mode is full and .git exists', () => { fs.mkdirSync(path.join(tmpHome, '.git'), { recursive: true }); run(['gstack-config', 'set', 'artifacts_sync_mode', 'full']); run(['gstack-brain-enqueue', 'projects/foo/learnings.jsonl']); - const queue = fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8'); - expect(queue).toContain('projects/foo/learnings.jsonl'); - const obj = JSON.parse(queue.trim()); + const files = spoolFiles(); + expect(files.length).toBe(1); + // Sortable maildir name: --.json. + expect(files[0]).toMatch(/^\d+-\d+-\d+\.json$/); + const obj = JSON.parse(fs.readFileSync(path.join(spoolDir(), files[0]), 'utf-8').trim()); expect(obj.file).toBe('projects/foo/learnings.jsonl'); expect(obj.ts).toBeTruthy(); + // No tmp-file droppings left behind. + expect(fs.readdirSync(spoolDir()).filter((f) => f.startsWith('.tmp-')).length).toBe(0); }); test('skip list honored', () => { @@ -157,12 +184,11 @@ describe('gstack-brain-enqueue', () => { fs.writeFileSync(path.join(tmpHome, '.brain-skip.txt'), 'projects/foo/secret.jsonl\n'); run(['gstack-brain-enqueue', 'projects/foo/secret.jsonl']); run(['gstack-brain-enqueue', 'projects/foo/ok.jsonl']); - const queue = fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8'); - expect(queue).not.toContain('secret.jsonl'); - expect(queue).toContain('ok.jsonl'); + expect(spoolText()).not.toContain('secret.jsonl'); + expect(spoolText()).toContain('ok.jsonl'); }); - test('concurrent enqueues all land (atomic append)', async () => { + test('concurrent enqueues all land (one spool file per record)', async () => { fs.mkdirSync(path.join(tmpHome, '.git'), { recursive: true }); run(['gstack-config', 'set', 'artifacts_sync_mode', 'full']); const procs = []; @@ -176,9 +202,10 @@ describe('gstack-brain-enqueue', () => { })); } await Promise.all(procs); - const queue = fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8'); - const lines = queue.trim().split('\n').filter(Boolean); - expect(lines.length).toBe(10); + expect(spoolFiles().length).toBe(10); + for (let i = 0; i < 10; i++) { + expect(spoolText()).toContain(`file-${i}.jsonl`); + } }); test('no args does not crash', () => { @@ -366,9 +393,8 @@ describe('gstack-brain-sync egress receipt gate', () => { expect(refused.stderr).toContain('EGRESS_RECEIPT_FAILED'); expect(refused.stderr).toContain('Fix: chmod -R u+w'); expect(refused.stderr).toContain('ATTEMPTS to send off-machine'); - // Queue intact (receipt is written BEFORE the commit consumes it). - const queue = fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8'); - expect(queue).toContain('projects/p/learnings.jsonl'); + // Spool intact (receipt is written BEFORE finalize consumes records). + expect(spoolText()).toContain('projects/p/learnings.jsonl'); // No local commit was created. expect(git(['rev-list', '--count', 'HEAD']).stdout.trim()).toBe(commitsBefore); // Nothing reached the remote. @@ -433,19 +459,17 @@ describe('gstack-brain-uninstall', () => { // --discover-new: cursor-based change detection // --------------------------------------------------------------- describe('gstack-brain-sync --discover-new', () => { - test('enqueues new allowlisted files; idempotent on re-run', () => { + test('enqueues new allowlisted files as spool records; idempotent on re-run', () => { run(['gstack-artifacts-init', '--remote', bareRemote]); run(['gstack-config', 'set', 'artifacts_sync_mode', 'full']); fs.mkdirSync(path.join(tmpHome, 'retros'), { recursive: true }); fs.writeFileSync(path.join(tmpHome, 'retros/week-1.md'), '# retro\n'); run(['gstack-brain-sync', '--discover-new']); - let queue = fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8'); - expect(queue).toContain('retros/week-1.md'); - // Clear queue, run again — idempotent (no new entries). - fs.writeFileSync(path.join(tmpHome, '.brain-queue.jsonl'), ''); + expect(spoolText()).toContain('retros/week-1.md'); + // Clear the spool, run again — idempotent (no new records). + for (const f of spoolFiles()) fs.unlinkSync(path.join(spoolDir(), f)); run(['gstack-brain-sync', '--discover-new']); - queue = fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8'); - expect(queue.trim()).toBe(''); + expect(spoolFiles().length).toBe(0); }); }); @@ -458,7 +482,6 @@ describe('#2549 queue integrity', () => { run(['gstack-artifacts-init', '--remote', bareRemote]); run(['gstack-config', 'set', 'artifacts_sync_mode', mode]); } - const queueText = () => fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8'); const statusJson = () => JSON.parse(fs.readFileSync(path.join(tmpHome, '.brain-sync-status.json'), 'utf-8')); test('privacy-held entries are RETAINED and classified, not wiped as "no allowlisted changes"', () => { @@ -470,9 +493,9 @@ describe('#2549 queue integrity', () => { const r = run(['gstack-brain-sync', '--once']); expect(r.status).toBe(0); // The exact #2549 repro: the old code truncated the queue here and said - // "no allowlisted changes in queue". The entry must survive, and the + // "no allowlisted changes in queue". The record must survive, and the // status must attribute the hold honestly. - expect(queueText()).toContain('projects/p/timeline.jsonl'); + expect(spoolText()).toContain('projects/p/timeline.jsonl'); const s = statusJson(); expect(s.status).toBe('idle'); expect(s.message).toContain('privacy-held retained'); @@ -484,13 +507,13 @@ describe('#2549 queue integrity', () => { fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); // Unmatched: no allowlist glob covers .txt scratch files. fs.writeFileSync(path.join(tmpHome, 'projects/p/scratch.txt'), 'x\n'); - fs.appendFileSync(path.join(tmpHome, '.brain-queue.jsonl'), '{"file":"projects/p/scratch.txt"}\n'); + seedSpool('{"file":"projects/p/scratch.txt"}'); // Missing: allowlisted name that does not exist on disk. - fs.appendFileSync(path.join(tmpHome, '.brain-queue.jsonl'), '{"file":"projects/p/learnings.jsonl"}\n'); + seedSpool('{"file":"projects/p/learnings.jsonl"}'); const r = run(['gstack-brain-sync', '--once']); expect(r.status).toBe(0); - expect(queueText()).not.toContain('scratch.txt'); - expect(queueText()).not.toContain('learnings.jsonl'); + expect(spoolText()).not.toContain('scratch.txt'); + expect(spoolText()).not.toContain('learnings.jsonl'); const s = statusJson(); expect(s.message).toContain('1 unmatched dropped'); expect(s.message).toContain('1 missing dropped'); @@ -504,17 +527,20 @@ describe('#2549 queue integrity', () => { expect(detail.dropped.missing).toContain('projects/p/learnings.jsonl'); }); - test('an unparseable queue line is preserved, never destroyed', () => { + test('an unparseable legacy queue line migrates as-is and is preserved, never destroyed', () => { + // The line lands in the legacy single-file queue (pre-spool writer); + // migration converts it verbatim to a spool record, and the drain keeps + // what it cannot parse. initWithMode('full'); fs.appendFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'not json at all\n'); const r = run(['gstack-brain-sync', '--once']); expect(r.status).toBe(0); - expect(queueText()).toContain('not json at all'); + expect(spoolText()).toContain('not json at all'); }); - test('surgical rewrite: a synced entry leaves the queue while a held sibling survives the same drain', () => { - // Proves the rewrite is a live filtered rewrite, not a truncation: two - // entries drain in one --once, one stages+pushes, one is mode-held. + test('finalize: a synced record leaves the spool while a held sibling survives the same drain', () => { + // Proves finalize is a per-record delete, not a truncation: two records + // drain in one --once, one stages+pushes, one is mode-held. initWithMode('artifacts-only'); fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); fs.writeFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"x","insight":"y","ts":"2026-01-01T00:00:00Z"}\n'); @@ -523,8 +549,8 @@ describe('#2549 queue integrity', () => { run(['gstack-brain-enqueue', 'projects/p/timeline.jsonl']); const r = run(['gstack-brain-sync', '--once']); expect(r.status).toBe(0); - expect(queueText()).not.toContain('learnings.jsonl'); // synced, removed - expect(queueText()).toContain('timeline.jsonl'); // held, retained + expect(spoolText()).not.toContain('learnings.jsonl'); // synced, removed + expect(spoolText()).toContain('timeline.jsonl'); // held, retained const log = spawnSync('git', ['--git-dir=' + bareRemote, 'log', '--oneline'], { encoding: 'utf-8' }); expect(log.stdout).toMatch(/sync: 1 file/); }); @@ -550,8 +576,8 @@ describe('#2549 queue integrity', () => { const s = statusJson(); expect(s.status).toBe('push_failed'); expect(s.message).toContain('commit retained locally'); - // Drained path left the queue — it lives in the local commit now. - expect(queueText()).not.toContain('learnings.jsonl'); + // Drained record left the spool — it lives in the local commit now. + expect(spoolText()).not.toContain('learnings.jsonl'); // The commit exists locally, ahead of origin. const ahead = git(['rev-list', '--count', 'origin/main..HEAD']).stdout.trim(); expect(Number(ahead)).toBeGreaterThan(0); @@ -680,3 +706,156 @@ describe('#2549 queue integrity', () => { expect(git(['rev-list', '--count', 'origin/main..HEAD']).stdout.trim()).toBe('0'); }); }); + +// --------------------------------------------------------------- +// C12 spool queue: per-record files kill the enqueue/drain race. +// One FILE per record under .brain-queue.d/ — writer and drainer never +// share an inode, so the lockless append-vs-rewrite race is structurally +// gone. Crash semantics are at-least-once (unfinalized records re-drain). +// --------------------------------------------------------------- +describe('C12 spool queue', () => { + function initWithMode(mode: string) { + run(['gstack-artifacts-init', '--remote', bareRemote]); + run(['gstack-config', 'set', 'artifacts_sync_mode', mode]); + } + const remoteLog = () => + spawnSync('git', ['--git-dir=' + bareRemote, 'log', '--oneline'], { encoding: 'utf-8' }).stdout; + + test('two rapid enqueues of different paths create two spool files; one drain syncs both', () => { + initWithMode('full'); + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); + fs.mkdirSync(path.join(tmpHome, 'retros'), { recursive: true }); + fs.writeFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"x","ts":"2026-01-01T00:00:00Z"}\n'); + fs.writeFileSync(path.join(tmpHome, 'retros/week-1.md'), '# retro\n'); + run(['gstack-brain-enqueue', 'projects/p/learnings.jsonl']); + run(['gstack-brain-enqueue', 'retros/week-1.md']); + expect(spoolFiles().length).toBe(2); + const r = run(['gstack-brain-sync', '--once']); + expect(r.status).toBe(0); + expect(spoolFiles().length).toBe(0); + expect(remoteLog()).toMatch(/sync: 2 file/); + }); + + test('a record created after a drain survives untouched and drains on the NEXT --once', () => { + // Structural form of the concurrent-append test: finalize deletes only + // snapshot-manifest files, so a record the drain never listed cannot be + // touched — whether it lands mid-drain or after. + initWithMode('full'); + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); + fs.mkdirSync(path.join(tmpHome, 'retros'), { recursive: true }); + fs.writeFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"a","ts":"2026-01-01T00:00:00Z"}\n'); + run(['gstack-brain-enqueue', 'projects/p/learnings.jsonl']); + expect(run(['gstack-brain-sync', '--once']).status).toBe(0); + expect(spoolFiles().length).toBe(0); + // New record arrives (a writer that raced the previous drain). + fs.writeFileSync(path.join(tmpHome, 'retros/week-1.md'), '# retro\n'); + run(['gstack-brain-enqueue', 'retros/week-1.md']); + const [pending] = spoolFiles(); + expect(pending).toBeTruthy(); + const pendingContent = fs.readFileSync(path.join(spoolDir(), pending), 'utf-8'); + expect(pendingContent).toContain('retros/week-1.md'); + // Untouched by the completed drain; the NEXT drain delivers it. + expect(run(['gstack-brain-sync', '--once']).status).toBe(0); + expect(spoolFiles().length).toBe(0); + expect(remoteLog()).toMatch(/sync: 1 file/); + }); + + test('at-least-once: a drain that fails before finalize leaves every spool file for the next run', () => { + if (process.platform === 'win32' || process.getuid?.() === 0) return; // chmod advisory there + initWithMode('full'); + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); + fs.mkdirSync(path.join(tmpHome, 'retros'), { recursive: true }); + fs.writeFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"a","ts":"2026-01-01T00:00:00Z"}\n'); + fs.writeFileSync(path.join(tmpHome, 'retros/week-1.md'), '# retro\n'); + run(['gstack-brain-enqueue', 'projects/p/learnings.jsonl']); + run(['gstack-brain-enqueue', 'retros/week-1.md']); + const seeded = spoolFiles(); + expect(seeded.length).toBe(2); + + // Break the egress-receipt ledger: the drain fails AFTER staging but + // BEFORE any commit or finalize — simulating a crash mid-drain. + fs.mkdirSync(path.join(tmpHome, 'security'), { recursive: true }); + fs.chmodSync(path.join(tmpHome, 'security'), 0o500); + try { + const refused = run(['gstack-brain-sync', '--once']); + expect(refused.status).toBe(1); + // The exact same spool files are still present — nothing consumed. + expect(spoolFiles()).toEqual(seeded); + } finally { + fs.chmodSync(path.join(tmpHome, 'security'), 0o700); + } + + // Next run re-drains the surviving records. + expect(run(['gstack-brain-sync', '--once']).status).toBe(0); + expect(spoolFiles().length).toBe(0); + expect(remoteLog()).toMatch(/sync: 2 file/); + }); + + test('legacy migration: .brain-queue.jsonl lines convert to spool records, nothing lost', () => { + // Pre-spool writers appended to the single-file queue. Three lines: two + // stageable artifacts, one behavioral (mode-held under artifacts-only). + initWithMode('artifacts-only'); + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); + fs.mkdirSync(path.join(tmpHome, 'retros'), { recursive: true }); + fs.writeFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"x","ts":"2026-01-01T00:00:00Z"}\n'); + fs.writeFileSync(path.join(tmpHome, 'retros/week-1.md'), '# retro\n'); + fs.writeFileSync(path.join(tmpHome, 'projects/p/timeline.jsonl'), '{"skill":"x","event":"started"}\n'); + fs.writeFileSync(path.join(tmpHome, '.brain-queue.jsonl'), + '{"file":"projects/p/learnings.jsonl","ts":"2026-01-01T00:00:00Z"}\n' + + '{"file":"retros/week-1.md","ts":"2026-01-01T00:00:01Z"}\n' + + '{"file":"projects/p/timeline.jsonl","ts":"2026-01-01T00:00:02Z"}\n'); + const r = run(['gstack-brain-sync', '--once']); + expect(r.status).toBe(0); + // Legacy file consumed; no .migrating remnant. + expect(fs.existsSync(path.join(tmpHome, '.brain-queue.jsonl'))).toBe(false); + expect(fs.existsSync(path.join(tmpHome, '.brain-queue.jsonl.migrating'))).toBe(false); + // Both artifacts synced; the behavioral record survives as a spool file. + expect(remoteLog()).toMatch(/sync: 2 file/); + expect(spoolText()).toContain('projects/p/timeline.jsonl'); + expect(spoolText()).not.toContain('learnings.jsonl'); + }); + + test('an unparseable spool record is kept and warned about; the drain continues', () => { + initWithMode('full'); + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); + fs.writeFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"x","ts":"2026-01-01T00:00:00Z"}\n'); + run(['gstack-brain-enqueue', 'projects/p/learnings.jsonl']); + const badFile = seedSpool('this is not json'); + const r = run(['gstack-brain-sync', '--once']); + expect(r.status).toBe(0); + expect(r.stderr).toContain('unparseable'); + // The good sibling synced; the unreadable record was never destroyed. + expect(remoteLog()).toMatch(/sync: 1 file/); + expect(spoolFiles()).toEqual([badFile]); + expect(spoolText()).toContain('this is not json'); + }); + + test('--status queue_depth counts spool records plus unmigrated legacy lines', () => { + initWithMode('full'); + seedSpool('{"file":"projects/p/a.jsonl","ts":"t"}'); + seedSpool('{"file":"projects/p/b.jsonl","ts":"t"}'); + fs.writeFileSync(path.join(tmpHome, '.brain-queue.jsonl'), '{"file":"projects/p/c.jsonl","ts":"t"}\n'); + const r = run(['gstack-brain-sync', '--status']); + expect(r.status).toBe(0); + const supplemental = JSON.parse(r.stdout.trim().split('\n').pop()!); + expect(supplemental.queue_depth).toBe(3); + }); + + test('--drop-queue keeps the --yes gate and counts spool + legacy entries', () => { + initWithMode('full'); + seedSpool('{"file":"projects/p/a.jsonl","ts":"t"}'); + seedSpool('{"file":"projects/p/b.jsonl","ts":"t"}'); + fs.writeFileSync(path.join(tmpHome, '.brain-queue.jsonl'), '{"file":"projects/p/c.jsonl","ts":"t"}\n'); + const refused = run(['gstack-brain-sync', '--drop-queue']); + expect(refused.status).toBe(1); + expect(refused.stderr).toContain('--yes'); + expect(spoolFiles().length).toBe(2); + const dropped = run(['gstack-brain-sync', '--drop-queue', '--yes']); + expect(dropped.status).toBe(0); + expect(dropped.stdout).toContain('dropped 3 queue entries'); + expect(spoolFiles().length).toBe(0); + expect(fs.readFileSync(path.join(tmpHome, '.brain-queue.jsonl'), 'utf-8')).toBe(''); + const again = run(['gstack-brain-sync', '--drop-queue', '--yes']); + expect(again.stdout).toContain('queue already empty'); + }); +});