gstack/bin/gstack-brain-sync

#!/usr/bin/env bash
# gstack-brain-sync — drain queue, commit allowlisted paths, push to remote.
#
# Usage:
#   gstack-brain-sync --once           drain queue, commit, push (default)
#   gstack-brain-sync --status         print sync health as JSON
#   gstack-brain-sync --skip-file <p>  permanently exclude <p> (and any coupled
#                                      group it belongs to) from future syncs
#   gstack-brain-sync --unskip-file <p>  undo --skip-file and queue <p> again
#   gstack-brain-sync --drop-queue --yes   clear queue without committing
#   gstack-brain-sync --discover-new   scan allowlist dirs, enqueue changed files
#   gstack-brain-sync --attention      print skill start's attention lines
#   gstack-brain-sync --publish-removals [--yes]
#                                      list (or, with --yes, commit and push)
#                                      the removal of synced files gone from disk
#
# Invoked by the preamble at skill START and END boundaries. No persistent
# daemon. Typical run <1s when queue empty; ~200-800ms with network push.
#
# One drain at a time: the drain lock (bin/gstack-brain-lock.sh) is shared
# with every other gstack writer to ~/.gstack's git state. A drain that finds
# it held skips; the next skill boundary catches up.
#
# Removals: with artifacts_sync_removals=on, each drain also publishes the
# removal of synced files that are gone from disk (a deleted design doc, the
# old directory after a project slug migration), so the remote and the brain
# stop serving stale copies. A drain that would remove more than
# REMOVAL_VALVE files at once publishes none of them and says so: that many
# at once looks like a loss, not a cleanup. --publish-removals --yes publishes
# them on demand, whatever the setting.
#
# A file the secret scan flags is held back on its own (with the rest of its
# coupled group) and stays queued; everything else syncs. `--status` lists
# each held file, the scanner rule that matched (never the matched text) and
# both fixes.
#
# Sync status table. Skill start prints the attention line of every row that
# applies; `ok` and `idle` print none. <bin> is this script's absolute path
# and <N> the held count. Rows are read by test/gstack-skill-start.test.ts.
#
#   held         trigger: status=held: the secret scan held back flagged files and synced the rest
#                line:    ARTIFACTS_SYNC: attention: status=held held=<N>. The secret scan is holding back <N> file(s); everything else still syncs. See which files and how to fix them: <bin> --status
#                fix:     edit the file so the scan no longer matches, or <bin> --skip-file <path> (permanent; undo with <bin> --unskip-file <path>)
#   blocked      trigger: status=blocked: a re-scan still flagged after the hold, so nothing was committed
#                line:    ARTIFACTS_SYNC: attention: status=blocked. The secret scan flagged a file it could not hold back on its own, so nothing is syncing. See the fix: <bin> --status
#                fix:     edit or skip the files named by <bin> --status, or clear the batch with <bin> --drop-queue --yes
#   push_failed  trigger: status=push_failed: a commit is saved locally but the push failed
#                line:    ARTIFACTS_SYNC: attention: status=push_failed. Synced files are committed locally but the push failed; it retries at skill start. See the cause: <bin> --status
#                fix:     the fix <bin> --status names (usually: gh auth status)
#   error        trigger: status=error: the drain stopped before committing and kept the queue
#                line:    ARTIFACTS_SYNC: attention: status=error. The last sync stopped before committing; the queue is kept and retried. See the cause: <bin> --status
#                fix:     the cause <bin> --status names (git index lock, commit hook, git add failure)
#   unknown      trigger: the status file holds a status code outside this table
#                line:    ARTIFACTS_SYNC: attention: status=unknown. The sync status file is unreadable. See it: <bin> --status
#                fix:     run <bin> --once to rewrite it
#   stale-push   trigger: no push for 24 hours while the drain's drainable count is above zero
#                line:    ARTIFACTS_SYNC: attention: stale-push. Files are ready to sync but nothing has been pushed for over 24 hours. See why: <bin> --status
#                fix:     the waiting reason <bin> --status names (a coupled group waiting for a consistent generation, a git add failure)
#   stale-drain  trigger: last_drain_at older than 24 hours while the queue directory holds records
#                line:    ARTIFACTS_SYNC: attention: stale-drain. The sync has not run for over 24 hours while files wait in the queue. See why: <bin> --status
#                fix:     if no gstack process is running, remove ~/.gstack/.brain-sync.lock.d, then run <bin> --once
#
# Env:
#   GSTACK_HOME — override ~/.gstack (aligns with writers).

# Heredoc delivery guard. bash 5.2+ writes a heredoc body <=64KiB through a
# pipe in the forked child before exec, with no reader on the other end. On
# macOS under pipe-KVA pressure a fresh pipe gets a 512-byte buffer, so any
# body >=512B blocks write() forever and the script hangs at startup with no
# output. Compat level 50 restores the tempfile path. These scripts are
# bash-3.2-clean, so the compat level costs them nothing. Not exported: the
# guard is per-script, and it survives `bash script.sh` call sites that
# bypass the shebang.
BASH_COMPAT=50

set -uo pipefail

. "$(dirname "$0")/gstack-state-root.sh" 2>/dev/null || { echo "$0: cannot resolve the gstack state root: $(dirname "$0")/gstack-state-root.sh is missing. fix: reinstall with ./setup or /gstack-upgrade (docs/state-root.md)" >&2; exit 1; }
gstack_state_root_select
GSTACK_HOME="$_gstack_sr_root"
# Maildir-style spool: one FILE per record, <epoch>-<pid>-<uniq>.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"
SKIP_FILE="$GSTACK_HOME/.brain-skip.txt"
STATUS_FILE="$GSTACK_HOME/.brain-sync-status.json"
LAST_PUSH_FILE="$GSTACK_HOME/.brain-last-push"
DISCOVER_CURSOR="$GSTACK_HOME/.brain-discover-cursor"

SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
CONFIG_BIN="$SCRIPT_DIR/gstack-config"
SELF_BIN="$SCRIPT_DIR/gstack-brain-sync"

# Egress receipt helpers (_receipted_git): receipt-before-send, fail-closed.
. "$SCRIPT_DIR/gstack-egress-lib.sh"
# The drain lock shared with every gstack writer to the artifacts repo.
. "$SCRIPT_DIR/gstack-brain-lock.sh"

# Coupled groups: artifacts that reference each other and must reach the
# remote as one generation. Members are globs sharing one "*" segment (the same
# value binds every member); `check` names the consistency test the staged
# generation must pass before it is committed. Paths not listed here are
# independent.
COUPLED_GROUPS='[{"name": "decisions", "members": ["projects/*/decisions.jsonl", "projects/*/decisions.active.json"], "check": "decisions-active"}]'

REMOVAL_VALVE=20

# Fields every status write of this drain carries (see write_status).
ST_HELD='[]'
ST_DRAINABLE=0
ST_NOTE=""

# origin host for receipt records (github.com etc).
remote_host() {
  local url host
  url=$(git -C "$GSTACK_HOME" remote get-url origin 2>/dev/null || echo "")
  host="${url#*://}"; host="${host#*@}"; host="${host%%[/:]*}"
  echo "${host:-unknown}"
}

# Remote-specific hint for auth errors (branch on origin URL).
remote_auth_hint() {
  local url
  url=$(git -C "$GSTACK_HOME" remote get-url origin 2>/dev/null || echo "")
  case "$url" in
    *github.com*|*@github.*) echo "run: gh auth status (and gh auth refresh if needed)" ;;
    *gitlab*) echo "run: glab auth status" ;;
    *) echo "check 'git remote -v' and your credentials" ;;
  esac
}

