test: close the mutation gaps a coverage audit named (BUG-2730, codex round 18)

Round 18 walked every behavioural change in the diff, named the smallest
edit that would break it, and listed the ones no test caught. Twelve. All
but two are now covered, and each new test was verified against the
mutation it exists for:

- the activity bus's gap channel coalescing (the watch twin had it, this
  one did not)
- Redis-backed atomic subscribe-and-replay, which reaches the guarantee
  by a different mechanism than MemoryBus and would break alone
- a resuming client being held to the per-workspace limit, so the new API
  is not a second door past a bound the fresh path enforces
- the resume-gap report on the new path, with a fresh-subscription
  control so it cannot fire on every subscribe and still pass
- the Redis drop metric, asserted per DROPPED SUBSCRIBER with two slow
  subscribers, so a report hoisted out of the fan-out loop halves the
  count and fails
- the gap channel surviving Unsubscribe, since closing it would make a
  consumer's select spin
- every subscribe API returning a non-nil signal
- the watch handler incrementing the WATCH counter (countMidStreamResync
  takes a bool to choose, which is the kind of argument that gets passed
  the wrong way round), with both wrong-counter legs asserted
- the production cooldown, which every handler test overrides, so
  nothing else would notice it set to zero
- the wrapper's gauge on the atomic-resume path, which the previous
  assertion checked only for non-nil-ness

Two left uncovered deliberately: an interleaving test at the handler
level for the atomic API (the bus-level tests carry that guarantee and
the handler cannot arrange the interleaving), and the same for the
handler choosing the atomic call over subscribe-plus-EventsSince.

Two existing tests were also repaired rather than kept green by luck: the
ordering stress test demanded every published event and a slow reader
legitimately loses some, and the new Redis atomicity test resumed against
a workspace the bus was not covering.

internal/events/gap_signal_test.go

