From 184cf84ca9c411406c3eec6940fd57f5085fb09c Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Sun, 16 Aug 2026 09:00:55 -0700 Subject: [PATCH] fix(brain-sync): classify queue entries, rewrite surgically, re-push stranded commits Fixes #2549 (P0 data loss). Every drain exit previously truncated the WHOLE queue (six `: > "$QUEUE"` sites), which (a) destroyed privacy/mode-held entries while misattributing them as "no allowlisted changes", (b) destroyed entries enqueued concurrently during the drain, and (c) left push-failed commits stranded locally with nothing ever re-pushing them until unrelated new work arrived. Now: compute_paths_to_stage classifies every entry (stageable / retained privacy-held / dropped skipped-invalid-unmatched-missing); rewrite_queue re-reads the LIVE queue at mv time and removes only this drain's processed paths (retained + concurrent appends + unparseable lines survive; atomic tmp+mv); an unpushed-commit detector at run start re-pushes stranded local commits (receipted fail-closed; a receipt refusal skips the retry rather than wedging the drain; guards missing origin/; runs inside the existing lock). Status lines carry counts; full drop paths go to a 0600 sidecar (.brain-sync-drops.json) so filenames stay out of transcripts. --drop-queue remains the one intentional truncation. Matrix added: privacy retention, unmatched/missing counted drops + sidecar mode, unparseable-line preservation, surgical same-drain retention, push-fail commit retention + detector re-delivery on an EMPTY queue, receipt-refusal skip. 35/35 in test/brain-sync.test.ts. Co-Authored-By: Claude Fable 5 --- bin/gstack-brain-sync | 173 +++++++++++++++++++++++++++++++++++----- test/brain-sync.test.ts | 153 +++++++++++++++++++++++++++++++++++ 2 files changed, 305 insertions(+), 21 deletions(-) diff --git a/bin/gstack-brain-sync b/bin/gstack-brain-sync index 2fa696869..d84666484 100755 --- a/bin/gstack-brain-sync +++ b/bin/gstack-brain-sync @@ -122,12 +122,23 @@ sys.exit(0) # Compute matched allowlisted, privacy-filtered path set from queue. # Output: newline-delimited relative paths that should be staged. +# +# #2549: every non-staged queue entry is CLASSIFIED, never silently discarded. +# When $2 is given, a JSON classification lands there: +# {"retained": [privacy/mode-held paths that stay queued], +# "dropped": {"skipped": [...], "invalid": [...], "unmatched": [...], "missing": [...]}} +# retained entries would sync if the user raises artifacts_sync_mode, so they +# stay in the queue; dropped classes can never sync (explicit skip, escape +# 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". compute_paths_to_stage() { local mode="$1" - python3 - "$GSTACK_HOME" "$QUEUE" "$ALLOWLIST" "$PRIVACY_MAP" "$SKIP_FILE" "$mode" <<'PYEOF' + local class_file="${2:-}" + python3 - "$GSTACK_HOME" "$QUEUE" "$ALLOWLIST" "$PRIVACY_MAP" "$SKIP_FILE" "$mode" "$class_file" <<'PYEOF' import sys, json, os, fnmatch, glob -gstack_home, queue, allowlist_path, privacy_path, skip_path, mode = sys.argv[1:7] +gstack_home, queue, allowlist_path, privacy_path, skip_path, mode, class_file = sys.argv[1:8] def load_lines(path): try: @@ -195,29 +206,119 @@ def mode_allows(cls, mode): return True # full final = [] +classified = {"retained": [], "dropped": {"skipped": [], "invalid": [], "unmatched": [], "missing": []}} for p in sorted(queue_paths): if p in skip_lines: + classified["dropped"]["skipped"].append(p) continue # Must be under GSTACK_HOME root. Reject absolute + reject ../ escape. if p.startswith("/") or ".." in p.split("/"): + classified["dropped"]["invalid"].append(p) continue # Must match at least one allowlist glob. if not path_matches_any(p, allowlist_globs): + classified["dropped"]["unmatched"].append(p) continue - # Must survive privacy mode filter. + # Must survive privacy mode filter — held entries STAY QUEUED (retained): + # they would sync under a higher artifacts_sync_mode, and reporting them + # as "no allowlisted changes" was #2549's misattribution. cls = privacy_class(p, privacy_map) if not mode_allows(cls, mode): + classified["retained"].append(p) continue # Must exist on disk — can't stage what isn't there. if not os.path.exists(os.path.join(gstack_home, p)): + classified["dropped"]["missing"].append(p) continue final.append(p) +if class_file: + with open(class_file, "w") as f: + json.dump(classified, f) + for p in final: print(p) 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 + python3 - "$QUEUE" "$paths_file" "$class_file" "$GSTACK_HOME/.brain-sync-drops.json" <<'PYEOF' 2>/dev/null || true +import json, os, sys, time +queue, paths_file, class_file, drops_file = sys.argv[1:5] + +def lines(path): + try: + with open(path) as f: + return [l.rstrip("\r\n") for l in f if l.strip()] + except FileNotFoundError: + return [] + +staged = set(lines(paths_file)) +try: + with open(class_file) as f: + classified = json.load(f) +except Exception: + classified = {"retained": [], "dropped": {}} +retained = set(classified.get("retained", [])) +dropped = set() +for group in (classified.get("dropped", {}) or {}).values(): + dropped.update(group) +processed = staged | dropped + +kept = [] +for line in lines(queue): # LIVE re-read: concurrent enqueues survive (F3) + try: + p = json.loads(line).get("file") + except Exception: + kept.append(line) # unparseable line: keep, never destroy + continue + if not isinstance(p, str) or p in retained or p not in processed: + kept.append(line) + +tmp = queue + ".tmp." + str(os.getpid()) +with open(tmp, "w") as f: + for l in kept: + f.write(l + "\n") +os.replace(tmp, queue) + +if dropped: + fd = os.open(drops_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + json.dump({"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "dropped": classified.get("dropped", {})}, f) +PYEOF +} + +# Human-readable classification counts for status messages. +queue_summary() { + local class_file="$1" + python3 - "$class_file" <<'PYEOF' 2>/dev/null || echo "" +import json, sys +try: + with open(sys.argv[1]) as f: + c = json.load(f) +except Exception: + print(""); sys.exit(0) +d = c.get("dropped", {}) or {} +parts = [] +r = len(c.get("retained", [])) +if r: parts.append(f"{r} privacy-held retained") +for k in ("skipped", "unmatched", "missing", "invalid"): + n = len(d.get(k, [])) + if n: parts.append(f"{n} {k} dropped") +print("; ".join(parts)) +PYEOF +} + subcmd_once() { if ! sync_active; then # Silent no-op when feature not initialized / disabled. @@ -253,16 +354,42 @@ subcmd_once() { local mode mode=$("$CONFIG_BIN" get artifacts_sync_mode 2>/dev/null || echo off) - local paths_file - paths_file=$(mktemp /tmp/brain-sync-paths.XXXXXX) || { rm -rf "$lock_dir" 2>/dev/null; write_status "error" "mktemp failed"; exit 1; } - # Single trap covers both: lock cleanup AND tempfile cleanup. - trap 'rm -f "$paths_file" 2>/dev/null; rm -rf "$lock_dir" 2>/dev/null || true' EXIT INT TERM + # #2549 unpushed-commit detector: a prior drain may have COMMITTED but + # failed to push (auth blip, offline). The data was never lost — it sits in + # a local commit — but nothing re-pushed it until NEW changes arrived. + # Retry the push up front, inside the lock. Receipted fail-closed like + # every other push; a receipt REFUSAL skips the retry without blocking the + # rest of the drain (local staging must not wedge on receipt problems). + # Guards: origin/ may not exist yet (first sync, deleted remote). + local det_branch det_unpushed + det_branch=$(git -C "$GSTACK_HOME" rev-parse --abbrev-ref HEAD 2>/dev/null || echo "") + if [ -n "$det_branch" ] && git -C "$GSTACK_HOME" rev-parse --verify --quiet "origin/$det_branch" >/dev/null 2>&1; then + det_unpushed=$(git -C "$GSTACK_HOME" rev-list --count "origin/$det_branch..HEAD" 2>/dev/null || echo 0) + case "$det_unpushed" in ''|*[!0-9]*) det_unpushed=0 ;; esac + if [ "$det_unpushed" -gt 0 ]; then + local det_host + det_host=$(remote_host) + if GSTACK_HOME="$GSTACK_HOME" _receipted_git closed brain-sync "$det_host" curated-memory-git-push "artifacts_sync_mode!=off" \ + bash -c 'git -C "$1" push origin HEAD 2>/dev/null' _ "$GSTACK_HOME"; then + date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE" + fi + fi + fi - compute_paths_to_stage "$mode" > "$paths_file" + local paths_file class_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; } + # 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 + + compute_paths_to_stage "$mode" "$class_file" > "$paths_file" if [ ! -s "$paths_file" ]; then - # Nothing to stage. Clear any stale queue entries and exit. - : > "$QUEUE" - write_status "idle" "no allowlisted changes in queue" + # Nothing stageable. Rewrite the queue (retained entries + concurrent + # appends survive; classified drops removed) instead of truncating it. + rewrite_queue "$paths_file" "$class_file" + local summary + summary=$(queue_summary "$class_file") + write_status "idle" "no stageable changes${summary:+ ($summary)}" exit 0 fi @@ -309,8 +436,9 @@ subcmd_once() { local msg="sync: $n file(s) | $ts" 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). - : > "$QUEUE" + # 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" write_status "idle" "queue drained but no new changes to commit" exit 0 } @@ -322,10 +450,12 @@ subcmd_once() { if echo "$push_err" | grep -qiE "auth|permission|403|401|forbidden"; then local hint hint=$(remote_auth_hint) - write_status "push_failed" "push failed: auth error. fix: $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 - # Queue cleared because the commit exists locally; next push will send it. - : > "$QUEUE" + # Drained paths leave the queue — 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" exit 0 fi @@ -339,20 +469,21 @@ 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 - : > "$QUEUE" + rewrite_queue "$paths_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 fi fi fi - write_status "push_failed" "push failed: $(printf '%s' "$push_err" | head -1)" - : > "$QUEUE" + # 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" exit 0 } - # Success: clear queue, update last-push. - : > "$QUEUE" + # Success: drained paths leave the queue (retained + concurrent survive). + rewrite_queue "$paths_file" "$class_file" date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE" write_status "ok" "pushed $n file(s)" exit 0 diff --git a/test/brain-sync.test.ts b/test/brain-sync.test.ts index a4587f806..433c73e0a 100644 --- a/test/brain-sync.test.ts +++ b/test/brain-sync.test.ts @@ -448,3 +448,156 @@ describe('gstack-brain-sync --discover-new', () => { expect(queue.trim()).toBe(''); }); }); + +// --------------------------------------------------------------- +// #2549 queue integrity: classified drops, privacy retention, +// surgical rewrite, unpushed-commit detector +// --------------------------------------------------------------- +describe('#2549 queue integrity', () => { + function initWithMode(mode: string) { + 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"', () => { + // timeline.jsonl is class=behavioral; artifacts-only mode holds it. + initWithMode('artifacts-only'); + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { recursive: true }); + fs.writeFileSync(path.join(tmpHome, 'projects/p/timeline.jsonl'), '{"skill":"x","event":"started"}\n'); + run(['gstack-brain-enqueue', 'projects/p/timeline.jsonl']); + 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 + // status must attribute the hold honestly. + expect(queueText()).toContain('projects/p/timeline.jsonl'); + const s = statusJson(); + expect(s.status).toBe('idle'); + expect(s.message).toContain('privacy-held retained'); + expect(s.message).not.toContain('no allowlisted changes'); + }); + + test('unmatched and missing entries drop WITH counts and a 0600 drops sidecar', () => { + initWithMode('full'); + 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'); + // Missing: allowlisted name that does not exist on disk. + fs.appendFileSync(path.join(tmpHome, '.brain-queue.jsonl'), '{"file":"projects/p/learnings.jsonl"}\n'); + const r = run(['gstack-brain-sync', '--once']); + expect(r.status).toBe(0); + expect(queueText()).not.toContain('scratch.txt'); + expect(queueText()).not.toContain('learnings.jsonl'); + const s = statusJson(); + expect(s.message).toContain('1 unmatched dropped'); + expect(s.message).toContain('1 missing dropped'); + const drops = path.join(tmpHome, '.brain-sync-drops.json'); + expect(fs.existsSync(drops)).toBe(true); + if (process.platform !== 'win32') { + expect(fs.statSync(drops).mode & 0o777).toBe(0o600); + } + const detail = JSON.parse(fs.readFileSync(drops, 'utf-8')); + expect(detail.dropped.unmatched).toContain('projects/p/scratch.txt'); + expect(detail.dropped.missing).toContain('projects/p/learnings.jsonl'); + }); + + test('an unparseable queue line is preserved, never destroyed', () => { + 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'); + }); + + 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. + 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'); + fs.writeFileSync(path.join(tmpHome, 'projects/p/timeline.jsonl'), '{"skill":"x","event":"started"}\n'); + run(['gstack-brain-enqueue', 'projects/p/learnings.jsonl']); + 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 + const log = spawnSync('git', ['--git-dir=' + bareRemote, 'log', '--oneline'], { encoding: 'utf-8' }); + expect(log.stdout).toMatch(/sync: 1 file/); + }); + + test('push failure retains the commit locally and the run-start detector re-pushes it', () => { + initWithMode('full'); + // Establish origin/main so the detector has a remote ref to compare. + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { 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); + + // Reject the next push at the remote (pre-receive hook exits 1 with an + // auth-shaped message so the auth branch is exercised too). + const hook = path.join(bareRemote, 'hooks', 'pre-receive'); + fs.writeFileSync(hook, '#!/bin/sh\necho "403 forbidden" >&2\nexit 1\n'); + fs.chmodSync(hook, 0o755); + + fs.appendFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"b","ts":"2026-01-02T00:00:00Z"}\n'); + run(['gstack-brain-enqueue', 'projects/p/learnings.jsonl']); + const fail = run(['gstack-brain-sync', '--once']); + expect(fail.status).toBe(0); + 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'); + // The commit exists locally, ahead of origin. + const ahead = git(['rev-list', '--count', 'origin/main..HEAD']).stdout.trim(); + expect(Number(ahead)).toBeGreaterThan(0); + + // Remote healthy again: an EMPTY-queue run must still deliver the + // stranded commit (the detector, not the drain, pushes it). + fs.rmSync(hook); + const retry = run(['gstack-brain-sync', '--once']); + expect(retry.status).toBe(0); + const log = spawnSync('git', ['--git-dir=' + bareRemote, 'log', '--oneline'], { encoding: 'utf-8' }); + expect(log.stdout).toMatch(/sync: 1 file/); + expect(git(['rev-list', '--count', 'origin/main..HEAD']).stdout.trim()).toBe('0'); + }); + + test('receipt refusal at the detector skips the retry without wedging the drain', () => { + if (process.platform === 'win32' || process.getuid?.() === 0) return; // chmod advisory there + initWithMode('full'); + fs.mkdirSync(path.join(tmpHome, 'projects', 'p'), { 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); + + // Strand a commit: reject pushes, drain once. + const hook = path.join(bareRemote, 'hooks', 'pre-receive'); + fs.writeFileSync(hook, '#!/bin/sh\nexit 1\n'); + fs.chmodSync(hook, 0o755); + fs.appendFileSync(path.join(tmpHome, 'projects/p/learnings.jsonl'), '{"skill":"b","ts":"2026-01-02T00:00:00Z"}\n'); + run(['gstack-brain-enqueue', 'projects/p/learnings.jsonl']); + expect(run(['gstack-brain-sync', '--once']).status).toBe(0); + fs.rmSync(hook); + + // Break receipts. The detector's retry must be SKIPPED (no wedge), and + // the run must still exit 0 with nothing else to do. + fs.mkdirSync(path.join(tmpHome, 'security'), { recursive: true }); + fs.chmodSync(path.join(tmpHome, 'security'), 0o500); + try { + const r = run(['gstack-brain-sync', '--once']); + expect(r.status).toBe(0); + // Commit still stranded (retry skipped, not attempted unreceipted). + expect(Number(git(['rev-list', '--count', 'origin/main..HEAD']).stdout.trim())).toBeGreaterThan(0); + } finally { + fs.chmodSync(path.join(tmpHome, 'security'), 0o700); + } + + // Receipts healthy: detector delivers. + expect(run(['gstack-brain-sync', '--once']).status).toBe(0); + expect(git(['rev-list', '--count', 'origin/main..HEAD']).stdout.trim()).toBe('0'); + }); +});