# Status file schema (one JSON object; readers: --status, --attention, doctor):
#   status        ok | idle | held | blocked | push_failed | error
#   held_count    files the secret scan held back this drain (hits + dependents)
#   drainable     queued files a drain could publish now: excludes scanner-held
#                 and privacy-held records
#   last_drain_at UTC time this drain ran (every write happens inside a drain)
#   last_push_at  UTC time of the last successful push, or null
#   ts            same as last_drain_at (kept for older readers)
#   held          [{path, rule, dependents[], fixes[]}]: one entry per flagged
#                 path; rule is the scanner rule name, never the matched text
#   message       human detail for --status; never printed at skill start
# Only status, the counts and the timestamps feed skill start's attention
# lines, each clamped to a closed set, digits, or a timestamp shape.
write_status() {
  local code="$1" msg="$2${ST_NOTE}" ts last_push=""
  ts=$(date -u +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo "")
  [ -f "$LAST_PUSH_FILE" ] && last_push=$(head -1 "$LAST_PUSH_FILE" 2>/dev/null || echo "")
  python3 - "$STATUS_FILE" "$code" "$msg" "$ts" "$last_push" "$ST_DRAINABLE" "$ST_HELD" <<'PYEOF' 2>/dev/null || true
import json, re, sys
path, code, msg, ts, last_push, drainable, held = sys.argv[1:8]
try:
    held = [h for h in json.loads(held) if isinstance(h, dict)]
except Exception:
    held = []
counted = set()
for h in held:
    counted.add(h.get("path"))
    counted.update(h.get("dependents") or [])
last_push = last_push.strip()
data = {
    "status": code,
    "held_count": len(counted),
    "drainable": int(drainable) if drainable.isdigit() else 0,
    "last_drain_at": ts,
    "last_push_at": last_push if re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ", last_push) else None,
    "ts": ts,
    "held": held,
    "message": msg,
}
with open(path, "w") as f:
    json.dump(data, f)
    f.write("\n")
PYEOF
}

# Read config; return 0 if sync active, 1 otherwise.
sync_active() {
  if [ ! -d "$GSTACK_HOME/.git" ]; then
    return 1
  fi
  local mode
  mode=$("$CONFIG_BIN" get artifacts_sync_mode 2>/dev/null || echo off)
  [ "$mode" = "off" ] && return 1
  return 0
}

# Secret scan over a staged diff file. Pattern families are the
# pre-commit hook's; only the family name ever leaves this function, never the
# matched text. The diff is read as bytes and decoded with replacement, so
# undecodable bytes cannot crash the scan into a false "clean".
#   secret_scan gate <diff>                     exit 0 clean, 3 hit (prints the
#                                               family names); anything else is
#                                               a scanner failure (fail closed)
#   secret_scan attribute <diff> <names> <out>  one pass: splits the diff on its
#                                               `diff --git` headers, pairs each
#                                               section with the NUL-separated
#                                               names from `git diff --name-only
#                                               -z` (same order), writes
#                                               "<family>\t<path>" per flagged
#                                               path; exit 0, or 4 when sections
#                                               and names disagree
secret_scan() {
  python3 - "$@" <<'PYEOF'
import re, sys
patterns = [
    ('aws-access-key', re.compile(r'AKIA[0-9A-Z]{16}')),
    ('github-token', re.compile(r'\b(gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,})')),
    ('openai-key', re.compile(r'\bsk-[A-Za-z0-9_-]{20,}')),
    ('pem-block', re.compile(r'-----BEGIN [A-Z ]{3,}-----')),
    ('jwt', re.compile(r'\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b')),
    ('bearer-token-json',
     # JSON-embedded auth headers. The optional Bearer/Basic/Token prefix
     # matters: real auth values include a literal space after the scheme
     # name, but the value charset below does not include spaces, so
     # without the optional prefix every Bearer token in a JSON blob slips
     # past the scanner.
     re.compile(r'"(authorization|api[_-]?key|apikey|token|secret|password)"\s*:\s*"(Bearer |Basic |Token )?[A-Za-z0-9_./+=-]{16,}"',
                re.IGNORECASE)),
]

def families(text):
    return [name for name, rx in patterns if rx.search(text)]

def read(path):
    with open(path, "rb") as f:
        return f.read()

mode = sys.argv[1]
if mode == "gate":
    hits = families(read(sys.argv[2]).decode("utf-8", "replace"))
    if hits:
        print(",".join(hits))
        sys.exit(3)
    sys.exit(0)

diff_path, names_path, out_path = sys.argv[2:5]
names = [n.decode("utf-8", "surrogateescape") for n in read(names_path).split(b"\0") if n]
sections = []
for line in read(diff_path).split(b"\n"):
    if line.startswith(b"diff --git "):
        sections.append([])
    elif sections:
        sections[-1].append(line)
if len(sections) != len(names):
    sys.exit(4)
with open(out_path, "w", encoding="utf-8", errors="surrogateescape", newline="\n") as out:
    for name, body in zip(names, sections):
        if "\n" in name or "\t" in name:
            sys.exit(4)
        hits = families(b"\n".join(body).decode("utf-8", "replace"))
        if hits:
            out.write(hits[0] + "\t" + name + "\n")
PYEOF
}

# 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
# NARROWS the tail-race window (transition-only: it applies to pre-spool
# writers, and a writer that appends after the second read but before the
# unlink can still lose that line; spool-native writers are immune).
# Unparseable lines migrate as-is; finalize_queue quarantines + 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 the leftover STILL holds records, the conversion failed (e.g. python3
  # unavailable). The mv below would overwrite it and destroy those records —
  # exactly the never-destroy invariant above. Defer this run's migration;
  # the next run retries both files.
  [ -s "$migrating" ] && return 0
  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.
# 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".
#
# 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.
#
# Coupled groups ($COUPLED_GROUPS): a stageable path pulls in the rest of its
# group (each member that exists, is allowlisted and passes the privacy mode),
# so a generation is staged whole. A skipped member drops the whole group
# (skip is per group, so the remote never gets a partial generation); a
# privacy-held member retains the whole group. When $5 is given, the groups
# with a staged member land there as JSON:
#   [{"name", "check", "members": [all member paths], "staged": [...]}]
compute_paths_to_stage() {
  local mode="$1"
  local class_file="${2:-}"
  local snapshot_file="${3:-}"
  local transcripts_ok="${4:-0}"
  local groups_file="${5:-}"
  python3 - "$GSTACK_HOME" "$QUEUE_DIR" "$ALLOWLIST" "$PRIVACY_MAP" "$SKIP_FILE" "$mode" "$class_file" "$snapshot_file" "$transcripts_ok" "$groups_file" "$COUPLED_GROUPS" <<'PYEOF'
import sys, json, os, fnmatch, glob, re

gstack_home, spool_dir, allowlist_path, privacy_path, skip_path, mode, class_file, snapshot_file, transcripts_ok, groups_file, coupled = sys.argv[1:12]

def load_lines(path):
    try:
        with open(path) as f:
            return [l.strip() for l in f if l.strip() and not l.lstrip().startswith("#")]
    except FileNotFoundError:
        return []

def load_privacy_map(path):
    # Returns (entries, corrupt). Non-dict entries are filtered out
    # defensively — the map may be PULLED from the artifacts remote, so a
    # malformed entry like ["bad"] is remotely triggerable and used to raise
    # mid-classification (after the snapshot manifest was written), which the
    # old finalize turned into a full queue wipe. Any malformed shape also
    # marks the map CORRUPT: privacy classification cannot be trusted, so the
    # caller holds every queued record instead of guessing (a corrupt privacy
    # map silently treated as empty would over-share behavioral data).
    try:
        with open(path) as f:
            data = json.load(f)
    except FileNotFoundError:
        return [], False
    except json.JSONDecodeError:
        return [], True
    if not isinstance(data, list):
        return [], True
    # Expected: [{"pattern": "glob", "class": "artifact" | "behavioral"}]
    entries = [e for e in data if isinstance(e, dict)]
    return entries, len(entries) != len(data)

allowlist_globs = load_lines(allowlist_path)
privacy_map, privacy_corrupt = load_privacy_map(privacy_path)
# Normalize skip entries to the POSIX form queued paths use, so a backslash
# entry in .brain-skip.txt still matches on Windows. The drain is the safety
# boundary that actually stages files, so it must normalize identically to
# discover_new — otherwise an explicitly-skipped file gets committed.
skip_lines = {s.replace(os.sep, "/") for s in load_lines(skip_path)}

# 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:
    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:
        if fnmatch.fnmatchcase(path, pattern):
            return True
    return False

def privacy_class(path, mapping):
    for entry in mapping:
        pat = entry.get("pattern")
        if pat and fnmatch.fnmatchcase(path, pat):
            return entry.get("class", "artifact")
    # Default class when no pattern matches: artifact (safe default).
    return "artifact"

# mode filter: 'off' → nothing; 'artifacts-only' → only artifact class;
# 'full' → both classes.
def mode_allows(cls, mode):
    if mode == "off":
        return False
    if mode == "artifacts-only":
        return cls == "artifact"
    return True  # full

def group_of(path):
    # (group spec, member paths) when path is a member of a coupled group.
    for spec in json.loads(coupled):
        for member in spec["members"]:
            m = re.fullmatch(re.escape(member).replace(r"\*", "([^/]+)"), path)
            if m:
                return spec, [x.replace("*", m.group(1), 1) for x in spec["members"]]
    return None, []

final = []
groups = {}
classified = {"retained": [], "dropped": {"skipped": [], "invalid": [], "unmatched": [], "missing": []}}
if privacy_corrupt:
    # Fail-safe: with an untrustworthy privacy map, stage NOTHING and drop
    # NOTHING — retain every queued record until the map is fixed. The next
    # drain re-classifies from scratch.
    print("BRAIN_SYNC: warning: privacy map at " + privacy_path +
          " is malformed — holding all queued records until it is fixed", file=sys.stderr)
    classified["retained"] = sorted(queue_paths)
    queue_paths = set()
for p in sorted(queue_paths):
    spec, members = group_of(p)
    if p in skip_lines or any(m in skip_lines for m in members):
        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
    # Staged transcript pages need transcript consent; without it they stay
    # queued (retained) and on disk, and push once the user consents.
    if transcripts_ok != "1" and p.split("/")[0] == "transcripts":
        classified["retained"].append(p)
        continue
    # 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
    siblings = [m for m in members if m != p
                and path_matches_any(m, allowlist_globs)
                and os.path.exists(os.path.join(gstack_home, m))]
    if any(not mode_allows(privacy_class(m, privacy_map), mode) for m in siblings):
        classified["retained"].append(p)
        continue
    for m in [p] + siblings:
        if m not in final:
            final.append(m)
    if spec:
        groups.setdefault(members[0], {"name": spec["name"], "check": spec.get("check", ""),
                                       "members": members, "staged": []})

for g in groups.values():
    g["staged"] = [m for m in g["members"] if m in final]

if class_file:
    with open(class_file, "w") as f:
        json.dump(classified, f)
if groups_file:
    with open(groups_file, "w") as f:
        json.dump(list(groups.values()), f)
    # `git cat-file --batch` input for every member's index entry: the
    # generation check reads exactly the bytes a commit would contain.
    with open(groups_file + ".batch", "w", newline="\n") as f:
        for g in groups.values():
            for m in g["members"]:
                f.write(":" + m + "\n")

for p in final:
    print(p)
PYEOF
}