@@ -377,18 +377,25 @@ func TestConcurrentPublishesDeliverInIDOrder(t *testing.T) {
 	const publishers, each = 8, 40
 	total := publishers * each
 
-	// Drained concurrently: a 64-deep channel would otherwise overflow and
-	// the dropped IDs would look like an ordering fault.
+	// Drained concurrently. The reader may legitimately MISS some — a 64-deep
+	// channel overflows and the bus drops for a slow subscriber, which is the
+	// behaviour tested elsewhere — so this reads until the stream goes quiet
+	// rather than demanding all of them. The claim under test is the ORDER of
+	// what arrives, and a minimum count below keeps it from passing vacuously.
 	received := make(chan int64, total)
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
-		for range total {
-			e, ok := <-ch
-			if !ok {
+		for {
+			select {
+			case e, ok := <-ch:
+				if !ok {
+					return
+				}
+				received <- e.ID
+			case <-time.After(500 * time.Millisecond):
 				return
 			}
-			received <- e.ID
 		}
 	}()
 
@@ -406,17 +413,245 @@ func TestConcurrentPublishesDeliverInIDOrder(t *testing.T) {
 
 	select {
 	case <-done:
-	case <-time.After(10 * time.Second):
-		t.Fatal("the reader did not receive every published event")
+	case <-time.After(20 * time.Second):
+		t.Fatal("the reader never went quiet")
 	}
 	close(received)
 
 	var prev int64
+	var count int
 	for id := range received {
+		count++
 		if id <= prev {
 			t.Fatalf("IDs arrived out of order: %d after %d — a subscriber's cursor regressed, "+
 				"so its next reconnect replays events it already has", id, prev)
 		}
 		prev = id
 	}
+	// Not vacuous: an empty or near-empty stream would satisfy the ordering
+	// check trivially.
+	if count < total/2 {
+		t.Fatalf("only %d of %d events arrived; too few to say anything about ordering", count, total)
+	}
+}
+
+// TestActivityGapSignalCoalesces is the load bound on THIS bus. The watch
+// bus's twin existed from the start; this one did not, and the bound is the
+// whole answer to "does telling a slow subscriber make it slower".
+func TestActivityGapSignalCoalesces(t *testing.T) {
+	b := New()
+	defer b.Close()
+
+	ch, gaps, ok := b.SubscribeIfAllowed("ws-1", 0)
+	if !ok {
+		t.Fatal("subscribe refused")
+	}
+	defer b.Unsubscribe(ch)
+
+	for range 200 {
+		b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+	}
+
+	if !raised(gaps) {
+		t.Fatal("no gap signal after 200 events into a 64-deep channel")
+	}
+	if raised(gaps) {
+		t.Error("more than one signal was queued; the channel must coalesce, not accumulate")
+	}
+}
+
+// TestRedisBusSubscribeAndReplayIsAtomic is the multi-instance twin of
+// TestSubscribeAndReplayIsAtomic. It gets its own test because the two
+// implementations reach the guarantee by different means — this one keeps
+// subscribers and buffers under a single mutex, so a future refactor that
+// split them would break here and nowhere else.
+func TestRedisBusSubscribeAndReplayIsAtomic(t *testing.T) {
+	b := newTestRedisBus(t)
+
+	// A holder first: this bus builds a workspace's replay buffer as part of
+	// COVERING it, and a resume against a workspace it has never covered is
+	// answered "cannot vouch" — correctly, and not what this test is about.
+	holder, _ := b.Subscribe("ws-1")
+	defer b.Unsubscribe(holder)
+
+	b.fanOutLocally(Event{ID: 10, Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "before-1"})
+	b.fanOutLocally(Event{ID: 11, Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "before-2"})
+
+	ch, missed, gaps, ok := b.SubscribeAndReplaySince("ws-1", 10, 0)
+	if !ok {
+		t.Fatal("subscribe refused")
+	}
+	defer b.Unsubscribe(ch)
+	if gaps == nil {
+		t.Error("no gap channel returned")
+	}
+	if !containsItem(missed, "before-2") {
+		t.Fatalf("the event above the cursor was not replayed: %+v", missed)
+	}
+	if containsItem(missed, "before-1") {
+		t.Error("the event AT the cursor was replayed; the client already has it")
+	}
+
+	// Published after: must arrive live and NOT be in the replay just read.
+	b.fanOutLocally(Event{ID: 12, Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "after"})
+	if !drainContains(t, ch, "after") {
+		t.Error("an event published after the subscribe was not delivered live")
+	}
+}
+
+// TestResumeIsSubjectToTheWorkspaceLimit closes a hole the new API could have
+// opened: the resume path is a second door onto the same subscriber set, and a
+// client that supplies Last-Event-ID must not thereby skip the bound that a
+// fresh one is held to.
+func TestResumeIsSubjectToTheWorkspaceLimit(t *testing.T) {
+	b := New()
+	defer b.Close()
+
+	first, _, ok := b.SubscribeIfAllowed("ws-1", 1)
+	if !ok {
+		t.Fatal("the first subscribe was refused")
+	}
+	defer b.Unsubscribe(first)
+
+	if _, _, _, ok := b.SubscribeAndReplaySince("ws-1", 1, 1); ok {
+		t.Error("a resuming client was admitted past the per-workspace limit")
+	}
+	// Control: the same call succeeds when there is room, so the refusal above
+	// is the limit and not the method being broken.
+	if _, _, _, ok := b.SubscribeAndReplaySince("ws-1", 1, 2); !ok {
+		t.Error("a resuming client was refused with room to spare")
+	}
+}
+
+// TestAtomicResumeReportsAnUnservableSpan pins the observer report on the NEW
+// path. EventsSince has always reported; this method reads the buffer directly
+// and had to carry its own, and the report is what makes the resync population
+// visible in production.
+func TestAtomicResumeReportsAnUnservableSpan(t *testing.T) {
+	b := New()
+	defer b.Close()
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
+
+	// A cursor for a workspace this process has never published to.
+	ch, missed, _, ok := b.SubscribeAndReplaySince("ws-unknown", b.base+9999, 0)
+	if !ok {
+		t.Fatal("subscribe refused")
+	}
+	defer b.Unsubscribe(ch)
+	if missed != nil {
+		t.Fatalf("a cursor for an unknown workspace was answered as servable: %+v", missed)
+	}
+	if got := obs.gaps(); len(got) != 1 || got[0] != "ws-unknown" {
+		t.Errorf("resume gaps reported = %v, want exactly [ws-unknown]", got)
+	}
+
+	// Control: a FRESH subscription on the same path reports nothing. Without
+	// this leg the report could fire on every subscribe and still pass.
+	obs2 := &recordingObserver{}
+	b.SetObserver(obs2)
+	ch2, _, _, ok := b.SubscribeAndReplaySince("ws-unknown", 0, 0)
+	if !ok {
+		t.Fatal("subscribe refused")
+	}
+	defer b.Unsubscribe(ch2)
+	if got := obs2.gaps(); len(got) != 0 {
+		t.Errorf("a fresh subscription reported a resume gap: %v", got)
+	}
+}
+
+// TestRedisBusDropIsReportedPerSubscriber pins two things the single-subscriber
+// tests cannot: that the Redis fan-out reports drops at all, and that it
+// reports once per DROPPED SUBSCRIBER rather than once per publish. A report
+// hoisted out of the loop would halve the counter on a two-slow-subscriber
+// workspace and look right on every other test.
+func TestRedisBusDropIsReportedPerSubscriber(t *testing.T) {
+	b := newTestRedisBus(t)
+	obs := &recordingObserver{}
+	b.SetObserver(obs)
+
+	slowA, _, ok := b.SubscribeIfAllowed("ws-1", 0)
+	if !ok {
+		t.Fatal("subscribe refused")
+	}
+	slowB, _, ok := b.SubscribeIfAllowed("ws-1", 0)
+	if !ok {
+		t.Fatal("subscribe refused")
+	}
+	defer b.Unsubscribe(slowA)
+	defer b.Unsubscribe(slowB)
+
+	// Fill both channels, then overflow both with ONE event.
+	for i := 1; i <= 64; i++ {
+		b.fanOutLocally(Event{ID: int64(i), Type: ItemUpdated, WorkspaceID: "ws-1"})
+	}
+	if got := obs.dropped(); len(got) != 0 {
+		t.Fatalf("drops reported before either channel overflowed: %v", got)
+	}
+
+	b.fanOutLocally(Event{ID: 65, Type: ItemUpdated, WorkspaceID: "ws-1"})
+
+	got := obs.dropped()
+	if len(got) != 2 {
+		t.Fatalf("one event dropped for TWO subscribers must report twice, got %d: %v", len(got), got)
+	}
+	for _, r := range got {
+		if r != DropReasonSlowSubscriber {
+			t.Errorf("reason = %q, want %q", r, DropReasonSlowSubscriber)
+		}
+	}
+}
+
+// TestGapChannelOutlivesUnsubscribe pins the lifetime rule stated on the field.
+// Closing the gap channel would make it permanently ready, and a consumer
+// selecting on both would spin at full speed between Unsubscribe and noticing
+// the event channel had closed.
+func TestGapChannelOutlivesUnsubscribe(t *testing.T) {
+	b := New()
+	defer b.Close()
+
+	ch, gaps, ok := b.SubscribeIfAllowed("ws-1", 0)
+	if !ok {
+		t.Fatal("subscribe refused")
+	}
+	b.Unsubscribe(ch)
+
+	if _, open := <-ch; open {
+		t.Fatal("Unsubscribe must close the event channel; the rest of this test assumes it")
+	}
+	select {
+	case <-gaps:
+		t.Error("the gap channel was closed or signalled by Unsubscribe; a consumer's select would spin")
+	default:
+	}
+}
+
+// TestEverySubscribeAPIReturnsAGapChannel is the small structural claim behind
+// the seam: no way of registering a subscriber may hand back a nil signal,
+// because a nil channel swallows every send through the default arm and the
+// subscriber would be silently unreachable.
+func TestEverySubscribeAPIReturnsAGapChannel(t *testing.T) {
+	for name, sub := range map[string]func(b *MemoryBus) (chan Event, <-chan struct{}){
+		"Subscribe": func(b *MemoryBus) (chan Event, <-chan struct{}) {
+			return b.Subscribe("ws-1")
+		},
+		"SubscribeIfAllowed": func(b *MemoryBus) (chan Event, <-chan struct{}) {
+			ch, gaps, _ := b.SubscribeIfAllowed("ws-1", 0)
+			return ch, gaps
+		},
+		"SubscribeAndReplaySince": func(b *MemoryBus) (chan Event, <-chan struct{}) {
+			ch, _, gaps, _ := b.SubscribeAndReplaySince("ws-1", 0, 0)
+			return ch, gaps
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			b := New()
+			defer b.Close()
+			ch, gaps := sub(b)
+			defer b.Unsubscribe(ch)
+			if gaps == nil {
+				t.Errorf("%s returned a nil gap channel; every drop for this subscriber would be silent", name)
+			}
+		})
+	}
 }

internal/events/observer_test.go

@@ -37,6 +37,12 @@ func (o *recordingObserver) EventDropped(reason string) {
 	o.drops = append(o.drops, reason)
 }
 
+func (o *recordingObserver) gaps() []string {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return append([]string(nil), o.resumeGaps...)
+}
+
 func (o *recordingObserver) resetReasons() []string {
 	o.mu.Lock()
 	defer o.mu.Unlock()

internal/metrics/events_gap_wiring_test.go

@@ -43,7 +43,8 @@ func TestInstrumentedBusPassesTheGapChannelThrough(t *testing.T) {
 func TestInstrumentedBusForwardsTheReplay(t *testing.T) {
 	inner := events.New()
 	defer inner.Close()
-	b := NewInstrumentedBus(inner, New())
+	m := New()
+	b := NewInstrumentedBus(inner, m)
 
 	b.Publish(events.Event{Type: events.ItemUpdated, WorkspaceID: "ws-1", ItemID: "one"})
 	b.Publish(events.Event{Type: events.ItemUpdated, WorkspaceID: "ws-1", ItemID: "two"})
@@ -60,9 +61,11 @@ func TestInstrumentedBusForwardsTheReplay(t *testing.T) {
 	if len(missed) != 1 || missed[0].ItemID != "two" {
 		t.Fatalf("the replay was not forwarded intact: %+v", missed)
 	}
-	if got := (*b.metrics.EventBusSubscribers); got == nil {
-		t.Error("the subscriber gauge was not wired on this path")
-	}
+	// The gauge must have been SET, not merely be non-nil — the wrapper's
+	// whole job on this path is the bookkeeping, and a delegation that
+	// forwarded the data and skipped trackSubscription would look identical
+	// to a nil check.
+	assertGauge(t, m, "pad_sse_connections_active", 1)
 }
 
 // The adapter is the last hop between the bus's report and Prometheus. It is
@@ -82,3 +85,26 @@ func TestEventsObserverRecordsADrop(t *testing.T) {
 	assertCounter(t, m, "pad_watchevents_notifications_dropped_total",
 		map[string]string{"reason": events.DropReasonSlowSubscriber}, 0)
 }
+
+// assertGauge mirrors assertCounter for gauge families.
+func assertGauge(t *testing.T, m *Metrics, name string, want float64) {
+	t.Helper()
+	families, err := m.Registry.Gather()
+	if err != nil {
+		t.Fatalf("gather: %v", err)
+	}
+	for _, f := range families {
+		if f.GetName() != name {
+			continue
+		}
+		for _, metric := range f.GetMetric() {
+			if got := metric.GetGauge().GetValue(); got != want {
+				t.Errorf("%s = %v, want %v", name, got, want)
+			}
+			return
+		}
+	}
+	if want != 0 {
+		t.Errorf("%s is not exported", name)
+	}
+}

internal/server/handlers_stream_gap_test.go

@@ -185,6 +185,8 @@ func waitForFrameRefusing(t *testing.T, frames <-chan string, want, refuse strin
 // stream, whose handler is a different function with its own select loop.
 func TestWatchStreamAnnouncesAGapMidStream(t *testing.T) {
 	srv := testServer(t)
+	m := metrics.New()
+	srv.SetMetrics(m)
 	bus := &gapWatchBus{Bus: watchevents.New(), gaps: make(chan struct{}, 1)}
 	srv.SetWatchEventsBus(bus)
 	_, _, tok, _ := setupWatchTestUser(t, srv)
@@ -202,6 +204,19 @@ func TestWatchStreamAnnouncesAGapMidStream(t *testing.T) {
 	frame := waitForFrameWithEvent(t, frames, "sync_required")
 	assertRetiresCursor(t, frame)
 
+	// The WATCH counter, not the activity one. countMidStreamResync takes a
+	// bool to pick between them, which is exactly the kind of argument that
+	// gets passed the wrong way round.
+	if got := counterValue(t, m.WatchMidstreamResyncsTotal); got != 1 {
+		t.Errorf("pad_watchevents_midstream_resyncs_total = %v, want 1", got)
+	}
+	if got := counterValue(t, m.EventMidstreamResyncsTotal); got != 0 {
+		t.Errorf("a watch-stream announcement moved the ACTIVITY counter: %v", got)
+	}
+	if got := counterValue(t, m.WatchResumeGapsTotal); got != 0 {
+		t.Errorf("a mid-stream gap moved the watch resume counter: %v", got)
+	}
+
 	// Same liveness claim as the activity twin (codex round 3). This stream
 	// has no cheap ordinary event to publish — a notification needs a watch
 	// predicate to match — so the assertion is that the handler did not

internal/server/stream_gap_announcer_test.go

@@ -78,3 +78,26 @@ func TestGapAnnouncerDoesNotAnnounceAnEmptyWindow(t *testing.T) {
 		t.Error("a gap after a quiet window was bounded; the window had already closed")
 	}
 }
+
+// The production cooldown is a real number with a stated reason (a delta-sync
+// round trip), and every handler test overrides it — so nothing else would
+// notice it being set to zero, which would disable the bound entirely on the
+// only deployment that matters.
+func TestProductionGapCooldownIsNotDisabled(t *testing.T) {
+	var s Server
+	if got := s.gapCooldown(); got != midStreamGapCooldown {
+		t.Errorf("a server with no override must use the production cooldown, got %v", got)
+	}
+	if midStreamGapCooldown <= 0 {
+		t.Fatal("the production cooldown is not positive; the rate limit is off")
+	}
+	if midStreamGapCooldown >= sseKeepaliveInterval {
+		t.Errorf("cooldown %v is not shorter than the %v keepalive; a new hole would wait "+
+			"longer than the connection's own heartbeat", midStreamGapCooldown, sseKeepaliveInterval)
+	}
+
+	s.midStreamGapCooldownOverride = 7 * time.Millisecond
+	if got := s.gapCooldown(); got != 7*time.Millisecond {
+		t.Errorf("the override was ignored: %v", got)
+	}
+}