fix(ingestion): build /mcp's bearer and the metrics wrap inside buildMux (#154) (#183)

Co-authored-by: Mathias <mthbqv@gmail.com>

ingestion/cmd/server/jwt_validator_test.go

@@ -81,3 +81,31 @@ func TestBuildMuxAnswers503ToAJWTWhileTheIssuerIsUnresolved(t *testing.T) {
 	code, _ := do(t, mux, "GET", "/pending", "Bearer aaa.bbb.ccc")
 	assert.Equal(t, http.StatusServiceUnavailable, code)
 }
+
+// PR #183 review: /mcp shares REST's bearer, but only the static token was
+// asserted on it. With the issuer unresolved, a JWT on /mcp answers 503 too,
+// so /mcp's middleware was built with the JWT validator.
+func TestBuildMuxAnswers503ToAJWTOnMCPWhileTheIssuerIsUnresolved(t *testing.T) {
+	srv, _ := fakeIssuer(t, 1000)
+	v, err := buildJWTValidator(context.Background(), slog.New(slog.DiscardHandler), srv.URL, "brain")
+	require.NoError(t, err)
+	mux, err := buildMux(restAuth{token: routesTok, validator: v, realm: "brain"}, testHandler(t),
+		okHandler("mcp"), metrics.New(), nil)
+	require.NoError(t, err)
+	code, _ := do(t, mux, "POST", "/mcp", "Bearer aaa.bbb.ccc")
+	assert.Equal(t, http.StatusServiceUnavailable, code)
+}
+
+// PR #183 review: a 401 on /mcp names the protected-resource metadata in
+// WWW-Authenticate (what the claude.ai OAuth connector discovers from), so
+// /mcp's middleware was built with resourceMetadataURL.
+func TestBuildMuxMCP401CarriesResourceMetadataInWWWAuthenticate(t *testing.T) {
+	const rm = "https://brain.example/.well-known/oauth-protected-resource"
+	mux, err := buildMux(restAuth{token: routesTok, realm: "brain", resourceMetadataURL: rm}, testHandler(t),
+		okHandler("mcp"), metrics.New(), nil)
+	require.NoError(t, err)
+	rec := httptest.NewRecorder()
+	mux.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/mcp", nil))
+	require.Equal(t, http.StatusUnauthorized, rec.Code)
+	assert.Contains(t, rec.Header().Get("WWW-Authenticate"), `resource_metadata="`+rm+`"`)
+}

ingestion/cmd/server/main.go