# Transcript consent comes from the one shared policy reader
# (lib/transcript-consent.ts): every merged state root, the new@ cutoff and the
# transcript_repos allowlist. Without bun nothing transcript-related can be
# checked, so transcript pages are held (fail-closed).
CONSENT_TS="$SCRIPT_DIR/../lib/transcript-consent.ts"
transcript_policy() {
  command -v bun >/dev/null 2>&1 || return 2
  GSTACK_STATE_ROOT="$GSTACK_HOME" bun run "$CONSENT_TS" "$@"
}

# 0 when staged transcript pages may sync at all.
transcripts_consented() {
  transcript_policy --check 2>/dev/null
}

# Before every push: scan the unpublished range for transcript pages outside
# the current consent; when some exist, the helper rewrites gstack-brain-sync's
# own unpublished commits without them, or refuses (exit 3) when it cannot.
# Returns 0 when the push may go ahead. A range that touches no transcript
# page needs no scan (and no bun).
guard_unpublished_range() {
  local branch range="HEAD" rc=0
  branch=$(git -C "$GSTACK_HOME" rev-parse --abbrev-ref HEAD 2>/dev/null || echo "")
  if [ -n "$branch" ] && [ "$branch" != "HEAD" ] && git -C "$GSTACK_HOME" rev-parse --verify --quiet "origin/$branch" >/dev/null 2>&1; then
    range="origin/$branch..HEAD"
  fi
  [ -n "$(git -C "$GSTACK_HOME" log --format= --name-only "$range" -- transcripts 2>/dev/null | head -1)" ] || return 0
  transcript_policy --guard-unpublished || rc=$?
  if [ "$rc" -eq 2 ]; then
    echo "BRAIN_SYNC: push blocked: unpushed commits hold transcript pages and bun is not on PATH to check them against your transcript consent" >&2
  fi
  [ "$rc" -eq 0 ]
}

# Finalize the drain: delete exactly the spool record files this drain
# consumed (per the snapshot manifest) AND positively classified. Deletion is
# EXPLICIT-DELETE-ONLY: a record is unlinked only when its path appears in
# (staged paths ∪ classified dropped). The old polarity ("delete unless
# retained") turned a missing/unparseable classification into retained=∅ and
# wiped every snapshotted record — remotely triggerable via a malformed
# pulled privacy map that raised AFTER the manifest write. Now a
# missing/unparseable class_file or paths_file deletes NOTHING (warn +
# return), and a path the classification never mentions stays 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. Unparseable records move to $QUEUE_DIR/quarantine/
# (never deleted) so they stop re-warning at every boundary. 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
  local paths_file="$3"     # staged paths (compute_paths_to_stage stdout), one per line
  # 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_DIR" "$snapshot_file" "$class_file" "$paths_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
spool_dir, snapshot_file, class_file, paths_file, drops_file = sys.argv[1:6]

def lines(path):
    try:
        with open(path) as f:
            return [l.rstrip("\r\n") for l in f if l.strip()]
    except FileNotFoundError:
        return []

# Explicit-delete-only inputs. Either input unreadable → delete NOTHING.
try:
    with open(class_file) as f:
        classified = json.load(f)
    if not isinstance(classified, dict):
        raise ValueError("classification is not an object")
except Exception:
    print("BRAIN_SYNC: warning: classification unreadable — no queue records deleted; next run re-drains", file=sys.stderr)
    sys.exit(0)
try:
    with open(paths_file) as f:
        staged = {l.strip() for l in f if l.strip()}
except Exception:
    print("BRAIN_SYNC: warning: staged-paths file unreadable — no queue records deleted; next run re-drains", file=sys.stderr)
    sys.exit(0)

dropped = set()
for group in (classified.get("dropped", {}) or {}).values():
    dropped.update(group)
deletable = staged | dropped
# A record that stays queued (privacy-held, scanner-held, waiting group) only
# needs one copy per path: every re-enqueue of such a path used to add a record
# that was never consumed, so the queue count grew without meaning anything.
kept = {}

unparseable = 0
for name in lines(snapshot_file):
    full = os.path.join(spool_dir, name)
    try:
        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:
        pass
    if not isinstance(p, str):
        # Never destroy what we can't read — but don't leave it re-warning at
        # every boundary either: move it aside for inspection.
        unparseable += 1
        try:
            qdir = os.path.join(spool_dir, "quarantine")
            os.makedirs(qdir, exist_ok=True)
            os.replace(full, os.path.join(qdir, name))
        except OSError:
            pass                 # quarantine move failed — leave in place; next run retries
        continue
    if p not in deletable:
        kept.setdefault(p, []).append(name)
        continue                 # retained / unclassified: stays queued (explicit-delete-only)
    try:
        os.unlink(full)          # staged or dropped: fully processed
    except FileNotFoundError:
        pass
def age_key(name):
    head = name.split("-", 1)[0]
    return (int(head) if head.isdigit() else 0, name)
for names in kept.values():
    for older in sorted(names, key=age_key)[:-1]:
        try:
            os.unlink(os.path.join(spool_dir, older))   # the newest record carries the path
        except OSError:
            pass
if unparseable:
    print(f"BRAIN_SYNC: {unparseable} unparseable spool record(s) moved to quarantine (inspect {os.path.join(spool_dir, 'quarantine')})", file=sys.stderr)

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
}

# Remove .git/index.lock when it is older than 10 minutes. Call only while
# holding the drain lock: every gstack writer takes that lock before running
# git here, so an old index lock is one a killed git process left behind.
# Returns 1 when a younger index lock is present (a live git process).
clear_stale_index_lock() {
  local idx="$GSTACK_HOME/.git/index.lock"
  [ -e "$idx" ] || return 0
  [ -n "$(find "$idx" -maxdepth 0 -mmin +10 2>/dev/null)" ] || return 1
  rm -f "$idx" 2>/dev/null || return 1
  ST_NOTE="$ST_NOTE; removed a stale git index lock left by an interrupted git process"
  echo "BRAIN_SYNC: removed a stale git index lock (older than 10 minutes)" >&2
  return 0
}

