fix(ingestion): tripwire alerts survive outages; drop only permanent rejections

PR #160 round 2: dropping a message after 5 failures could not tell a
poison message from an ntfy outage. A 15-minute outage dropped the only
incident with nothing reported, and drops were not counted (74 of 300
removals were neither delivered nor counted).

- audit.NtfyNotifier returns *RejectedError for non-2xx; Permanent() is
  true for 4xx other than 429.
- The tripwire queue drops a message only on a permanent rejection and
  counts it; every other failure is an outage and is retried, however
  long. Each flush tries every queued message, so an undeliverable one
  cannot block the rest.
- Overflow and drops go into one counter, sent as a summary retried
  every tick and never dropped. Every removal is delivered or counted.

Tests (red first): an outage of ~100 ticks still delivers the only
incident; delivered + counted equals the number of removals; a
permanently rejected message is dropped and counted. The short-outage
test is renamed to say it is short. The Watch doc no longer claims
'loses none' unconditionally.

Refs #157

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

ingestion/cmd/server/main.go

@@ -33,6 +33,7 @@ import (
 	"github.com/mathiasbq/hyperguild/ingestion/internal/reranker"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/search"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/startupretry"
+	"github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/watcher"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/webhook"
@@ -173,7 +174,8 @@ func alertingWarnings(getenv func(string) string) []string {
 	var out []string
 	if getenv("BRAIN_NTFY_URL") == "" {
 		out = append(out, "alerting DISABLED: BRAIN_NTFY_URL unset — "+
-			"brain_edit_root governance rewrites and capture audit degradation will not notify")
+			"brain_edit_root governance rewrites, capture audit degradation and the "+
+			"brain symlink tripwire (hyperguild#157) will not notify")
 	}
 	return out
 }
@@ -239,6 +241,10 @@ func main() {
 	}
 
 	brainDir := envOr("INGEST_BRAIN_DIR", "../brain")
+	if err := startupSymlinkCheck(brainDir); err != nil {
+		logger.Error("brain symlink tripwire", "err", err)
+		os.Exit(1)
+	}
 	port := envOr("INGEST_PORT", "3300")
 
 	llmURL := envOr("INGEST_LLM_URL", "http://iguana:4000/v1")
@@ -360,6 +366,8 @@ func main() {
 	}
 
 	ctx := context.Background()
+	// hyperguild#157: report a symlink that appears in the brain while running.
+	go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier())
 
 	// Built before the watcher, the graph backfill and session ingestion, and
 	// retried rather than fatal (#83). Two separate reasons:

ingestion/cmd/server/symlink_check.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard"
+)
+
+// startupSymlinkCheck is the hyperguild#157 tripwire at boot: a symlink in the
+// brain dir would let brain readers serve a file from outside the brain, so
+// the server does not start until it is removed. A directory the scan cannot
+// read could hide one, so it refuses on that too. A missing brain dir (first
+// boot) is the only case it lets pass.
+func startupSymlinkCheck(brainDir string) error {
+	r, err := symlinkguard.Scan(brainDir)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return nil
+		}
+		return fmt.Errorf("refusing to start: cannot scan the brain dir %s (hyperguild#157): %w", brainDir, err)
+	}
+	if !r.Clean() {
+		return fmt.Errorf("refusing to start: brain dir %s has %s (hyperguild#157); fix them first", brainDir, r)
+	}
+	return nil
+}
+
+// symlinkScanInterval reads BRAIN_SYMLINK_SCAN_INTERVAL (seconds), falling
+// back to the default on an unset, non-numeric or non-positive value.
+func symlinkScanInterval(getenv func(string) string) time.Duration {
+	if s, err := strconv.Atoi(getenv("BRAIN_SYMLINK_SCAN_INTERVAL")); err == nil && s > 0 {
+		return time.Duration(s) * time.Second
+	}
+	return symlinkguard.DefaultInterval
+}