@@ -587,8 +587,10 @@ func main() {
 	//
 	// The whole router, /mcp's bearer included, is built in buildMux so its
 	// tests are tests of production wiring (PR #152 review, hyperguild#154).
-	// The REST API's credential is the same as /mcp's. ListenAndServe below
-	// must serve its result unchanged: that one line is the untested part.
+	// The REST API's credential is the same as /mcp's. Untested here: the
+	// restAuth fields and the public map this call passes in (e.g. a nil
+	// validator would drop JWT auth on /mcp and REST and the suite stays
+	// green), and ListenAndServe serving the result unchanged (hyperguild#182).
 	handler, err := buildMux(restAuth{token: mcpToken, validator: jwtValidator, realm: "brain",
 		resourceMetadataURL: resourceMetadataURL}, h, mcpSrv, metrics.New(), public)
 	if err != nil {

ingestion/cmd/server/routes.go

@@ -41,8 +41,9 @@ type restAuth struct {
 // wrapped in reg's middleware. Other public handlers are registered as given,
 // and only for patterns on the publicRoutes allowlist. It returns an
 // http.Handler, not the *ServeMux, so nothing can register a route on the
-// result. What stays untested is main's own line: a handler main built around
-// this one before ListenAndServe would not be seen by these tests.
+// result. What stays untested is main's call site: the restAuth fields and
+// the public map main passes in, and ListenAndServe serving this handler
+// unchanged (hyperguild#182). The tests build their own arguments.
 func buildMux(
 	auth restAuth, h *api.Handler, mcpHandler http.Handler, reg *metrics.Registry, public map[string]http.Handler,
 ) (http.Handler, error) {
@@ -92,8 +93,12 @@ func mountREST(mux *http.ServeMux, h *api.Handler, auth func(http.Handler) http.
 // labelRESTPattern sets the request's Pattern to the REST route it addresses
 // before auth runs, so the metrics middleware (which reads Pattern from the
 // same *http.Request) labels a refused request by its route rather than by
-// the "/" mount (hyperguild#155). A path no route matches keeps "/", so the
-// label set stays bounded by the registered routes.
+// the "/" mount (hyperguild#155). A refused request to a path no route
+// matches keeps "/", so for refused requests the label set stays bounded by
+// the registered routes. That bound does not hold for authenticated
+// requests: an unregistered path reaches the REST mux, which 404s with an
+// empty Pattern, and the metrics middleware falls back to the raw URL path
+// (hyperguild#185).
 func labelRESTPattern(rest *http.ServeMux, next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if _, pattern := rest.Handler(r); pattern != "" {

ingestion/cmd/server/routes_observability_test.go

@@ -11,20 +11,41 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// wrongTokSentinel is a presented credential distinctive enough that any
+// occurrence in a log line is the raw token, not a coincidence.
+const wrongTokSentinel = "wrong-tok-sentinel-q7zx"
+
 // hyperguild#155: since #149 every REST route sits behind BearerMiddleware,
 // and chassis v0.2.0 refused without logging, so an unknown caller got
-// silent 401s. The chassis now writes an `mcp auth rejected` line.
-func TestBuildMuxLogsARESTAuthRejection(t *testing.T) {
+// silent 401s. The chassis now writes an `mcp auth rejected` line, and the
+// presented token appears nowhere in the log as a bare value (PR #181
+// review: asserting on the "Bearer " header form could never fail, since
+// the chassis logs fields, not headers).
+func TestBuildMuxRejectionLogLineOmitsTheBarePresentedToken(t *testing.T) {
 	var buf bytes.Buffer
 	prev := slog.Default()
 	slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
 	t.Cleanup(func() { slog.SetDefault(prev) })
 
-	code, _ := do(t, testMux(t), "POST", "/query", "Bearer wrong")
+	code, _ := do(t, testMux(t), "POST", "/query", "Bearer "+wrongTokSentinel)
 	require.Equal(t, http.StatusUnauthorized, code)
 	assert.Contains(t, buf.String(), "mcp auth rejected")
 	assert.Contains(t, buf.String(), "path=/query")
-	assert.NotContains(t, buf.String(), "Bearer wrong", "the credential itself is never logged")
+	assert.NotContains(t, buf.String(), wrongTokSentinel, "the rejected token is logged as a bare value")
+	assert.NotContains(t, buf.String(), routesTok, "the valid static token is logged")
+}
+
+// The `mcp auth accepted` line leaves out the valid static token.
+func TestBuildMuxAcceptedLogLineOmitsTheBareStaticToken(t *testing.T) {
+	var buf bytes.Buffer
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+
+	code, _ := do(t, testMux(t), "GET", "/pending", "Bearer "+routesTok)
+	require.Equal(t, http.StatusOK, code)
+	assert.Contains(t, buf.String(), "mcp auth accepted")
+	assert.NotContains(t, buf.String(), routesTok, "the accepted static token is logged as a bare value")
 }
 
 // An authenticated REST request is labelled with the REST mux's pattern:

ingestion/cmd/server/routes_test.go

@@ -22,11 +22,13 @@ func okHandler(body string) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(body)) })
 }
 
-// testMux builds the router exactly as main does: buildMux with the full
-// public set main registers (stub handlers), a stub MCP server, and real
-// bearer config. hyperguild#149: the REST API is authenticated as a whole;
-// the PR #152 review showed a test of a COPY of the router let main.go
-// regress unseen.
+// testMux builds the router through buildMux, the function main calls, with
+// every allowlisted public pattern (stub handlers), a stub MCP server and a
+// static token only: no JWT validator and no resourceMetadataURL (the tests
+// in jwt_validator_test.go pass those). It does not use main's own restAuth
+// or public map (hyperguild#182). hyperguild#149: the REST API is
+// authenticated as a whole; the PR #152 review showed a test of a COPY of
+// the router let main.go regress unseen.
 func testMux(t *testing.T) http.Handler {
 	t.Helper()
 	mux, _ := testMuxAndRegistry(t)
@@ -87,8 +89,9 @@ func TestBuildMuxAuthenticatesAnUnregisteredPath(t *testing.T) {
 }
 
 // buildMux registers public handlers as given, without adding auth: each
-// public route owns its own (bearer on /mcp, HMAC on the webhook, client
-// secret on /oauth/token) or is meant to be open (/metrics, OAuth metadata).
+// public route owns its own (bearer/JWT on /capture, HMAC on the webhook,
+// client secret on /oauth/token) or is meant to be open (OAuth metadata).
+// /mcp and /metrics are not public routes: buildMux builds them itself.
 func TestBuildMuxDoesNotWrapPublicRoutes(t *testing.T) {
 	r := testMux(t)
 	for p := range publicRoutes {

ingestion/internal/startupretry/startupretry.go

@@ -1,75 +0,0 @@
-// Package startupretry retries a startup dependency that may not be serving
-// yet, instead of exiting on its first failure.
-//
-// It exists for the cold-boot ordering race (#83): on a cluster restart the
-// ingestion pod can reach OIDC discovery before dex is answering, gets a 502
-// from ingress, and dies — after having already connected postgres, backfilled
-// the graph and ingested session batches, all of which the restart redoes.
-// Nothing is misconfigured in that scenario; the pod simply started first.
-//
-// The budget is bounded on purpose. Retrying forever would turn a genuine
-// misconfiguration (wrong issuer URL, bad audience) into a process that looks
-// healthy and never serves, which is harder to diagnose than a crash loop.
-package startupretry
-
-import (
-	"context"
-	"fmt"
-	"time"
-)
-
-// Policy bounds the retry: at most Attempts calls, with the delay doubling
-// from Base and capped at Max.
-type Policy struct {
-	Attempts int           // total calls, including the first; < 1 means 1
-	Base     time.Duration // delay before the second attempt
-	Max      time.Duration // ceiling for the doubling delay
-}
-
-// Do calls fn until it succeeds, the attempt budget is spent, or ctx ends.
-//
-// The returned error wraps the last failure, so callers keep the diagnosis
-// they would have had without the retry.
-func Do[T any](ctx context.Context, p Policy, fn func(context.Context) (T, error)) (T, error) {
-	var zero T
-
-	attempts := p.Attempts
-	if attempts < 1 {
-		attempts = 1
-	}
-
-	delay := p.Base
-	var lastErr error
-
-	for i := range attempts {
-		if err := ctx.Err(); err != nil {
-			return zero, fmt.Errorf("startupretry: %w", err)
-		}
-
-		v, err := fn(ctx)
-		if err == nil {
-			return v, nil
-		}
-		lastErr = err
-
-		if i == attempts-1 {
-			break
-		}
-
-		if delay <= 0 {
-			delay = time.Second
-		}
-		if p.Max > 0 && delay > p.Max {
-			delay = p.Max
-		}
-
-		select {
-		case <-ctx.Done():
-			return zero, fmt.Errorf("startupretry: %w", ctx.Err())
-		case <-time.After(delay):
-		}
-		delay *= 2
-	}
-
-	return zero, fmt.Errorf("startupretry: gave up after %d attempts: %w", attempts, lastErr)
-}

ingestion/internal/startupretry/startupretry_test.go

@@ -1,142 +0,0 @@
-// ingestion/internal/startupretry/startupretry_test.go
-package startupretry
-
-import (
-	"context"
-	"errors"
-	"testing"
-	"time"
-)
-
-func fastPolicy(attempts int) Policy {
-	return Policy{Attempts: attempts, Base: time.Millisecond, Max: 2 * time.Millisecond}
-}
-
-func TestDo_ReturnsImmediatelyOnFirstSuccess(t *testing.T) {
-	calls := 0
-	got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
-		calls++
-		return "ready", nil
-	})
-	if err != nil {
-		t.Fatalf("Do: %v", err)
-	}
-	if got != "ready" {
-		t.Errorf("got %q, want %q", got, "ready")
-	}
-	if calls != 1 {
-		t.Errorf("calls = %d, want 1", calls)
-	}
-}
-
-func TestDo_SurvivesTransientFailuresThenSucceeds(t *testing.T) {
-	// The observed cold-boot race: dex is not serving yet, ingress answers
-	// 502 for the first few attempts, then discovery works.
-	calls := 0
-	got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
-		calls++
-		if calls < 4 {
-			return "", errors.New("oidc discovery: status 502")
-		}
-		return "ready", nil
-	})
-	if err != nil {
-		t.Fatalf("Do returned %v — a transient 502 must not be fatal", err)
-	}
-	if got != "ready" {
-		t.Errorf("got %q, want %q", got, "ready")
-	}
-	if calls != 4 {
-		t.Errorf("calls = %d, want 4", calls)
-	}
-}
-
-func TestDo_GivesUpAfterTheAttemptBudgetAndReportsTheLastError(t *testing.T) {
-	// A genuinely broken issuer must still fail — retrying forever would turn
-	// a config error into a silent hang, which is worse than crash-looping.
-	last := errors.New("oidc discovery: status 404")
-	calls := 0
-
-	_, err := Do(context.Background(), fastPolicy(3), func(context.Context) (string, error) {
-		calls++
-		return "", last
-	})
-
-	if !errors.Is(err, last) {
-		t.Fatalf("err = %v, want it to wrap the last failure", err)
-	}
-	if calls != 3 {
-		t.Errorf("calls = %d, want 3 (the attempt budget)", calls)
-	}
-}
-
-func TestDo_StopsPromptlyWhenTheContextIsCancelled(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	calls := 0
-
-	done := make(chan struct{})
-	var err error
-	go func() {
-		defer close(done)
-		_, err = Do(ctx, Policy{Attempts: 100, Base: 50 * time.Millisecond, Max: time.Second},
-			func(context.Context) (string, error) {
-				calls++
-				if calls == 1 {
-					cancel()
-				}
-				return "", errors.New("still down")
-			})
-	}()
-
-	select {
-	case <-done:
-	case <-time.After(2 * time.Second):
-		t.Fatal("Do did not return after context cancellation")
-	}
-	if !errors.Is(err, context.Canceled) {
-		t.Errorf("err = %v, want context.Canceled", err)
-	}
-}
-
-func TestDo_BacksOffAndIsCappedByMax(t *testing.T) {
-	var delays []time.Duration
-	p := Policy{Attempts: 5, Base: 10 * time.Millisecond, Max: 25 * time.Millisecond}
-
-	last := time.Now()
-	_, _ = Do(context.Background(), p, func(context.Context) (string, error) {
-		now := time.Now()
-		delays = append(delays, now.Sub(last))
-		last = now
-		return "", errors.New("down")
-	})
-
-	// delays[0] is the first call (no wait); the rest are the sleeps.
-	if len(delays) != 5 {
-		t.Fatalf("attempts = %d, want 5", len(delays))
-	}
-	if delays[1] >= delays[2] {
-		t.Errorf("backoff is not growing: %v then %v", delays[1], delays[2])
-	}
-	for i, d := range delays[1:] {
-		if d > 2*p.Max {
-			t.Errorf("delay %d = %v, exceeds the Max cap %v with slack", i+1, d, p.Max)
-		}
-	}
-}
-
-func TestPolicy_ZeroValueStillMakesOneAttempt(t *testing.T) {
-	// A caller that forgets to configure the policy must not silently get a
-	// no-op that returns a nil validator and a nil error.
-	calls := 0
-	_, err := Do(context.Background(), Policy{}, func(context.Context) (string, error) {
-		calls++
-		return "", errors.New("down")
-	})
-
-	if calls != 1 {
-		t.Errorf("calls = %d, want 1", calls)
-	}
-	if err == nil {
-		t.Error("err = nil, want the failure surfaced")
-	}
-}

ingestion/internal/symlinkguard/symlinkguard.go

@@ -18,9 +18,14 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 )
 
+// shutdownFlushTimeout bounds the final flush when Watch stops.
+var shutdownFlushTimeout = 5 * time.Second
+
 // DefaultInterval is used when Watch is given a non-positive interval.
 const DefaultInterval = 30 * time.Second
 
@@ -133,13 +138,32 @@ func Scan(brainDir string) (Result, error) {
 // 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.
+//
+// Delivery runs in its own goroutine (hyperguild#164): a hung notifier, with
+// a full queue, blocked one flush for several notify timeouts, and while the
+// scan waited on it a planted link stayed readable for several intervals.
+// The scan hands alerts over without blocking; any that do not fit are
+// counted in the summary. When ctx ends, Watch returns after a final flush
+// of what was handed over (bounded by shutdownFlushTimeout); alerts that
+// flush cannot deliver are counted in one error log line instead. That
+// shutdown path only runs for a caller that cancels ctx: cmd/server passes
+// context.Background(), so in production the flush on handoff is what keeps
+// alerts from waiting on a restart.
 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{}
+	alerts := make(chan string, maxPending)
+	var overflow atomic.Int64
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }()
+	// The scan is the only sender: closing alerts tells deliver to drain and
+	// flush once more, and Watch returns only after it has.
+	defer func() { close(alerts); wg.Wait() }()
+	q := alertSender{alerts: alerts, overflow: &overflow}
 	reported := ""
 	for {
 		select {
@@ -168,7 +192,76 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
 					". Readers may follow symlinks out of the brain; fix them.")
 			}
 		}
-		q.flush(ctx, n)
+	}
+}
+
+// alertSender hands alerts to deliver without ever blocking the scan; an
+// alert that does not fit is counted in overflow, which deliver adds to the
+// summary.
+type alertSender struct {
+	alerts   chan<- string
+	overflow *atomic.Int64
+}
+
+func (s alertSender) push(msg string) {
+	select {
+	case s.alerts <- msg:
+	default:
+		s.overflow.Add(1)
+	}
+}
+
+// deliver owns the alertQueue. It flushes as soon as an alert is handed
+// over, and again every interval while any are pending (outage retries).
+// When alerts is closed (Watch is stopping) it drains it and flushes once
+// more, bounded by shutdownFlushTimeout, so an alert handed over just before
+// shutdown is not lost (PR #180 review). Whatever that flush cannot deliver
+// is logged as one error with its counts, never dropped silently.
+func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-chan string, overflow *atomic.Int64) {
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	q := &alertQueue{}
+	// drain moves everything handed over into the queue; it reports false
+	// once alerts is closed.
+	drain := func() bool {
+		defer func() { q.suppressed += int(overflow.Swap(0)) }()
+		for {
+			select {
+			case msg, ok := <-alerts:
+				if !ok {
+					return false
+				}
+				q.push(msg)
+			default:
+				return true
+			}
+		}
+	}
+	for {
+		select {
+		case msg, ok := <-alerts:
+			if ok {
+				q.push(msg)
+				ok = drain()
+			}
+			if !ok {
+				q.suppressed += int(overflow.Swap(0))
+				fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout)
+				q.flush(fctx, n)
+				cancel()
+				if len(q.items) > 0 || q.suppressed > 0 {
+					slog.Error("symlink tripwire alerts not delivered at shutdown",
+						"undelivered", len(q.items), "suppressed", q.suppressed)
+				}
+				return
+			}
+			q.flush(ctx, n)
+		case <-t.C:
+			if !drain() {
+				continue // the closed channel is handled by the case above
+			}
+			q.flush(ctx, n)
+		}
 	}
 }
 