# Stage the paths listed in $1 with `git add -f` (forces past .gitignore=*).
# One batched add; when git refuses it, a per-path pass finds the paths it
# refuses. Rewrites $1 to the staged paths; sets ADD_FAILED and FIRST_ADD_ERR.
stage_paths() {
  local paths_file="$1" p err
  ADD_FAILED=0
  FIRST_ADD_ERR=""
  # Windows: compute_paths_to_stage's python print() emits CRLF; a trailing
  # CR makes the pathspec match nothing.
  tr -d '\r' < "$paths_file" > "$paths_file.lf"
  if git --literal-pathspecs -C "$GSTACK_HOME" add -f --pathspec-from-file=- < "$paths_file.lf" >/dev/null 2>&1; then
    mv -f "$paths_file.lf" "$paths_file"
    return 0
  fi
  : > "$paths_file"
  while IFS= read -r p; do
    p="${p%$'\r'}"
    [ -z "$p" ] && continue
    if err=$(git --literal-pathspecs -C "$GSTACK_HOME" add -f -- "$p" 2>&1 >/dev/null); then
      printf '%s\n' "$p" >> "$paths_file"
    else
      ADD_FAILED=$(( ADD_FAILED + 1 ))
      [ -z "$FIRST_ADD_ERR" ] && FIRST_ADD_ERR=$(printf '%s' "$err" | head -1)
    fi
  done < "$paths_file.lf"
  rm -f "$paths_file.lf"
}

# Plan the holds for one drain (one Python pass):
#   $1 scanner hits ("<family>\t<path>"), $2 groups JSON from
#   compute_paths_to_stage, $3 `git cat-file --batch` output for every group
#   member's index entry, $4 the staged paths.
# A flagged path is held with the rest of its staged group (its dependents).
# A group that is not held must stage whole and pass its generation check, or
# all its staged members wait. Writes the held entries (JSON) to $5, the
# paths to unstage to $6 and the waiting paths to $7.
plan_holds() {
  python3 - "$@" "$SELF_BIN" <<'PYEOF'
import json, sys
hits_path, groups_path, blobs_path, staged_path, held_out, unstage_out, wait_out, self_bin = sys.argv[1:9]

def lines(path):
    with open(path, encoding="utf-8", errors="surrogateescape") as f:
        return [l.rstrip("\r\n") for l in f if l.strip()]

hits = [tuple(l.split("\t", 1)) for l in lines(hits_path) if "\t" in l]
with open(groups_path) as f:
    groups = json.load(f)
staged = set(lines(staged_path))
with open(blobs_path, "rb") as f:
    data = f.read()

# Index blobs in batch order (None for an entry the index does not have).
blobs, pos = {}, 0
for g in groups:
    for m in g["members"]:
        nl = data.index(b"\n", pos)
        header = data[pos:nl].split(b" ")
        pos = nl + 1
        if header[-1] == b"missing":
            blobs[m] = None
            continue
        size = int(header[2])
        blobs[m] = data[pos:pos + size]
        pos += size + 1

def decisions_active(members):
    log, snap = blobs.get(members[0]), blobs.get(members[1])
    events = []
    for line in (log or b"").decode("utf-8", "replace").split("\n"):
        try:
            e = json.loads(line)
        except ValueError:
            continue
        if isinstance(e, dict):
            events.append(e)
    retired = {e.get("supersedes") for e in events if e.get("kind") in ("supersede", "redact")}
    active = {e.get("id") for e in events if e.get("kind") == "decide" and e.get("id") not in retired}
    if snap is None:
        return not active
    try:
        rows = json.loads(snap.decode("utf-8", "replace"))
    except ValueError:
        return False
    return isinstance(rows, list) and all(isinstance(r, dict) for r in rows) \
        and {r.get("id") for r in rows} == active

CHECKS = {"decisions-active": decisions_active}

hit_paths = {p for _, p in hits}
held, unstage, waiting, held_groups = [], [], [], set()
for rule, path in hits:
    group = next((g for g in groups if path in g["members"]), None)
    deps = [m for m in (group["staged"] if group else []) if m != path and m not in hit_paths and m in staged]
    if group:
        held_groups.add(group["members"][0])
    also = " and the files coupled with it" if deps else ""
    held.append({
        "path": path,
        "rule": rule,
        "dependents": deps,
        "fixes": [
            f"edit {path} so the secret scan no longer matches ({rule}); the next sync re-scans it",
            f"{self_bin} --skip-file {path}  (permanently excludes it{also} from future syncs; undo with {self_bin} --unskip-file {path})",
        ],
    })
    for m in [path] + deps:
        if m not in unstage:
            unstage.append(m)
for g in groups:
    if g["members"][0] in held_groups:
        continue
    members_staged = [m for m in g["staged"] if m in staged]
    if not members_staged:
        continue
    check = CHECKS.get(g.get("check"))
    if len(members_staged) == len(g["staged"]) and check and check(g["members"]):
        continue
    for m in members_staged:
        waiting.append(m)
        if m not in unstage:
            unstage.append(m)

with open(held_out, "w") as f:
    json.dump(held, f)
for path, rows in ((unstage_out, unstage), (wait_out, waiting)):
    with open(path, "w", encoding="utf-8", errors="surrogateescape", newline="\n") as f:
        f.writelines(r + "\n" for r in rows)
PYEOF
}

# Fail closed: unstage everything, keep the queue, status blocked.
block_drain() {
  git -C "$GSTACK_HOME" reset -q HEAD -- . >/dev/null 2>&1 || true
  ST_DRAINABLE=0
  write_status "blocked" "$1; nothing was committed and the queue is kept. Edit or skip the flagged files ($SELF_BIN --skip-file <path>), or clear the batch with $SELF_BIN --drop-queue --yes"
  echo "BRAIN_SYNC: blocked: $1" >&2
}

# Secret scan and group check for the staged paths in $1 (groups JSON in $2):
# one scanner pass attributes findings to staged paths, plan_holds decides
# what is held or waits, those paths are unstaged and leave $1 (their records
# stay queued), then the whole remaining staged diff must re-scan clean.
# Sets ST_HELD and WAITING. Returns 1 after falling back to block_drain.
hold_flagged_paths() {
  local paths_file="$1" groups_file="$2" scan="$1.scan" rc=0 gate_rc=0 gate_out
  WAITING=0
  : > "$scan.hits"
  : > "$scan.blobs"
  git -C "$GSTACK_HOME" diff --cached --no-renames > "$scan.diff" 2>/dev/null || rc=1
  git -C "$GSTACK_HOME" diff --cached --no-renames -z --name-only > "$scan.names" 2>/dev/null || rc=1
  [ "$rc" -eq 0 ] && { secret_scan attribute "$scan.diff" "$scan.names" "$scan.hits" || rc=1; }
  if [ "$rc" -eq 0 ] && [ -s "$groups_file.batch" ]; then
    tr -d '\r' < "$groups_file.batch" | git -C "$GSTACK_HOME" cat-file --batch > "$scan.blobs" 2>/dev/null || rc=1
  fi
  [ "$rc" -eq 0 ] && { plan_holds "$scan.hits" "$groups_file" "$scan.blobs" "$paths_file" "$scan.held" "$scan.unstage" "$scan.wait" || rc=1; }
  if [ "$rc" -ne 0 ]; then
    block_drain "the secret scan could not tell which staged files it flagged"
    return 1
  fi
  ST_HELD=$(cat "$scan.held")
  WAITING=$(grep -c . "$scan.wait" 2>/dev/null || true)
  WAITING=${WAITING:-0}
  if [ -s "$scan.unstage" ]; then
    if ! tr -d '\r' < "$scan.unstage" | git --literal-pathspecs -C "$GSTACK_HOME" reset -q HEAD --pathspec-from-file=- >/dev/null 2>&1; then
      block_drain "git could not unstage the held files"
      return 1
    fi
    { grep -vxF -f "$scan.unstage" "$paths_file" || true; } > "$paths_file.kept"
    mv -f "$paths_file.kept" "$paths_file"
  fi
  if ! git -C "$GSTACK_HOME" diff --cached --no-renames > "$scan.diff" 2>/dev/null; then
    block_drain "git could not read the staged diff for the re-scan"
    return 1
  fi
  gate_out=$(secret_scan gate "$scan.diff") || gate_rc=$?
  [ "$gate_rc" -eq 0 ] && return 0
  if [ "$gate_rc" -eq 3 ]; then
    block_drain "a re-scan after holding back the flagged files still matched ($gate_out)"
  else
    block_drain "the secret scan failed (exit $gate_rc)"
  fi
  return 1
}