ingestion/cmd/server/symlink_check_test.go

@@ -0,0 +1,65 @@
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// hyperguild#157: the server refuses to start while a symlink exists in the
+// brain dir, or while any part of it cannot be scanned (which could hide
+// one); a missing brain dir is the only thing it lets pass.
+func TestStartupSymlinkCheckRefusesABrainWithASymlink(t *testing.T) {
+	brain := t.TempDir()
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "planted")))
+	err := startupSymlinkCheck(brain)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "planted")
+}
+
+func TestStartupSymlinkCheckRefusesABrainWithAnUnreadableDir(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root reads mode-000 directories")
+	}
+	brain := t.TempDir()
+	locked := filepath.Join(brain, "aaa")
+	require.NoError(t, os.MkdirAll(locked, 0o755))
+	require.NoError(t, os.Chmod(locked, 0o000))
+	t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
+	err := startupSymlinkCheck(brain)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "aaa")
+}
+
+func TestStartupSymlinkCheckPassesACleanOrMissingBrain(t *testing.T) {
+	brain := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(brain, "n.md"), []byte("x"), 0o644))
+	assert.NoError(t, startupSymlinkCheck(brain))
+	assert.NoError(t, startupSymlinkCheck(filepath.Join(brain, "missing")))
+}
+
+func TestSymlinkScanIntervalFallsBackOnNonPositive(t *testing.T) {
+	for _, v := range []string{"0", "-5", "", "junk"} {
+		got := symlinkScanInterval(func(string) string { return v })
+		assert.Positive(t, got, "env %q", v)
+	}
+}
+
+func TestAlertingWarningsNamesTheSymlinkTripwire(t *testing.T) {
+	w := strings.Join(alertingWarnings(func(string) string { return "" }), " ")
+	assert.Contains(t, w, "symlink tripwire")
+}
+
+func TestStartupSymlinkCheckRefusesAnUnlistableBrainRoot(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root lists mode-0100 directories")
+	}
+	brain := t.TempDir()
+	require.NoError(t, os.Chmod(brain, 0o100))
+	t.Cleanup(func() { _ = os.Chmod(brain, 0o755) })
+	assert.Error(t, startupSymlinkCheck(brain))
+}

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
+	startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, false, stop) })
-	go swap(raw, outside, false, &stop)
-	defer stop.Store(true)
 	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
+	startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, true, stop) })
-	go 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
+	startSwapper(t, func(stop *atomic.Bool) {
-	go func() {
 		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/audit/ntfy.go

@@ -49,7 +49,19 @@ func (n *NtfyNotifier) Notify(ctx context.Context, msg string) error {
 	}
 	defer func() { _ = resp.Body.Close() }()
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return fmt.Errorf("ntfy notify: status %d", resp.StatusCode)
+		return &RejectedError{Status: resp.StatusCode}
 	}
 	return nil
 }
+
+// RejectedError is a non-2xx ntfy response.
+type RejectedError struct{ Status int }
+
+func (e *RejectedError) Error() string { return fmt.Sprintf("ntfy notify: status %d", e.Status) }
+
+// Permanent reports whether resending the same message cannot succeed: a 4xx
+// other than 429 (rate limited). Callers may drop such a message; anything
+// else (5xx, 429, transport errors) is an outage and must be retried.
+func (e *RejectedError) Permanent() bool {
+	return e.Status >= 400 && e.Status < 500 && e.Status != http.StatusTooManyRequests
+}

ingestion/internal/audit/ntfy_rejected_test.go

@@ -0,0 +1,27 @@
+package audit
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A non-2xx ntfy response is a *RejectedError, and only a client-side refusal
+// other than rate limiting is permanent: callers may drop a message on a
+// permanent rejection and must retry the rest (hyperguild#157, PR #160).
+func TestNotifyClassifiesRejectionsAsPermanentOrNot(t *testing.T) {
+	for status, permanent := range map[int]bool{413: true, 400: true, 403: true, 429: false, 500: false, 503: false} {
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(status) }))
+		err := NewNtfyNotifier(srv.URL, "").Notify(context.Background(), "m")
+		srv.Close()
+		var rej *RejectedError
+		require.True(t, errors.As(err, &rej), "status %d: %v", status, err)
+		assert.Equal(t, status, rej.Status)
+		assert.Equal(t, permanent, rej.Permanent(), "status %d", status)
+	}
+}

