fix(ingestion): bound the tripwire's alerts; test removal guards; deflake race tests

PR #160 review:
- The planter chooses the paths, and one alert listed them all, while the
  queue retried oldest-first forever: a message ntfy refuses (too large)
  blocked every later alert (2000 links: 0 alerts delivered), and an
  outage queued one message per removal, unbounded. Alerts now name at
  most 10 paths and count the rest; the queue holds at most 5, further
  alerts become one suppression summary; a message failing 5 times is
  dropped so it cannot block the rest.
- removeSymlinks' Lstat re-check had no test: a direct test passes a
  regular file, a path through an in-brain dir link, a ../ escape and a
  missing path, and asserts all fail with nothing deleted. A Watch test
  covers dir and in-brain targets. root.Remove (not RemoveAll) is
  commented: it refuses a dir swapped in after the Lstat.
- Every queued alert is delivered after an outage (truncating the queue
  after the first send is now red).
- hyperguild#162: race tests did not wait for their swapper goroutines
  before t.TempDir cleanup ("directory not empty" in CI run 8912). Each
  swapper is now stopped and awaited in a t.Cleanup.

Fixes #162
Refs #157

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

ingestion/internal/api/raw_race_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"syscall"
 	"testing"
@@ -25,6 +26,18 @@ import (
 
 const raceSentinel = "SENTINEL-153-RACE"
 
+// startSwapper runs fn until the test ends, and waits for it to exit before
+// t.TempDir's cleanup removes the directory it writes into (hyperguild#162:
+// an unawaited swapper raced the cleanup, "directory not empty").
+func startSwapper(t *testing.T, fn func(stop *atomic.Bool)) {
+	t.Helper()
+	var stop atomic.Bool
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() { defer wg.Done(); fn(&stop) }()
+	t.Cleanup(func() { stop.Store(true); wg.Wait() })
+}
+
 func swap(raw, outside string, fifo bool, stop *atomic.Bool) {
 	tgt := filepath.Join(raw, "r.md")
 	for i := 0; !stop.Load(); i++ {
@@ -76,9 +89,7 @@ func readers(brain, raw string) string {
 
 func TestRawReadersLeakNothingUnderASymlinkSwapRace(t *testing.T) {
 	brain, raw, outside := raceBrain(t)
-	var stop atomic.Bool
-	go swap(raw, outside, false, &stop)
-	defer stop.Store(true)
+	startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, false, stop) })
 	leaks, iters := 0, 0
 	for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ {
 		leaks += strings.Count(readers(brain, raw), raceSentinel)
@@ -91,8 +102,7 @@ func TestRawReadersLeakNothingUnderASymlinkSwapRace(t *testing.T) {
 
 func TestRawReadersDoNotHangUnderAFIFOSwapRace(t *testing.T) {
 	brain, raw, outside := raceBrain(t)
-	var stop atomic.Bool
-	go swap(raw, outside, true, &stop)
+	startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, true, stop) })
 	var iters atomic.Int64
 	done := make(chan struct{})
 	go func() {
@@ -104,9 +114,7 @@ func TestRawReadersDoNotHangUnderAFIFOSwapRace(t *testing.T) {
 	}()
 	select {
 	case <-done:
-		stop.Store(true)
 	case <-time.After(6 * time.Second):
-		stop.Store(true)
 		// Unblock the stuck open so the test process can exit.
 		if f, err := os.OpenFile(filepath.Join(raw, "r.md"), os.O_WRONLY|syscall.O_NONBLOCK, 0); err == nil {
 			_ = f.Close()
@@ -147,8 +155,7 @@ func TestDocmarkExtractionLeaksNothingUnderASymlinkSwapRace(t *testing.T) {
 	if err := os.WriteFile(tgt, []byte("inside docx bytes"), 0o644); err != nil {
 		t.Fatal(err)
 	}
-	var stop atomic.Bool
-	go func() {
+	startSwapper(t, func(stop *atomic.Bool) {
 		for i := 0; !stop.Load(); i++ {
 			tmp := filepath.Join(raw, ".swapdocx")
 			_ = os.Remove(tmp)
@@ -159,8 +166,7 @@ func TestDocmarkExtractionLeaksNothingUnderASymlinkSwapRace(t *testing.T) {
 			}
 			_ = os.Rename(tmp, tgt)
 		}
-	}()
-	defer stop.Store(true)
+	})
 	iters := 0
 	for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ {
 		_, _ = api.ReadIngestFile(brain, tgt)

ingestion/internal/symlinkguard/symlinkguard.go

@@ -23,6 +23,14 @@ import (
 // DefaultInterval is used when Watch is given a non-positive interval.
 const DefaultInterval = 30 * time.Second
 
+// Alert bounds (PR #160 review): the planter chooses the paths, so neither an
+// alert's size nor the queue may grow with what was planted.
+const (
+	maxPathsPerAlert = 10 // further paths are counted, not listed
+	maxPending       = 5  // queued alerts; the rest become one summary
+	maxAttempts      = 5  // a message failing this often is dropped
+)
+
 // Notifier is the out-of-band alert sink (audit.Notifier satisfies it).
 type Notifier interface {
 	Notify(ctx context.Context, msg string) error
@@ -52,13 +60,22 @@ func (r Result) String() string {
 	return strings.Join(parts, "; ")
 }
 
-// quoted joins paths with each one strconv-quoted.
+// quoted joins at most maxPathsPerAlert paths, each strconv-quoted, and
+// counts the rest.
 func (Result) quoted(xs []string) string {
-	out := make([]string, len(xs))
-	for i, x := range xs {
+	shown := xs
+	if len(shown) > maxPathsPerAlert {
+		shown = shown[:maxPathsPerAlert]
+	}
+	out := make([]string, len(shown))
+	for i, x := range shown {
 		out[i] = strconv.Quote(x)
 	}
-	return strings.Join(out, ", ")
+	s := strings.Join(out, ", ")
+	if extra := len(xs) - len(shown); extra > 0 {
+		s += fmt.Sprintf(" and %d more", extra)
+	}
+	return s
 }
 
 // Scan walks the brain dir and reports every symlink in it and every
@@ -121,7 +138,7 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
 	}
 	t := time.NewTicker(interval)
 	defer t.Stop()
-	var pending []string
+	q := &alertQueue{}
 	reported := ""
 	for {
 		select {
@@ -136,10 +153,9 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
 		} else {
 			removed, failed := removeSymlinks(brainDir, r.Symlinks)
 			if len(removed) > 0 {
-				msg := fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s",
-					len(removed), Result{Symlinks: removed}.quoted(removed))
-				slog.Error("brain symlink tripwire removed symlinks", "paths", removed)
-				pending = append(pending, msg)
+				slog.Error("brain symlink tripwire removed symlinks", "count", len(removed))
+				q.push(fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s",
+					len(removed), Result{}.quoted(removed)))
 			}
 			key = Result{Symlinks: failed, Unreadable: r.Unreadable}.String()
 		}
@@ -147,21 +163,59 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
 			reported = key
 			if key != "" {
 				slog.Error("brain symlink tripwire", "finding", key)
-				pending = append(pending, "brain symlink tripwire (hyperguild#157): "+key+
+				q.push("brain symlink tripwire (hyperguild#157): " + key +
 					". Readers may follow symlinks out of the brain; fix them.")
 			}
 		}
+		q.flush(ctx, n)
+	}
+}
+
+// alertQueue holds alerts until Notify accepts them, bounded: at most
+// maxPending queued, further ones counted into one summary, and a message
+// that fails maxAttempts times dropped so it cannot block the rest.
+type alertQueue struct {
+	items      []queued
+	suppressed int
+}
+
+type queued struct {
+	msg      string
+	attempts int
+}
+
+func (q *alertQueue) push(msg string) {
+	if len(q.items) >= maxPending {
+		q.suppressed++
+		return
+	}
+	q.items = append(q.items, queued{msg: msg})
+}
+
+func (q *alertQueue) flush(ctx context.Context, n Notifier) {
 	if n == nil {
-			pending = nil
+		q.items, q.suppressed = nil, 0
+		return
+	}
+	for len(q.items) > 0 {
+		head := &q.items[0]
+		if err := n.Notify(ctx, head.msg); err != nil {
+			head.attempts++
+			if head.attempts >= maxAttempts {
+				slog.Error("symlink tripwire alert dropped after repeated failures", "attempts", head.attempts, "err", err)
+				q.items = q.items[1:]
 				continue
 			}
-		for len(pending) > 0 {
-			if err := n.Notify(ctx, pending[0]); err != nil {
 			slog.Warn("symlink tripwire notify failed; will retry", "err", err)
-				break
+			return
 		}
-			pending = pending[1:]
+		q.items = q.items[1:]
 	}
+	if q.suppressed > 0 {
+		q.items = append(q.items, queued{msg: fmt.Sprintf(
+			"brain symlink tripwire (hyperguild#157): %d further alert(s) were suppressed while the notifier was failing; check the ingestion logs", q.suppressed)})
+		q.suppressed = 0
+		q.flush(ctx, n)
 	}
 }
 
@@ -187,6 +241,9 @@ func removeSymlinks(brainDir string, rels []string) (removed, failed []string) {
 			failed = append(failed, rel)
 			continue
 		}
+		// root.Remove, not os.RemoveAll: if the link is swapped for a
+		// non-empty directory after the Lstat, Remove refuses rather than
+		// deleting it recursively, and it cannot act outside the root.
 		if err := root.Remove(rel); err != nil {
 			failed = append(failed, rel)
 			continue

ingestion/internal/symlinkguard/symlinkguard_test.go

@@ -5,7 +5,10 @@ import (
 	"errors"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -191,7 +194,7 @@ func TestWatchReportsASetAgainAfterItWasCleared(t *testing.T) {
 // hyperguild#157 decision: the runtime tripwire removes a symlink it finds
 // (the link, never its target), so it is readable for at most one interval
 // instead of until a human acts, and the alert says what was removed.
-func TestWatchRemovesSymlinksButNeverTheirTargetsOrRegularFiles(t *testing.T) {
+func TestWatchRemovesAnOutsideFileLinkButNotItsTargetOrRegularFiles(t *testing.T) {
 	brain := t.TempDir()
 	outside := filepath.Join(t.TempDir(), "target.md")
 	require.NoError(t, os.WriteFile(outside, []byte("keep me"), 0o644))
@@ -226,3 +229,177 @@ func TestWatchRemovesSymlinksButNeverTheirTargetsOrRegularFiles(t *testing.T) {
 func TestDefaultIntervalIsThirtySeconds(t *testing.T) {
 	assert.Equal(t, 30*time.Second, DefaultInterval)
 }
+
+// PR #160 review: the Lstat re-check and the os.Root confinement in
+// removeSymlinks had no test. Anything that is not a symlink inside the brain
+// is reported failed and nothing is deleted.
+func TestRemoveSymlinksRefusesNonSymlinksEscapesAndPathsThroughLinks(t *testing.T) {
+	brain := t.TempDir()
+	require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "real"), 0o755))
+	regular := filepath.Join(brain, "wiki", "note.md")
+	require.NoError(t, os.WriteFile(regular, []byte("note"), 0o644))
+	inLinked := filepath.Join(brain, "wiki", "real", "x.md")
+	require.NoError(t, os.WriteFile(inLinked, []byte("x"), 0o644))
+	require.NoError(t, os.Symlink(filepath.Join(brain, "wiki", "real"), filepath.Join(brain, "wiki", "linkdir")))
+	outside := filepath.Join(filepath.Dir(brain), "outside-157.md")
+	require.NoError(t, os.WriteFile(outside, []byte("o"), 0o644))
+	t.Cleanup(func() { _ = os.Remove(outside) })
+
+	removed, failed := removeSymlinks(brain, []string{
+		"wiki/note.md", "wiki/linkdir/x.md", "../outside-157.md", "wiki/missing.md",
+	})
+	assert.Empty(t, removed)
+	assert.ElementsMatch(t, []string{"wiki/note.md", "wiki/linkdir/x.md", "../outside-157.md", "wiki/missing.md"}, failed)
+	for _, p := range []string{regular, inLinked, outside} {
+		_, err := os.Stat(p)
+		assert.NoError(t, err, "%s must not be deleted", p)
+	}
+}
+
+// Dir targets and in-brain targets survive too, not only an outside file.
+func TestWatchRemovesDirAndInBrainLinksButNotTheirTargets(t *testing.T) {
+	brain := t.TempDir()
+	outsideDir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(outsideDir, "keep.md"), []byte("keep"), 0o644))
+	require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
+	inBrain := filepath.Join(brain, "wiki", "target.md")
+	require.NoError(t, os.WriteFile(inBrain, []byte("t"), 0o644))
+	require.NoError(t, os.Symlink(outsideDir, filepath.Join(brain, "wiki", "dirlink")))
+	require.NoError(t, os.Symlink(inBrain, filepath.Join(brain, "wiki", "inlink.md")))
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 10*time.Millisecond, nil)
+	require.Eventually(t, func() bool {
+		r, err := Scan(brain)
+		return err == nil && len(r.Symlinks) == 0
+	}, 2*time.Second, 10*time.Millisecond)
+	for _, p := range []string{filepath.Join(outsideDir, "keep.md"), inBrain} {
+		_, err := os.Stat(p)
+		assert.NoError(t, err, "%s must survive", p)
+	}
+}
+
+// PR #160 review: the planter chooses the paths. An alert names at most
+// maxPathsPerAlert of them and counts the rest, so its size stays bounded.
+func TestWatchAlertNamesBoundedPathsAndCountsTheRest(t *testing.T) {
+	brain := t.TempDir()
+	for i := 0; i < 50; i++ {
+		require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, strings.Repeat("n", 100)+strconv.Itoa(i))))
+	}
+	n := &recNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 10*time.Millisecond, n)
+	require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 10*time.Millisecond)
+	n.mu.Lock()
+	msg := n.msgs[0]
+	n.mu.Unlock()
+	assert.Contains(t, msg, "removed 50 symlink(s)")
+	assert.Contains(t, msg, "and 40 more")
+	assert.Less(t, len(msg), 2500, "alert size is bounded regardless of how many links were planted")
+}
+
+// rejectNotifier refuses every message containing a marker, forever.
+type rejectNotifier struct {
+	recNotifier
+	reject string
+}
+
+func (r *rejectNotifier) Notify(ctx context.Context, m string) error {
+	if strings.Contains(m, r.reject) {
+		r.mu.Lock()
+		r.msgs = append(r.msgs, "REJECTED")
+		r.mu.Unlock()
+		return errors.New("413")
+	}
+	return r.recNotifier.Notify(ctx, m)
+}
+
+// PR #160 review: one message the server will never accept blocked every
+// later alert forever (head-of-line). It is dropped after maxAttempts.
+func TestWatchDropsAMessageThatNeverSendsSoLaterAlertsFlow(t *testing.T) {
+	brain := t.TempDir()
+	n := &rejectNotifier{reject: "POISON"}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 10*time.Millisecond, n)
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "POISON")))
+	time.Sleep(30 * time.Millisecond)
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "later")))
+	require.Eventually(t, func() bool {
+		n.mu.Lock()
+		defer n.mu.Unlock()
+		for _, m := range n.msgs {
+			if strings.Contains(m, "later") {
+				return true
+			}
+		}
+		return false
+	}, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message")
+}
+
+// toggleNotifier fails while down is set.
+type toggleNotifier struct {
+	recNotifier
+	down atomic.Bool
+}
+
+func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
+	if r.down.Load() {
+		return errors.New("ntfy down")
+	}
+	return r.recNotifier.Notify(ctx, m)
+}
+
+// PR #160 review: an outage queued one message per removal, unbounded. The
+// queue holds at most maxPending messages; the rest become one summary.
+func TestWatchQueueIsBoundedDuringAnOutage(t *testing.T) {
+	brain := t.TempDir()
+	n := &toggleNotifier{}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 5*time.Millisecond, n)
+	for i := 0; i < 30; i++ {
+		link := filepath.Join(brain, "p"+strconv.Itoa(i))
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
+	}
+	n.down.Store(false)
+	require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 5*time.Millisecond)
+	time.Sleep(100 * time.Millisecond)
+	assert.LessOrEqual(t, n.count(), maxPending+1, "queued alerts plus at most one suppression summary")
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	found := false
+	for _, m := range n.msgs {
+		if strings.Contains(m, "suppressed") {
+			found = true
+		}
+	}
+	assert.True(t, found, "suppressed alerts are counted in a summary")
+}
+
+// Within the bound, every alert queued during an outage is delivered after
+// it, each naming its own removal (PR #160 review: truncating the queue after
+// the first send stayed green).
+func TestWatchDeliversEveryQueuedAlertAfterAnOutage(t *testing.T) {
+	brain := t.TempDir()
+	n := &toggleNotifier{}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 5*time.Millisecond, n)
+	for _, name := range []string{"qa", "qb", "qc"} {
+		link := filepath.Join(brain, name)
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
+	}
+	n.down.Store(false)
+	require.Eventually(t, func() bool { return n.count() == 3 }, 2*time.Second, 5*time.Millisecond)
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	for i, name := range []string{"qa", "qb", "qc"} {
+		assert.Contains(t, n.msgs[i], strconv.Quote(name))
+	}
+}

ingestion/internal/watcher/watcher_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"syscall"
 	"testing"
@@ -343,7 +344,12 @@ func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) {
 	brainDir := setupBrainDir(t)
 	raw := filepath.Join(brainDir, "raw")
 	var stop atomic.Bool
+	var wg sync.WaitGroup
+	wg.Add(1)
+	// Awaited before t.TempDir's cleanup (hyperguild#162).
+	t.Cleanup(func() { stop.Store(true); wg.Wait() })
 	go func() {
+		defer wg.Done()
 		for i := 0; !stop.Load(); i++ {
 			tmp := filepath.Join(raw, ".swap")
 			_ = os.Remove(tmp)