# Write to $1 the tracked files gone from disk that the sync may remove, one
# per line: allowlisted, not transcript pages (their consent guard owns those),
# and a coupled group only when every member is gone.
removal_candidates() {
  git -C "$GSTACK_HOME" ls-files -d -z > "$1.gone" 2>/dev/null || { rm -f "$1.gone"; return 1; }
  python3 - "$GSTACK_HOME" "$ALLOWLIST" "$COUPLED_GROUPS" "$1" <<'PYEOF'
import fnmatch, json, os, re, sys
home, allowlist_path, coupled, out = sys.argv[1:5]
try:
    with open(allowlist_path) as f:
        globs = [l.strip() for l in f if l.strip() and not l.lstrip().startswith("#")]
except FileNotFoundError:
    globs = []
with open(out + ".gone", "rb") as f:
    gone = {p for p in f.read().decode("utf-8", "surrogateescape").split("\0") if p}
os.remove(out + ".gone")
def removable(p):
    if p.split("/")[0] == "transcripts" or not any(fnmatch.fnmatchcase(p, g) for g in globs):
        return False
    for spec in json.loads(coupled):
        for member in spec["members"]:
            m = re.fullmatch(re.escape(member).replace(r"\*", "([^/]+)"), p)
            if m:
                return all(not os.path.exists(os.path.join(home, x.replace("*", m.group(1), 1))) for x in spec["members"])
    return True
with open(out, "w", newline="\n", encoding="utf-8", errors="surrogateescape") as f:
    for p in sorted(gone):
        if removable(p):
            f.write(p + "\n")
PYEOF
}

# Commit the removal of the paths listed in $1 and push it, receipted like
# every push. Returns 0 pushed, 1 commit failed, 2 committed but not pushed
# (the run-start detector retries it).
publish_removals() {
  local list="$1" n
  n=$(grep -c . "$list" 2>/dev/null || true)
  git --literal-pathspecs -C "$GSTACK_HOME" -c user.email="gstack@localhost" -c user.name="gstack-brain-sync" \
    commit -q -m "sync: remove $n file(s) | $(date -u +%Y-%m-%dT%H:%M:%SZ)" --pathspec-from-file="$list" >/dev/null 2>&1 || return 1
  guard_unpublished_range || return 2
  GSTACK_HOME="$GSTACK_HOME" _receipted_git closed brain-sync "$(remote_host)" curated-memory-git-push "artifacts_sync_mode!=off" \
    bash -c 'GIT_TERMINAL_PROMPT=0 git -c http.lowSpeedLimit=1024 -c http.lowSpeedTime=30 -C "$1" push origin HEAD 2>/dev/null' _ "$GSTACK_HOME" || return 2
  date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE"
}

# The drain's removal step (artifacts_sync_removals=on). Records the outcome
# in ST_NOTE; sets REMOVALS_PUSHED when the removals reached the remote.
sync_removals() {
  local list n rc=0
  REMOVALS_PUSHED=0
  list=$(mktemp "${TMPDIR:-/tmp}/brain-sync-removals.XXXXXX") || return 0
  if ! removal_candidates "$list"; then
    rm -f "$list" "$list.gone"
    ST_NOTE="$ST_NOTE; could not list removed files; none published"
    return 0
  fi
  n=$(grep -c . "$list" 2>/dev/null || true)
  if [ "$n" -gt "$REMOVAL_VALVE" ]; then
    ST_NOTE="$ST_NOTE; $n synced file(s) are gone from disk, more than $REMOVAL_VALVE at once, so none of the removals were published (review: $SELF_BIN --publish-removals; publish: $SELF_BIN --publish-removals --yes)"
    echo "BRAIN_SYNC: holding $n removal(s), more than $REMOVAL_VALVE at once; review with $SELF_BIN --publish-removals" >&2
  elif [ "$n" -gt 0 ]; then
    publish_removals "$list" || rc=$?
    case "$rc" in
      0) REMOVALS_PUSHED=1; ST_NOTE="$ST_NOTE; published $n removal(s)" ;;
      1) ST_NOTE="$ST_NOTE; committing $n removal(s) failed; the next sync retries" ;;
      *) ST_NOTE="$ST_NOTE; $n removal(s) committed locally but not pushed" ;;
    esac
  fi
  rm -f "$list"
}

subcmd_publish_removals() {
  local confirm="${1:-}" list n rc=0
  sync_active || { echo "BRAIN_SYNC: artifacts sync is not active; nothing to publish" >&2; exit 1; }
  gstack_brain_lock_acquire "$GSTACK_HOME" try || { echo "BRAIN_SYNC: another sync is running; try again" >&2; exit 1; }
  list=$(mktemp "${TMPDIR:-/tmp}/brain-sync-removals.XXXXXX") || exit 1
  trap 'rm -f "$list" "$list.gone"; gstack_brain_lock_release' EXIT INT TERM
  removal_candidates "$list" || { echo "BRAIN_SYNC: could not list removed files" >&2; exit 1; }
  n=$(grep -c . "$list" 2>/dev/null || true)
  if [ "$n" -eq 0 ]; then
    echo "no removals to publish"
    exit 0
  fi
  if [ "$confirm" != "--yes" ]; then
    echo "$n synced file(s) are gone from disk:"
    sed 's/^/  /' "$list"
    echo "publish their removal: $SELF_BIN --publish-removals --yes"
    exit 0
  fi
  publish_removals "$list" || rc=$?
  case "$rc" in
    0) echo "published $n removal(s)" ;;
    1) echo "BRAIN_SYNC: error: committing the removals failed" >&2; exit 1 ;;
    *) echo "BRAIN_SYNC: $n removal(s) committed locally; the push failed and retries at skill start" >&2; exit 1 ;;
  esac
}