ingestion/internal/symlinkguard/symlinkguard.go

@@ -0,0 +1,264 @@
+// Package symlinkguard is the hyperguild#157 tripwire. The brain directory is
+// a Syncthing sendreceive folder and Syncthing syncs symlinks; several brain
+// readers (search, vectorstore, context, graph) follow a symlink out of the
+// brain and would index or return whatever it points to. Until every reader
+// reads through an os.Root, a symlink inside the brain, or a directory the
+// scan cannot read (which could hide one), is an incident: the server refuses
+// to start on one (main), and Watch reports one that appears while it runs.
+package symlinkguard
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/fs"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// 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
+)
+
+// Notifier is the out-of-band alert sink (audit.Notifier satisfies it).
+type Notifier interface {
+	Notify(ctx context.Context, msg string) error
+}
+
+// Result is what a scan found: symlinks, and directories it could not read
+// (each of which could hide a symlink). Paths are brain-relative.
+type Result struct {
+	Symlinks   []string
+	Unreadable []string
+}
+
+// Clean reports whether the scan found nothing to act on.
+func (r Result) Clean() bool { return len(r.Symlinks) == 0 && len(r.Unreadable) == 0 }
+
+// String describes the findings for a log line or alert. Paths are quoted, so
+// a name containing a newline cannot blur two findings together.
+func (r Result) String() string {
+	q := r.quoted
+	var parts []string
+	if len(r.Symlinks) > 0 {
+		parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks)))
+	}
+	if len(r.Unreadable) > 0 {
+		parts = append(parts, fmt.Sprintf("%d unreadable dir(s): %s", len(r.Unreadable), q(r.Unreadable)))
+	}
+	return strings.Join(parts, "; ")
+}
+
+// quoted joins at most maxPathsPerAlert paths, each strconv-quoted, and
+// counts the rest.
+func (Result) quoted(xs []string) string {
+	shown := xs
+	if len(shown) > maxPathsPerAlert {
+		shown = shown[:maxPathsPerAlert]
+	}
+	out := make([]string, len(shown))
+	for i, x := range shown {
+		out[i] = strconv.Quote(x)
+	}
+	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
+// directory it could not read, continuing past both. Only the ROOT .git is
+// skipped (it is brain-sync's, and no reader walks it); a nested .git is
+// scanned like any other directory. brainDir itself may be a symlink (a
+// volume mount); it is resolved first. The error is non-nil when the root
+// cannot be resolved (wrapping os.ErrNotExist when it is missing) or cannot be
+// listed.
+func Scan(brainDir string) (Result, error) {
+	root, err := filepath.EvalSymlinks(brainDir)
+	if err != nil {
+		return Result{}, fmt.Errorf("resolve brain dir: %w", err)
+	}
+	var r Result
+	rootGit := filepath.Join(root, ".git")
+	walkErr := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
+		rel, _ := filepath.Rel(root, p)
+		rel = filepath.ToSlash(rel)
+		if err != nil {
+			if p == root {
+				return err
+			}
+			r.Unreadable = append(r.Unreadable, rel)
+			if d != nil && d.IsDir() {
+				return filepath.SkipDir
+			}
+			return nil
+		}
+		if d.IsDir() && p == rootGit {
+			return filepath.SkipDir
+		}
+		if d.Type()&os.ModeSymlink != 0 {
+			r.Symlinks = append(r.Symlinks, rel)
+		}
+		return nil
+	})
+	sort.Strings(r.Symlinks)
+	sort.Strings(r.Unreadable)
+	if walkErr != nil {
+		// Only the root's own error reaches here (deeper errors are recorded
+		// as Unreadable). A root that cannot be listed cannot be shown clean
+		// (PR #159 round 2: discarding this reported clean while search
+		// served a planted symlink).
+		return r, fmt.Errorf("scan brain dir: %w", walkErr)
+	}
+	return r, nil
+}
+
+// Watch rescans every interval (DefaultInterval if non-positive) until ctx
+// ends. Every symlink found is removed (the link, never its target:
+// hyperguild#157 decision), so one is readable for at most one interval.
+// Each removal is an alert; a persistent finding (a link it could not remove,
+// an unreadable dir, a scan error) alerts once per distinct state. Alerts
+// queue until Notify succeeds (see alertQueue): an outage delays them, and
+// any that do not fit the bounded queue, or are undeliverable, are counted in
+// a summary that is never dropped, so every incident is delivered or counted.
+func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
+	if interval <= 0 {
+		interval = DefaultInterval
+	}
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	q := &alertQueue{}
+	reported := ""
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+		}
+		r, err := Scan(brainDir)
+		key := ""
+		if err != nil {
+			key = "cannot scan the brain dir: " + err.Error()
+		} else {
+			removed, failed := removeSymlinks(brainDir, r.Symlinks)
+			if len(removed) > 0 {
+				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()
+		}
+		if key != reported {
+			reported = key
+			if key != "" {
+				slog.Error("brain symlink tripwire", "finding", 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. It is bounded (at most
+// maxPending queued) and loses nothing from the record: every overflow and
+// every dropped message increments suppressed, sent as a summary that is
+// retried every tick and never dropped.
+//
+// A message is dropped only when the notifier rejects it permanently (an
+// error with Permanent() == true, e.g. ntfy 413). Every other failure is an
+// outage, and nothing is dropped however long it lasts (PR #160 round 2:
+// dropping after N raw failures lost the only incident of a long outage).
+// Each flush tries every queued message, so one undeliverable message cannot
+// block the others.
+type alertQueue struct {
+	items      []string
+	suppressed int
+}
+
+func (q *alertQueue) push(msg string) {
+	if len(q.items) >= maxPending {
+		q.suppressed++
+		return
+	}
+	q.items = append(q.items, msg)
+}
+
+func (q *alertQueue) flush(ctx context.Context, n Notifier) {
+	if n == nil {
+		q.items, q.suppressed = nil, 0
+		return
+	}
+	kept := q.items[:0]
+	for _, msg := range q.items {
+		err := n.Notify(ctx, msg)
+		switch {
+		case err == nil:
+		case permanent(err):
+			slog.Error("symlink tripwire alert rejected permanently; counted in the summary", "err", err)
+			q.suppressed++
+		default:
+			slog.Warn("symlink tripwire notify failed; will retry", "err", err)
+			kept = append(kept, msg)
+		}
+	}
+	q.items = kept
+	if q.suppressed > 0 {
+		msg := fmt.Sprintf("brain symlink tripwire (hyperguild#157): %d further alert(s) were suppressed or undeliverable; check the ingestion logs", q.suppressed)
+		if err := n.Notify(ctx, msg); err == nil {
+			q.suppressed = 0
+		}
+	}
+}
+
+// permanent reports whether err says the message itself can never be sent.
+func permanent(err error) bool {
+	var p interface{ Permanent() bool }
+	return errors.As(err, &p) && p.Permanent()
+}
+
+// removeSymlinks unlinks each brain-relative symlink through an os.Root on
+// the brain dir, after re-checking it is still a symlink. Removing a symlink
+// never touches its target. It returns what was removed and what was not.
+func removeSymlinks(brainDir string, rels []string) (removed, failed []string) {
+	if len(rels) == 0 {
+		return nil, nil
+	}
+	resolved, err := filepath.EvalSymlinks(brainDir)
+	if err != nil {
+		return nil, rels
+	}
+	root, err := os.OpenRoot(resolved)
+	if err != nil {
+		return nil, rels
+	}
+	defer func() { _ = root.Close() }()
+	for _, rel := range rels {
+		fi, err := root.Lstat(rel)
+		if err != nil || fi.Mode()&os.ModeSymlink == 0 {
+			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
+		}
+		removed = append(removed, rel)
+	}
+	return removed, failed
+}

ingestion/internal/symlinkguard/symlinkguard_test.go

@@ -0,0 +1,498 @@
+package symlinkguard
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// hyperguild#157: the brain dir is a Syncthing sendreceive folder and
+// Syncthing syncs symlinks; several readers follow them out of the brain.
+// Until every reader goes through an os.Root, a symlink anywhere in the brain,
+// or a directory the scan cannot read (which could hide one), is an incident.
+
+func TestScanReportsSymlinksAtEveryDepthIncludingNestedGit(t *testing.T) {
+	brain := t.TempDir()
+	outside := t.TempDir()
+	require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "w", "facts", ".git"), 0o755))
+	require.NoError(t, os.MkdirAll(filepath.Join(brain, ".git", "objects"), 0o755))
+	require.NoError(t, os.Symlink(filepath.Join(outside, "f"), filepath.Join(brain, "wiki", "w", "facts", "leak.md")))
+	require.NoError(t, os.Symlink(outside, filepath.Join(brain, "wiki", "dirlink")))
+	// PR #159 review: only the ROOT .git is skipped; a nested .git is not
+	// special to the brain readers, which walk into it.
+	require.NoError(t, os.Symlink(filepath.Join(outside, "g"), filepath.Join(brain, "wiki", "w", "facts", ".git", "leak.md")))
+	require.NoError(t, os.Symlink(outside, filepath.Join(brain, ".git", "objects", "alt")))
+
+	r, err := Scan(brain)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"wiki/w/facts/leak.md", "wiki/dirlink", "wiki/w/facts/.git/leak.md"}, r.Symlinks)
+	assert.Empty(t, r.Unreadable)
+}
+
+// PR #159 review: WalkDir stopped at the first unreadable directory and the
+// startup check passed, while search (which keeps walking) served a planted
+// symlink. Scan now records the unreadable dir and keeps going.
+func TestScanReportsUnreadableDirsAndKeepsWalkingPastThem(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root reads mode-000 directories")
+	}
+	brain := t.TempDir()
+	locked := filepath.Join(brain, "aaa")
+	require.NoError(t, os.MkdirAll(locked, 0o755))
+	require.NoError(t, os.Chmod(locked, 0o000))
+	t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "zzz-planted")))
+
+	r, err := Scan(brain)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"aaa"}, r.Unreadable)
+	assert.Equal(t, []string{"zzz-planted"}, r.Symlinks, "the walk continues past the unreadable dir")
+}
+
+func TestScanResolvesASymlinkedRootAndReportsSymlinkedEntriesInsideIt(t *testing.T) {
+	real := t.TempDir()
+	link := filepath.Join(t.TempDir(), "brain")
+	require.NoError(t, os.Symlink(real, link))
+	require.NoError(t, os.WriteFile(filepath.Join(real, "n.md"), []byte("x"), 0o644))
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(real, "inner")))
+	r, err := Scan(link)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"inner"}, r.Symlinks)
+}
+
+func TestScanOfAMissingRootIsErrNotExist(t *testing.T) {
+	_, err := Scan(filepath.Join(t.TempDir(), "nope"))
+	assert.True(t, errors.Is(err, os.ErrNotExist), "got %v", err)
+}
+
+type recNotifier struct {
+	mu   sync.Mutex
+	msgs []string
+	fail int // fail the first N calls
+}
+
+func (r *recNotifier) Notify(_ context.Context, m string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.msgs = append(r.msgs, m)
+	if r.fail > 0 {
+		r.fail--
+		return errors.New("ntfy down")
+	}
+	return nil
+}
+
+func (r *recNotifier) count() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.msgs) }
+
+func TestWatchNotifiesOncePerDistinctSetAndAgainForANewSet(t *testing.T) {
+	brain := t.TempDir()
+	n := &recNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 10*time.Millisecond, n)
+
+	time.Sleep(50 * time.Millisecond)
+	assert.Zero(t, n.count(), "a clean brain raises nothing")
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "first")))
+	require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond)
+	time.Sleep(60 * time.Millisecond)
+	assert.Equal(t, 1, n.count(), "the same set is reported once, not every tick")
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "second")))
+	require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond)
+}
+
+// PR #159 review: last was recorded before Notify, so an alert lost to an
+// ntfy outage was never retried while the symlink stayed.
+func TestWatchRetriesAnAlertThatFailedToSend(t *testing.T) {
+	brain := t.TempDir()
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "planted")))
+	n := &recNotifier{fail: 2}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 10*time.Millisecond, n)
+	require.Eventually(t, func() bool { return n.count() == 3 }, 2*time.Second, 10*time.Millisecond)
+	time.Sleep(60 * time.Millisecond)
+	assert.Equal(t, 3, n.count(), "after one successful send, no more")
+}
+
+func TestWatchAlertsOnAnUnreadableDir(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root reads mode-000 directories")
+	}
+	brain := t.TempDir()
+	locked := filepath.Join(brain, "locked")
+	require.NoError(t, os.MkdirAll(locked, 0o755))
+	require.NoError(t, os.Chmod(locked, 0o000))
+	t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
+	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)
+	assert.Contains(t, n.msgs[0], "locked")
+}
+
+// PR #159 review: a non-positive interval panicked time.NewTicker inside the
+// goroutine and crashlooped the server.
+func TestWatchToleratesANonPositiveInterval(t *testing.T) {
+	brain := t.TempDir()
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { Watch(ctx, brain, 0, nil); close(done) }()
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Watch did not return on cancel")
+	}
+}
+
+// PR #159 round 2: a brain root that can be entered but not listed (mode
+// 0100) made WalkDir fail at the root; the error was discarded and Scan
+// reported clean, while search (which goes straight into wiki/) still served
+// a planted symlink. A root that cannot be listed is now an error.
+func TestScanOfAnUnlistableRootIsAnError(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root lists mode-0100 directories")
+	}
+	brain := t.TempDir()
+	require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "wiki", "leak.md")))
+	require.NoError(t, os.Chmod(brain, 0o100))
+	t.Cleanup(func() { _ = os.Chmod(brain, 0o755) })
+	_, err := Scan(brain)
+	require.Error(t, err)
+	assert.False(t, errors.Is(err, os.ErrNotExist), "an unlistable root is not a missing root")
+}
+
+// The same set that disappears and comes back is reported again.
+func TestWatchReportsASetAgainAfterItWasCleared(t *testing.T) {
+	brain := t.TempDir()
+	link := filepath.Join(brain, "planted")
+	n := &recNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 10*time.Millisecond, n)
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond)
+	// The tripwire removed it; planting it again is a new event.
+	require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 10*time.Millisecond)
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond)
+}
+
+// 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 TestWatchRemovesAnOutsideFileLinkButNotItsTargetOrRegularFiles(t *testing.T) {
+	brain := t.TempDir()
+	outside := filepath.Join(t.TempDir(), "target.md")
+	require.NoError(t, os.WriteFile(outside, []byte("keep me"), 0o644))
+	require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
+	link := filepath.Join(brain, "wiki", "leak.md")
+	require.NoError(t, os.Symlink(outside, link))
+	regular := filepath.Join(brain, "wiki", "note.md")
+	require.NoError(t, os.WriteFile(regular, []byte("note"), 0o644))
+
+	n := &recNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 10*time.Millisecond, n)
+
+	require.Eventually(t, func() bool {
+		_, err := os.Lstat(link)
+		return os.IsNotExist(err)
+	}, 2*time.Second, 10*time.Millisecond, "the symlink is removed")
+	require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 10*time.Millisecond)
+	b, err := os.ReadFile(outside)
+	require.NoError(t, err)
+	assert.Equal(t, "keep me", string(b), "the target is untouched")
+	b, err = os.ReadFile(regular)
+	require.NoError(t, err)
+	assert.Equal(t, "note", string(b), "regular files are untouched")
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	assert.Contains(t, n.msgs[0], "removed")
+	assert.Contains(t, n.msgs[0], "wiki/leak.md")
+}
+
+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
+}
+
+// permanentErr is a rejection the notifier says can never succeed (like ntfy 413).
+type permanentErr struct{}
+
+func (permanentErr) Error() string   { return "413 payload too large" }
+func (permanentErr) Permanent() bool { return true }
+
+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 permanentErr{}
+	}
+	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 TestWatchDeliversEveryQueuedAlertAfterAShortOutage(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))
+	}
+}
+
+// PR #160 round 2: dropping after N failures could not tell a poison message
+// from an outage, so a 15-minute outage silently lost the only incident. A
+// message is only dropped when it keeps failing while others send; through
+// an outage nothing is dropped, however long.
+func TestWatchLosesNoAlertThroughALongOutage(t *testing.T) {
+	brain := t.TempDir()
+	n := &toggleNotifier{}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 2*time.Millisecond, n)
+	link := filepath.Join(brain, "only-incident")
+	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)
+	time.Sleep(200 * time.Millisecond) // ~100 ticks of outage, far past any attempt limit
+	n.down.Store(false)
+	require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 5*time.Millisecond)
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	assert.Contains(t, n.msgs[0], strconv.Quote("only-incident"))
+}
+
+// Every removal is either delivered or counted in the suppression summary:
+// none disappears from the record (round 2: 74 of 300 were in neither).
+func TestWatchAccountsForEveryRemovalDuringAnOutage(t *testing.T) {
+	brain := t.TempDir()
+	n := &toggleNotifier{}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 2*time.Millisecond, n)
+	const plants = 25
+	for i := 0; i < plants; i++ {
+		link := filepath.Join(brain, "acct"+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, 1*time.Millisecond)
+	}
+	n.down.Store(false)
+	require.Eventually(t, func() bool {
+		n.mu.Lock()
+		defer n.mu.Unlock()
+		for _, m := range n.msgs {
+			if strings.Contains(m, "suppressed") {
+				return true
+			}
+		}
+		return false
+	}, 2*time.Second, 5*time.Millisecond)
+	time.Sleep(50 * time.Millisecond)
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	delivered, counted := 0, 0
+	for _, m := range n.msgs {
+		if strings.Contains(m, "removed 1 symlink(s)") {
+			delivered++
+		}
+		if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
+			_, _ = fmt.Sscanf(m[i+3:], "%d", &counted)
+		}
+	}
+	assert.Equal(t, plants, delivered+counted, "delivered %d + counted %d must equal %d removals", delivered, counted, plants)
+}
+
+// A poison message is still dropped (others send while it fails), and the
+// drop is counted in the summary.
+func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
+	brain := t.TempDir()
+	n := &rejectNotifier{reject: "POISON"}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 5*time.Millisecond, n)
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "POISON")))
+	time.Sleep(20 * time.Millisecond)
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "fine")))
+	require.Eventually(t, func() bool {
+		n.mu.Lock()
+		defer n.mu.Unlock()
+		for _, m := range n.msgs {
+			if strings.Contains(m, "1 further alert(s)") {
+				return true
+			}
+		}
+		return false
+	}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
+}

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)