diff --git a/internal/events/gap_signal_test.go b/internal/events/gap_signal_test.go
index 155f18bb..30a02ad5 100644
--- a/internal/events/gap_signal_test.go
+++ b/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)
+			}
+		})
+	}
 }
diff --git a/internal/events/observer_test.go b/internal/events/observer_test.go
index 3dac498e..9e4e8618 100644
--- a/internal/events/observer_test.go
+++ b/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()
diff --git a/internal/metrics/events_gap_wiring_test.go b/internal/metrics/events_gap_wiring_test.go
index e2161ec7..ceb401c7 100644
--- a/internal/metrics/events_gap_wiring_test.go
+++ b/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)
+	}
+}
diff --git a/internal/server/handlers_stream_gap_test.go b/internal/server/handlers_stream_gap_test.go
index 156bee29..dfcda7b6 100644
--- a/internal/server/handlers_stream_gap_test.go
+++ b/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
diff --git a/internal/server/stream_gap_announcer_test.go b/internal/server/stream_gap_announcer_test.go
index 48795737..4dc59206 100644
--- a/internal/server/stream_gap_announcer_test.go
+++ b/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)
+	}
+}