subcmd_once() {
  if ! sync_active; then
    # Silent no-op when feature not initialized / disabled.
    exit 0
  fi

  # One drain at a time, and never while another gstack writer runs git in
  # $GSTACK_HOME (bin/gstack-brain-lock.sh). A held lock means skip, not wait:
  # the next skill boundary catches up. Released on EVERY exit from here on,
  # including the empty-queue fast path and an INT during the detector's
  # network push; the mktemp block below re-traps with tempfile cleanup added.
  gstack_brain_lock_acquire "$GSTACK_HOME" try || exit 0
  trap 'gstack_brain_lock_release' 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

  # Janitor: reap orphaned enqueue temp files. A writer killed between its
  # tmp write and the atomic rename leaves `.tmp-*` behind forever — it never
  # becomes a record and nothing else touches it. One hour is far beyond any
  # live writer's write→rename window, so a fresh tmp (an in-flight enqueue)
  # is never touched. Runs inside the run lock, so it can't race the drain.
  find "$QUEUE_DIR" -maxdepth 1 -type f -name '.tmp-*' -mmin +60 -delete 2>/dev/null || true

  local mode
  mode=$("$CONFIG_BIN" get artifacts_sync_mode 2>/dev/null || echo off)

  # #2516: advance the brain worktree gbrain indexes to the artifacts repo's
  # HEAD once a day — previously it only moved when setup-gbrain / sync-gbrain
  # / brain-restore ran, so brains silently served stale code forever. Runs
  # inside THIS run lock (never concurrent with the ingest steps below) and
  # before they touch the worktree. Attempt-throttled: the stamp is written on
  # ATTEMPT, so a persistently-failing advance warns once per 24h, not at
  # every skill boundary. The advance itself refuses dirty or unmanaged
  # worktrees and never force-removes (see gstack-gbrain-source-wireup).
  if [ -e "${GSTACK_BRAIN_WORKTREE:-$HOME/.gstack-brain-worktree}" ]; then
    local adv_stamp adv_now adv_last adv_age
    adv_stamp="$GSTACK_HOME/.brain-worktree-last-advance"
    adv_now=$(date +%s)
    adv_last=$(cat "$adv_stamp" 2>/dev/null || echo 0)
    case "$adv_last" in ''|*[!0-9]*) adv_last=0 ;; esac
    adv_age=$(( adv_now - adv_last ))
    if [ "$adv_age" -ge 86400 ]; then
      echo "$adv_now" > "$adv_stamp" 2>/dev/null || true
      if ! "$SCRIPT_DIR/gstack-gbrain-source-wireup" --advance-only 1>&2; then
        echo "BRAIN_SYNC: warning: brain worktree advance failed — gbrain may be indexing stale code (run gstack-gbrain-source-wireup to repair)" >&2
      fi
    fi
  fi

  # Removals go first, so a removal commit whose push fails is one more
  # unpushed commit for the detector below.
  REMOVALS_PUSHED=0
  [ "$("$CONFIG_BIN" get artifacts_sync_removals 2>/dev/null || echo off)" = "on" ] && sync_removals

  # #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/<branch> may not exist yet (first sync, deleted remote).
  #
  # Throttled: the preamble runs --once at EVERY skill boundary, so an
  # unthrottled retry would pay a full network push attempt per boundary in
  # exactly the steady states this targets (offline, broken auth) — and a
  # captive-portal push can block 30-75s against the header's "<1s when
  # idle" promise. Attempts are recorded (success or fail) and retried at
  # most every 10 minutes; the push itself never prompts for credentials and
  # bounds stalled transfers via git's own low-speed limits (portable — stock
  # macOS ships no `timeout` binary).
  #
  # Author-scoped — EXCLUSIVELY: `git push origin HEAD` publishes every
  # unpushed commit, so the retry fires only when ALL unpushed commits are
  # gstack-brain-sync's own. One interleaved user commit disables the
  # auto-retry entirely (adversarial review: an existential check would
  # silently auto-publish a user's manual ~/.gstack commit the moment a bot
  # commit sat in front of it). User commits ride along when a REAL drain
  # pushes, as before — the detector never publishes work it didn't create.
  local det_branch det_unpushed det_total=0 det_now det_last
  det_branch=$(git -C "$GSTACK_HOME" rev-parse --abbrev-ref HEAD 2>/dev/null || echo "")
  # Detached HEAD reads as the literal "HEAD" — origin/HEAD usually resolves,
  # so without this exclusion the detector would retry a doomed push forever.
  [ "$det_branch" = "HEAD" ] && det_branch=""
  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 --author="gstack-brain-sync" "origin/$det_branch..HEAD" 2>/dev/null || echo 0)
    det_total=$(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
    case "$det_total" in ''|*[!0-9]*) det_total=0 ;; esac
    det_now=$(date +%s)
    det_last=$(cat "$GSTACK_HOME/.brain-last-push-attempt" 2>/dev/null || echo 0)
    case "$det_last" in ''|*[!0-9]*) det_last=0 ;; esac
    if [ "$det_unpushed" -gt 0 ] && [ "$det_unpushed" -eq "$det_total" ] && [ $(( det_now - det_last )) -ge 600 ] && guard_unpublished_range; then
      echo "$det_now" > "$GSTACK_HOME/.brain-last-push-attempt" 2>/dev/null || true
      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_TERMINAL_PROMPT=0 git -c http.lowSpeedLimit=1024 -c http.lowSpeedTime=30 -C "$1" push origin HEAD 2>/dev/null' _ "$GSTACK_HOME"; then
        date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE"
        det_total=0
      fi
    fi
  fi

  # Empty-queue fast path: this is the steady state at every skill boundary.
  # 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) — and so is a leftover
  # .migrating file: if its conversion failed above (e.g. python3 missing),
  # records are still pending, so "idle" would be dishonest. (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" ] && [ ! -s "$QUEUE.migrating" ]; then
    # Stranded local commits that this run did not push keep the status at
    # push_failed: an "idle" here used to erase the failure one boundary later,
    # so a push that never recovered went silent.
    if [ "$det_total" -gt 0 ]; then
      write_status "push_failed" "$det_total local commit(s) are not pushed yet; the push retries at a skill start at most every 10 minutes (commits made outside gstack-brain-sync are never auto-pushed: push them with git -C $GSTACK_HOME push origin HEAD)"
      exit 0
    fi
    if [ "$REMOVALS_PUSHED" = 1 ]; then write_status "ok" "queue empty"; else write_status "idle" "queue empty"; fi
    exit 0
  fi

  local paths_file class_file snapshot_file groups_file
  paths_file=$(mktemp "${TMPDIR:-/tmp}/brain-sync-paths.XXXXXX") || { write_status "error" "mktemp failed"; exit 1; }
  class_file=$(mktemp "${TMPDIR:-/tmp}/brain-sync-class.XXXXXX") || { rm -f "$paths_file"; write_status "error" "mktemp failed"; exit 1; }
  snapshot_file=$(mktemp "${TMPDIR:-/tmp}/brain-sync-snapshot.XXXXXX") || { rm -f "$paths_file" "$class_file"; write_status "error" "mktemp failed"; exit 1; }
  groups_file=$(mktemp "${TMPDIR:-/tmp}/brain-sync-groups.XXXXXX") || { rm -f "$paths_file" "$class_file" "$snapshot_file"; write_status "error" "mktemp failed"; exit 1; }
  # Single trap covers all: lock release AND tempfile cleanup (the siblings
  # are the staging and hold steps' scratch files).
  trap 'rm -f "$paths_file" "$paths_file".* "$class_file" "$snapshot_file" "$groups_file" "$groups_file".* 2>/dev/null; gstack_brain_lock_release' EXIT INT TERM

  # Fail-safe (G1): a classifier that dies mid-run (ENOSPC/OOM/SIGKILL, or a
  # shape the defensive filters don't cover) may have already written the
  # snapshot manifest but no classification. Finalizing on that state is what
  # used to wipe the queue — so on a nonzero exit, warn loudly, do NOT call
  # finalize_queue, and leave everything queued for the next drain.
  # Tightened consent (new@ cutoff, narrower transcript_repos): remove the
  # unpublished staged pages it now excludes before anything is staged.
  [ -d "$GSTACK_HOME/transcripts" ] && { transcript_policy --purge || true; }
  local transcripts_ok=0
  transcripts_consented && transcripts_ok=1
  if ! compute_paths_to_stage "$mode" "$class_file" "$snapshot_file" "$transcripts_ok" "$groups_file" > "$paths_file"; then
    echo "BRAIN_SYNC: warning: queue classification failed — no records consumed; next run re-drains" >&2
    write_status "error" "classification failed; queue preserved (next run retries)"
    exit 0
  fi
  if [ ! -s "$paths_file" ]; then
    # Nothing stageable. Finalize the snapshot (retained entries survive;
    # classified drops removed; records created after the snapshot untouched).
    finalize_queue "$snapshot_file" "$class_file" "$paths_file"
    local summary
    summary=$(queue_summary "$class_file")
    write_status "idle" "no stageable changes${summary:+ ($summary)}"
    exit 0
  fi
  local n_stageable
  n_stageable=$(grep -c . "$paths_file" 2>/dev/null || true)

  # Every add, reset and commit below needs the git index. The drain lock
  # proves no other gstack writer is running git here, so an index lock older
  # than 10 minutes is abandoned and is removed; a younger one is a live git
  # process (the user's own), which the drain reports and waits out.
  if ! clear_stale_index_lock; then
    ST_DRAINABLE=$n_stageable
    write_status "error" "a git index lock ($GSTACK_HOME/.git/index.lock) younger than 10 minutes shows another git process running in the artifacts repo; queue kept, the next sync retries"
    echo "BRAIN_SYNC: error: git index lock present; another git process is running in $GSTACK_HOME" >&2
    exit 0
  fi

  # Stage explicit paths only. A path git refuses to add leaves $paths_file,
  # so its record stays queued and the commit count stays honest.
  stage_paths "$paths_file"
  if [ ! -s "$paths_file" ]; then
    ST_DRAINABLE=$n_stageable
    write_status "error" "git add failed for all $ADD_FAILED path(s): ${FIRST_ADD_ERR:-unknown error}; queue preserved (next run retries)"
    echo "BRAIN_SYNC: error: git add failed for all $ADD_FAILED path(s): ${FIRST_ADD_ERR:-unknown error}" >&2
    exit 0
  fi
  if [ "$ADD_FAILED" -gt 0 ]; then
    ST_NOTE="$ST_NOTE; git add failed for $ADD_FAILED path(s), still queued (${FIRST_ADD_ERR:-unknown error})"
    echo "BRAIN_SYNC: warning: git add failed for $ADD_FAILED path(s): ${FIRST_ADD_ERR:-unknown error}" >&2
  fi

  # Secret scan: flagged files (with their coupled groups) are held back and
  # stay queued; groups whose staged generation is inconsistent wait. Nothing
  # matching a scanner pattern is ever committed: the remaining staged diff
  # must re-scan clean, or the drain unstages everything (blocked).
  hold_flagged_paths "$paths_file" "$groups_file" || exit 0
  ST_DRAINABLE=$(( ADD_FAILED + WAITING ))
  [ "$WAITING" -gt 0 ] && ST_NOTE="$ST_NOTE; $WAITING file(s) of a coupled group wait for a consistent generation (still queued)"
  local held_n ok_code="ok"
  held_n=$(printf '%s' "$ST_HELD" | grep -o '"rule"' | wc -l | tr -d ' ')
  if [ "$held_n" -gt 0 ]; then
    ok_code="held"
    ST_NOTE="$ST_NOTE; the secret scan held back $held_n flagged file(s) (and any coupled files), NOT synced; see held[] for each file's fixes"
    echo "BRAIN_SYNC: held $held_n flagged file(s); see $SELF_BIN --status" >&2
  fi

  # A verified empty staged diff is the only "nothing to commit"; anything
  # else that stops the commit keeps the queue (status error).
  local cached_rc=0
  git -C "$GSTACK_HOME" diff --cached --quiet >/dev/null 2>&1 || cached_rc=$?
  if [ "$cached_rc" -eq 0 ]; then
    finalize_queue "$snapshot_file" "$class_file" "$paths_file"
    if [ "$ok_code" = "held" ]; then
      write_status "held" "nothing else to sync"
    else
      write_status "idle" "queue drained but no new changes to commit"
    fi
    exit 0
  fi
  if [ "$cached_rc" -ne 1 ]; then
    git -C "$GSTACK_HOME" reset -q >/dev/null 2>&1 || true
    ST_DRAINABLE=$n_stageable
    write_status "error" "git could not read the staged changes (exit $cached_rc); queue kept, the next sync retries"
    exit 0
  fi

  # Egress receipt for the push, written BEFORE the commit consumes the
  # queue (amendment C7 ordering): a refused receipt exits HERE, before any
  # queue mutation or local commit, so the queue stays intact and the next
  # run retries the whole drain. Content-free: git owns the bytes
  # (sha256:null). Fail-closed.
  local push_host receipt_err
  push_host=$(remote_host)
  if ! receipt_err=$(GSTACK_HOME="$GSTACK_HOME" "$SCRIPT_DIR/gstack-egress-receipt" write \
      --sink brain-sync --host "$push_host" --class curated-memory-git-push \
      --no-payload --consent "artifacts_sync_mode!=off" 2>&1 >/dev/null); then
    write_status "push_failed" "EGRESS_RECEIPT_FAILED: receipt not writable; push refused (queue preserved)"
    _gstack_egress_refusal "brain-sync push" "$(printf '%s' "$receipt_err" | head -c 300)"
    exit 1
  fi

  # Commit with template message. A failed commit (a rejecting hook, a git
  # error) unstages, keeps the queue and says why.
  local n ts commit_err
  n=$(wc -l < "$paths_file" | tr -d ' ')
  ts=$(date -u +%Y-%m-%dT%H:%M:%SZ)
  local msg="sync: $n file(s) | $ts"
  if ! commit_err=$(git -C "$GSTACK_HOME" -c user.email="gstack@localhost" -c user.name="gstack-brain-sync" \
      commit -q -m "$msg" 2>&1 >/dev/null); then
    git -C "$GSTACK_HOME" reset -q >/dev/null 2>&1 || true
    ST_DRAINABLE=$(( ST_DRAINABLE + n ))
    write_status "error" "git commit failed: $(printf '%s' "$commit_err" | head -1 | head -c 200); queue kept, the next sync retries (check $GSTACK_HOME/.git/hooks)"
    echo "BRAIN_SYNC: error: git commit failed" >&2
    exit 0
  fi

  # Nothing outside the transcript consent is pushed; a refused rewrite keeps the commit local.
  guard_unpublished_range || { write_status "push_failed" "push blocked: unpushed commits hold transcript pages outside your transcript consent; commit retained locally"; finalize_queue "$snapshot_file" "$class_file" "$paths_file"; exit 0; }
  # Push. On reject, fetch + merge (merge driver handles JSONL) + retry once.
  local push_err
  push_err=$(git -C "$GSTACK_HOME" push origin HEAD 2>&1 >/dev/null) || {
    # Check if this is an auth error first — no point retrying.
    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; commit retained locally, will retry next run. fix: $hint"
      echo "BRAIN_SYNC: push failed: auth. fix: $hint" >&2
      # Drained records leave the spool — they live in the local commit, which
      # the run-start detector re-pushes next time (#2549). Retained +
      # post-snapshot records survive the finalize.
      finalize_queue "$snapshot_file" "$class_file" "$paths_file"
      exit 0
    fi

    # Try a fetch-and-merge + retry. The fetch and the retry push are their
    # own attempted-egress ops, each receipted fail-closed (a refusal falls
    # through to the push_failed path below).
    if GSTACK_HOME="$GSTACK_HOME" _receipted_git closed brain-sync "$push_host" curated-memory-git-fetch "artifacts_sync_mode!=off" \
        bash -c 'git -C "$1" fetch origin 2>/dev/null' _ "$GSTACK_HOME"; then
      local branch
      branch=$(git -C "$GSTACK_HOME" rev-parse --abbrev-ref HEAD 2>/dev/null || echo main)
      if git -C "$GSTACK_HOME" merge --no-edit "origin/$branch" >/dev/null 2>&1 && guard_unpublished_range; 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
          finalize_queue "$snapshot_file" "$class_file" "$paths_file"
          date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE"
          write_status "$ok_code" "pushed $n file(s) after rebase"
          exit 0
        fi
      fi
    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"
    finalize_queue "$snapshot_file" "$class_file" "$paths_file"
    exit 0
  }

  # Success: drained records leave the spool (retained + post-snapshot survive).
  finalize_queue "$snapshot_file" "$class_file" "$paths_file"
  date -u +%Y-%m-%dT%H:%M:%SZ > "$LAST_PUSH_FILE"
  write_status "$ok_code" "pushed $n file(s)"
  exit 0
}