ingestion/internal/symlinkguard/symlinkguard_test.go

@@ -1,10 +1,12 @@
 package symlinkguard
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"os"
+	"log/slog"
 	"path/filepath"
 	"strconv"
 	"strings"
@@ -345,13 +347,19 @@ func TestWatchDropsAMessageThatNeverSendsSoLaterAlertsFlow(t *testing.T) {
 	}, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message")
 }
 
-// toggleNotifier fails while down is set.
+// toggleNotifier fails while down is set. Like the real audit.NtfyNotifier
+// (http.NewRequestWithContext), it also fails, without recording, when its
+// ctx is already done (PR #180 round 2: notifiers that ignored ctx let a
+// final flush on the cancelled ctx pass).
 type toggleNotifier struct {
 	recNotifier
 	down atomic.Bool
 }
 
 func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
 	if r.down.Load() {
 		return errors.New("ntfy down")
 	}
@@ -496,3 +504,271 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
 		return false
 	}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
 }
+
+// hungNotifier answers every Notify only after latency (or when ctx ends),
+// with an outage error: an ntfy that accepts connections but never responds
+// until its timeout.
+type hungNotifier struct{ latency time.Duration }
+
+func (h hungNotifier) Notify(ctx context.Context, _ string) error {
+	select {
+	case <-time.After(h.latency):
+	case <-ctx.Done():
+	}
+	return errors.New("ntfy timeout")
+}
+
+func waitGone(t *testing.T, link string) {
+	t.Helper()
+	require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 10*time.Second, time.Millisecond)
+}
+
+// hyperguild#164: alerts were flushed in the scan goroutine, so with a hung
+// ntfy and a full queue one flush (6 x notify timeout) delayed the next scan.
+// The notifier here is slower than the interval, as in production (10 s
+// timeout vs 30 s interval with 6 messages), so the alert handoff is full:
+// a scan that waited on it would keep a link for seconds.
+func TestWatchLinkLifetimeStaysNearOneIntervalWithANotifierSlowerThanTheInterval(t *testing.T) {
+	const interval = 100 * time.Millisecond
+	brain := t.TempDir()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, interval, hungNotifier{latency: 400 * time.Millisecond})
+
+	for i := 0; i < 2*maxPending+3; i++ { // overfill queue and handoff
+		link := filepath.Join(brain, "fill"+strconv.Itoa(i))
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		waitGone(t, link)
+	}
+	var worst time.Duration
+	for i := 0; i < 6; i++ {
+		link := filepath.Join(brain, "probe"+strconv.Itoa(i))
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		planted := time.Now()
+		waitGone(t, link)
+		worst = max(worst, time.Since(planted))
+	}
+	assert.LessOrEqual(t, worst, interval+interval/2, "worst link lifetime %v with a hung notifier", worst)
+}
+
+// Review of PR #180: delivery waited for its own tick, about one interval
+// after the removal. An alert is now flushed as soon as it is handed over.
+func TestWatchDeliversARemovalAlertWellWithinOneInterval(t *testing.T) {
+	const interval = 400 * time.Millisecond
+	brain := t.TempDir()
+	n := &recNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, interval, n)
+	link := filepath.Join(brain, "prompt")
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	waitGone(t, link)
+	removed := time.Now()
+	require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, time.Millisecond)
+	assert.Less(t, time.Since(removed), interval/4)
+}
+
+// Review of PR #180: on ctx end, deliver returned without draining what the
+// scan had handed over, so a removal just before shutdown was neither
+// delivered nor counted. Watch now drains and flushes before returning.
+func TestWatchDeliversAnAlertHandedOverJustBeforeShutdown(t *testing.T) {
+	for i := 0; i < 20; i++ {
+		brain := t.TempDir()
+		n := &recNotifier{}
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
+		link := filepath.Join(brain, "late")
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		waitGone(t, link)
+		cancel()
+		<-done
+		require.Equal(t, 1, n.count(), "iteration %d: the removal alert was lost at shutdown", i)
+	}
+}
+
+// blockingNotifier holds every Notify until its ctx ends, tracking calls in
+// flight.
+type blockingNotifier struct{ inFlight atomic.Int32 }
+
+func (b *blockingNotifier) Notify(ctx context.Context, _ string) error {
+	b.inFlight.Add(1)
+	defer b.inFlight.Add(-1)
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// Watch does not return while its delivery goroutine is still notifying.
+func TestWatchReturnsOnlyAfterDeliveryHasStopped(t *testing.T) {
+	prev := shutdownFlushTimeout
+	shutdownFlushTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { shutdownFlushTimeout = prev })
+	brain := t.TempDir()
+	n := &blockingNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
+	link := filepath.Join(brain, "x")
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	waitGone(t, link)
+	require.Eventually(t, func() bool { return n.inFlight.Load() == 1 }, 2*time.Second, time.Millisecond)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Watch did not return after cancel")
+	}
+	assert.Zero(t, n.inFlight.Load(), "Notify still running after Watch returned")
+}
+
+// slowToggleNotifier fails slowly (a hung ntfy) while down, then records.
+type slowToggleNotifier struct {
+	recNotifier
+	down    atomic.Bool
+	latency time.Duration
+}
+
+func (r *slowToggleNotifier) Notify(ctx context.Context, m string) error {
+	if r.down.Load() {
+		select {
+		case <-time.After(r.latency):
+		case <-ctx.Done():
+		}
+		return errors.New("ntfy timeout")
+	}
+	return r.recNotifier.Notify(ctx, m)
+}
+
+// Review of PR #180: with a slow notifier the scan's handoff overflows, and
+// those alerts exist only in the overflow counter. Every removal is still
+// delivered or counted in the summary.
+func TestWatchCountsHandoffOverflowWithASlowNotifier(t *testing.T) {
+	brain := t.TempDir()
+	n := &slowToggleNotifier{latency: 30 * time.Millisecond}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 2*time.Millisecond, n)
+	const plants = 40
+	for i := 0; i < plants; i++ {
+		link := filepath.Join(brain, "ov"+strconv.Itoa(i))
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		waitGone(t, link)
+	}
+	n.down.Store(false)
+	delivered, counted := 0, 0
+	require.Eventually(t, func() bool {
+		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") {
+				var c int
+				_, _ = fmt.Sscanf(m[i+3:], "%d", &c)
+				counted += c
+			}
+		}
+		return delivered+counted == plants
+	}, 5*time.Second, 10*time.Millisecond, "delivered %d + counted %d of %d", delivered, counted, plants)
+}
+
+// An alert still queued when Watch stops (its send failed, the notifier has
+// since recovered, the retry tick has not come) is flushed once more before
+// Watch returns, on a ctx that is not the cancelled one: toggleNotifier fails
+// on a done ctx, as the real ntfy notifier does.
+func TestWatchFinalFlushSendsAQueuedAlertOnACtxThatOutlivesTheCancel(t *testing.T) {
+	brain := t.TempDir()
+	n := &toggleNotifier{}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { Watch(ctx, brain, 300*time.Millisecond, n); close(done) }()
+	link := filepath.Join(brain, "queued")
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	waitGone(t, link)
+	time.Sleep(20 * time.Millisecond) // let the immediate send fail
+	n.down.Store(false)
+	cancel()
+	<-done
+	assert.Equal(t, 1, n.count(), "the queued alert was not flushed at shutdown")
+}
+
+// lockedBuffer is a goroutine-safe log sink: Watch goroutines of earlier
+// tests may still log while one test captures.
+type lockedBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *lockedBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *lockedBuffer) String() string { b.mu.Lock(); defer b.mu.Unlock(); return b.buf.String() }
+
+func captureLog(t *testing.T) *lockedBuffer {
+	t.Helper()
+	b := &lockedBuffer{}
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(b, nil)))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+	return b
+}
+
+// PR #180 round 2: when the final flush fails (ntfy hung at shutdown) the
+// undelivered alerts vanished with only "will retry" warnings. Watch now logs
+// one error with both counts.
+func TestWatchLogsUndeliveredAndSuppressedCountsWhenTheFinalFlushFails(t *testing.T) {
+	prev := shutdownFlushTimeout
+	shutdownFlushTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { shutdownFlushTimeout = prev })
+	logs := captureLog(t)
+	brain := t.TempDir()
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { Watch(ctx, brain, 5*time.Millisecond, hungNotifier{latency: time.Hour}); close(done) }()
+	link := filepath.Join(brain, "lost")
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	waitGone(t, link)
+	cancel()
+	<-done
+	out := logs.String()
+	assert.Contains(t, out, "level=ERROR")
+	assert.Contains(t, out, `msg="symlink tripwire alerts not delivered at shutdown"`)
+	assert.Contains(t, out, "undelivered=1")
+	assert.Contains(t, out, "suppressed=0")
+}
+
+// deliver, given a handoff that already holds alerts and is closed, moves them
+// into the queue and runs the final flush, even though the ctx is cancelled:
+// the drain-on-close path, without depending on when the close lands.
+func TestDeliverFlushesAlertsDrainedFromAClosedHandoffOnACancelledCtx(t *testing.T) {
+	alerts := make(chan string, 3)
+	alerts <- "a1"
+	alerts <- "a2"
+	close(alerts)
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	n := &toggleNotifier{}
+	var overflow atomic.Int64
+	deliver(ctx, time.Hour, n, alerts, &overflow)
+	assert.Equal(t, []string{"a1", "a2"}, n.msgs)
+}
+
+// PR #180 round 2: when the first receive after close finds the handoff empty,
+// overflow counted by the scan was never moved into the summary.
+func TestDeliverCountsOverflowWhenTheHandoffIsClosedAndEmpty(t *testing.T) {
+	alerts := make(chan string)
+	close(alerts)
+	n := &recNotifier{}
+	var overflow atomic.Int64
+	overflow.Store(3)
+	deliver(context.Background(), time.Hour, n, alerts, &overflow)
+	require.Len(t, n.msgs, 1)
+	assert.Contains(t, n.msgs[0], "3 further alert(s)")
+}