subcmd_status() {
  if [ -f "$STATUS_FILE" ]; then
    cat "$STATUS_FILE"
  else
    echo '{"status":"unknown","message":"no status file yet"}'
  fi
  # Supplemental info (not in status file). Depth = spool record files plus
  # any not-yet-migrated legacy queue lines (transition window), including a
  # crash-leftover .migrating file — its records are still pending too.
  local queue_depth spool_depth legacy_depth
  # find, not `ls "$QUEUE_DIR"/*.json`: a deep spool overflows the argv limit
  # and the glob form then reported 0.
  spool_depth=$(find "$QUEUE_DIR" -maxdepth 1 -type f -name '*.json' 2>/dev/null | wc -l | tr -d ' ')
  legacy_depth=0
  [ -f "$QUEUE" ] && legacy_depth=$(wc -l < "$QUEUE" | tr -d ' ')
  [ -f "$QUEUE.migrating" ] && legacy_depth=$(( legacy_depth + $(wc -l < "$QUEUE.migrating" | 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
  mode=$("$CONFIG_BIN" get artifacts_sync_mode 2>/dev/null || echo off)
  printf '{"queue_depth":%s,"last_push":"%s","mode":"%s"}\n' "$queue_depth" "$last_push" "$mode"
}

subcmd_skip_file() {
  local path="${1:-}"
  if [ -z "$path" ]; then
    echo "Usage: $SELF_BIN --skip-file <path>" >&2
    exit 1
  fi
  mkdir -p "$GSTACK_HOME"
  # Avoid duplicate entries.
  if [ -f "$SKIP_FILE" ] && grep -Fxq "$path" "$SKIP_FILE"; then
    echo "already in skip list: $path"
    exit 0
  fi
  echo "$path" >> "$SKIP_FILE"
  echo "added to skip list: $path"
  echo "(permanent: it no longer syncs, nor do the files of any coupled group it belongs to; queued records drop on the next --once; undo with $SELF_BIN --unskip-file $path)"
}

subcmd_unskip_file() {
  local path="${1:-}"
  if [ -z "$path" ]; then
    echo "Usage: $SELF_BIN --unskip-file <path>" >&2
    exit 1
  fi
  if [ ! -f "$SKIP_FILE" ] || ! grep -Fxq "$path" "$SKIP_FILE"; then
    echo "not in skip list: $path"
    exit 0
  fi
  local kept
  kept=$(mktemp "$SKIP_FILE.XXXXXX") || { echo "could not update $SKIP_FILE" >&2; exit 1; }
  { grep -Fxv "$path" "$SKIP_FILE" || true; } > "$kept"
  mv -f "$kept" "$SKIP_FILE"
  echo "removed from skip list: $path"
  if [ -f "$GSTACK_HOME/$path" ]; then
    "$SCRIPT_DIR/gstack-brain-enqueue" "$path"
    echo "(queued; the next sync publishes it, together with any coupled group it belongs to)"
  fi
}

# Skill start's attention lines: one per row of the sync status table in this
# script's header that applies. Every value read from the status file is
# clamped (status code to the table's closed set, counts to digits,
# timestamps to their shape); the message, held paths and scanner output never
# reach this output, only fixed text and this script's absolute path.
subcmd_attention() {
  sync_active || exit 0
  local line="" prefix code="" held="" drainable="" last_drain="" last_push="" bin="$SELF_BIN"
  if [ -f "$STATUS_FILE" ]; then
    line=$(head -c 65536 "$STATUS_FILE" 2>/dev/null | tr -d '\r\n')
    # The counts and timestamps precede every free-text field.
    prefix=${line%%, \"held\": *}
    code=$(printf '%s' "$prefix" | sed -n 's/^{"status": *"\([a-z_]*\)".*/\1/p')
    held=$(printf '%s' "$prefix" | sed -n 's/.*"held_count": *\([0-9][0-9]*\).*/\1/p')
    drainable=$(printf '%s' "$prefix" | sed -n 's/.*"drainable": *\([0-9][0-9]*\).*/\1/p')
    last_drain=$(printf '%s' "$prefix" | sed -n 's/.*"last_drain_at": *"\([^"]*\)".*/\1/p')
    last_push=$(printf '%s' "$prefix" | sed -n 's/.*"last_push_at": *"\([^"]*\)".*/\1/p')
    case "$code" in ok|idle|held|blocked|push_failed|error) ;; *) code="unknown" ;; esac
  fi
  case "$held" in ''|*[!0-9]*) held=0 ;; esac
  case "$drainable" in ''|*[!0-9]*) drainable=0 ;; esac
  local iso='[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9]Z'
  case "$last_drain" in $iso) ;; *) last_drain="" ;; esac
  case "$last_push" in $iso) ;; *) last_push="" ;; esac

  case "$code" in
    held) echo "ARTIFACTS_SYNC: attention: status=held held=$held. The secret scan is holding back $held file(s); everything else still syncs. See which files and how to fix them: $bin --status" ;;
    blocked) echo "ARTIFACTS_SYNC: attention: status=blocked. The secret scan flagged a file it could not hold back on its own, so nothing is syncing. See the fix: $bin --status" ;;
    push_failed) echo "ARTIFACTS_SYNC: attention: status=push_failed. Synced files are committed locally but the push failed; it retries at skill start. See the cause: $bin --status" ;;
    error) echo "ARTIFACTS_SYNC: attention: status=error. The last sync stopped before committing; the queue is kept and retried. See the cause: $bin --status" ;;
    unknown) echo "ARTIFACTS_SYNC: attention: status=unknown. The sync status file is unreadable. See it: $bin --status" ;;
  esac

  local now cutoff oldest="" oldest_epoch
  now=$(date +%s 2>/dev/null || echo "")
  case "$now" in ''|*[!0-9]*) return 0 ;; esac
  cutoff=$(date -u -d "@$(( now - 86400 ))" +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || date -u -r "$(( now - 86400 ))" +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo "")
  [ -n "$cutoff" ] || return 0
  oldest=$(find "$QUEUE_DIR" -maxdepth 1 -type f -name '*.json' 2>/dev/null | sed 's#.*/##' | sort | head -1)
  if [ "$drainable" -gt 0 ]; then
    oldest_epoch=${oldest%%-*}
    case "$oldest_epoch" in ''|*[!0-9]*) oldest_epoch=$now ;; esac
    if { [ -n "$last_push" ] && [[ "$last_push" < "$cutoff" ]]; } \
        || { [ -z "$last_push" ] && [ "$oldest_epoch" -lt $(( now - 86400 )) ]; }; then
      echo "ARTIFACTS_SYNC: attention: stale-push. Files are ready to sync but nothing has been pushed for over 24 hours. See why: $bin --status"
    fi
  fi
  if [ -n "$oldest" ] || [ -s "$QUEUE" ] || [ -s "$QUEUE.migrating" ]; then
    if [ -z "$last_drain" ] || [[ "$last_drain" < "$cutoff" ]]; then
      echo "ARTIFACTS_SYNC: attention: stale-drain. The sync has not run for over 24 hours while files wait in the queue. See why: $bin --status"
    fi
  fi
  return 0
}

subcmd_drop_queue() {
  local force="${1:-}"
  if [ "$force" != "--yes" ]; then
    echo "Refusing: --drop-queue discards pending syncs. Pass --yes to confirm." >&2
    exit 1
  fi
  # Remove spool record files, then truncate any legacy queue remnant —
  # including a crash-leftover .migrating file, whose records would otherwise
  # resurrect on the next drain via migrate_legacy_queue after the user
  # explicitly discarded the queue.
  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 [ -f "$QUEUE.migrating" ]; then
    local mig_n
    mig_n=$(wc -l < "$QUEUE.migrating" | tr -d ' ')
    n=$(( n + mig_n ))
    rm -f "$QUEUE.migrating" 2>/dev/null || true
  fi
  if [ "$n" -eq 0 ]; then
    echo "queue already empty"
    exit 0
  fi
  echo "dropped $n queue entries"
}

subcmd_discover_new() {
  if ! sync_active; then
    exit 0
  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, time
from datetime import datetime, timezone

gstack_home, allowlist_path, cursor_path = sys.argv[1:4]
spool_dir = os.path.join(gstack_home, ".brain-queue.d")
skip_path = os.path.join(gstack_home, ".brain-skip.txt")

def load_lines(path):
    try:
        with open(path) as f:
            return [l.strip() for l in f if l.strip() and not l.lstrip().startswith("#")]
    except FileNotFoundError:
        return []

def load_cursor(path):
    try:
        with open(path) as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {}

def save_cursor(path, data):
    try:
        with open(path, "w") as f:
            json.dump(data, f)
    except OSError:
        pass

allowlist = load_lines(allowlist_path)
# Normalize skip entries to the same POSIX form as `rel` below, so a
# backslash entry in .brain-skip.txt still matches a normalized path on Windows.
skip = {s.replace(os.sep, "/") for s in load_lines(skip_path)}
cursor = load_cursor(cursor_path)
new_cursor = dict(cursor)
to_enqueue = []

# Walk all files under gstack_home, match against allowlist.
for root, dirs, files in os.walk(gstack_home):
    # Skip .git and .brain-* state files.
    if ".git" in root.split(os.sep):
        continue
    for name in files:
        full = os.path.join(root, name)
        # Repo paths are POSIX-relative. os.path.relpath yields backslash
        # separators on Windows, which never match the forward-slash allowlist
        # globs (e.g. "projects/*/learnings.jsonl"), so discovery silently
        # enqueued nothing under projects/ on Windows. Normalize to "/".
        rel = os.path.relpath(full, gstack_home).replace(os.sep, "/")
        if rel.startswith(".brain-"):
            continue
        if not any(fnmatch.fnmatchcase(rel, pat) for pat in allowlist):
            continue
        if rel in skip:
            continue
        try:
            st = os.stat(full)
            key = f"{int(st.st_mtime)}:{st.st_size}"
        except OSError:
            continue
        if cursor.get(rel) != key:
            to_enqueue.append((rel, key))

# 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 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 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:
        # 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 written:
        new_cursor[rel] = key

save_cursor(cursor_path, new_cursor)
PYEOF
}

# -------- dispatch --------
case "${1:-}" in
  --once|"") subcmd_once ;;
  --status) subcmd_status ;;
  --skip-file) shift; subcmd_skip_file "${1:-}" ;;
  --unskip-file) shift; subcmd_unskip_file "${1:-}" ;;
  --drop-queue) shift; subcmd_drop_queue "${1:-}" ;;
  --discover-new) subcmd_discover_new ;;
  --attention) subcmd_attention ;;
  --publish-removals) shift; subcmd_publish_removals "${1:-}" ;;
  --help|-h)
    awk 'NR > 1 && /^# Heredoc delivery guard/ { exit } NR > 1' "$0" | sed 's/^# \{0,1\}//'
    ;;
  *)
    echo "Unknown subcommand: $1" >&2
    echo "Run: gstack-brain-sync --help" >&2
    exit 1
    ;;
esac