feat(watchevents,metrics): operational observability for the Redis notification bus (BUG-2727)

The watch bus detects four conditions an operator would want to alert on —
a notification dropped for a slow local subscriber, a gap in the received
id sequence, an id-space reset, and the receive loop stopping — and until
now reported all four to slog and nowhere else. Log lines are not
alertable without someone already looking, and the last of the four was
not even logged: the loop returned silently, leaving an instance that
publishes fine and receives nothing indistinguishable from a quiet
workspace.

Adds watchevents.Observer, an adapter seam rather than a bus wrapper.
The events.EventBus wrapper shape does not work here: every condition is
detected on the RECEIVE path, inside the bus, and is invisible at the Bus
interface — a wrapper can count publishes and subscribers, but not a
notification that never arrived.

Two corrections to BUG-2727's filing, both verified against go-redis
v9.22.0 rather than assumed:

- Its proposed fix — "re-subscribe rather than exiting where the cause is
  recoverable" — would be dead code. PubSub.Channel's message channel is
  closed ONLY on pool.ErrClosed; every other receive error is retried
  indefinitely, and a health-check goroutine pings every 3s and
  reconnects on failure. So go-redis already does the re-subscribing. The
  exit gets an ERROR log and a counter instead, which is what the
  condition actually needs.
- The genuinely silent path is go-redis DROPPING messages when a
  subscription's 100-deep buffer stays full past its 60s send timeout,
  logged only through go-redis's own logger. Pad cannot count that
  directly, so it is reported by its CONSEQUENCE (a sequence gap) and its
  cause is made visible by routing go-redis's logger into slog.
  Observer's doc comment states that boundary, so a gap is not misread as
  evidence of any particular cause.

Session presence gets the same treatment for the same reason: it is
fail-soft everywhere by design, so its failures have no user-visible
signal beyond a push that quietly reaches fewer sessions than it should.
The renew counter is deliberately NOT throttled where its log line is —
throttling the metric would make it under-report during the incident it
exists for.

Tests assert the CONDITION increments the counter, not that the counter
exists, and each asserts its own premise first (a healthy subscriber
reports nothing; contiguous ids report nothing; a cold start reports
nothing) so a bus that reported on every notification could not pass. The
receive-loop test drives the real closed-client condition rather than
calling the reporter.

Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X

cmd/pad/cmd_server.go

@@ -648,6 +648,16 @@ func serveCmd() *cobra.Command {
 				if err := rc.Ping(context.Background()).Err(); err != nil {
 					return fmt.Errorf("redis connection failed: %w", err)
 				}
+				// Route go-redis's own diagnostics into slog (BUG-2727).
+				// Its default logger writes to the standard log package,
+				// so its messages bypass the structured pipeline entirely
+				// — including the one that matters most here: "channel is
+				// full ... message is dropped", emitted when a
+				// subscription's 100-deep buffer stays full past the 60s
+				// send timeout. That is a silent notification loss whose
+				// only trace was a line nobody's log aggregator was
+				// shaped to catch.
+				redis.SetLogger(redisSlogLogger{})
 				eventBus = events.NewRedisBus(rc)
 				watchRedis = rc
 				slog.Info("Event bus using Redis pub/sub", "addr", opts.Addr, "db", opts.DB)
@@ -671,10 +681,20 @@ func serveCmd() *cobra.Command {
 			// doc for what each implementation does and does not fix.
 			var watchBus watchevents.Bus
 			if watchRedis != nil {
-				watchBus = watchevents.NewRedisBus(watchRedis)
+				redisWatchBus := watchevents.NewRedisBus(watchRedis)
+				// Operational instrumentation (BUG-2727). Attached to the
+				// concrete type because the conditions it reports —
+				// dropped notifications, sequence gaps, id-space resets,
+				// the receive loop stopping — are detected inside the bus
+				// and are not visible at the Bus interface, so a wrapper
+				// of the events.EventBus kind cannot see them.
+				redisWatchBus.SetObserver(metrics.NewWatchEventsObserver(m))
+				watchBus = redisWatchBus
 				slog.Info("Watch notification bus using Redis pub/sub")
 			} else {
-				watchBus = watchevents.New()
+				memWatchBus := watchevents.New()
+				memWatchBus.SetObserver(metrics.NewWatchEventsObserver(m))
+				watchBus = memWatchBus
 				slog.Info("Watch notification bus using in-memory (single instance)")
 			}
 			srv.SetWatchEventsBus(watchBus)
@@ -697,6 +717,13 @@ func serveCmd() *cobra.Command {
 			var redisPresence *server.RedisSessionPresence
 			if watchRedis != nil {
 				redisPresence = server.NewRedisSessionPresence(watchRedis)
+				// Presence is fail-soft by design — a failed write leaves
+				// a live session unlisted rather than dropping its
+				// connection — so its failures have no user-visible
+				// signal beyond a push that quietly reaches fewer
+				// sessions. The counter is the only alertable trace
+				// (BUG-2727).
+				redisPresence.SetObserver(server.NewMetricsPresenceObserver(m))
 				// Backstop for early-return paths that never reach the
 				// shutdown sequence; Close is idempotent. Deliberately does
 				// NOT delete this instance's entries — a shutdown racing a
@@ -711,6 +738,24 @@ func serveCmd() *cobra.Command {
 			}
 			srv.SetSessionPresence(sessionPresence)
 
+			// Redis reachability prober (BUG-2727). Reports into
+			// /health/ready's payload and pad_redis_up; deliberately does
+			// NOT gate readiness — see server.RedisHealth's doc comment.
+			// Only wired when there IS a Redis, so a single-process
+			// deployment reports no redis block rather than a false one.
+			if watchRedis != nil {
+				redisHealth := server.NewRedisHealth(watchRedis, func(ok bool) {
+					if ok {
+						m.RedisUp.Set(1)
+					} else {
+						m.RedisUp.Set(0)
+					}
+				})
+				redisHealth.Start()
+				defer redisHealth.Stop()
+				srv.SetRedisHealth(redisHealth)
+			}
+
 			// Yjs collab room manager (PLAN-1248). Single-instance only
 			// today; multi-replica fanout via Redis is a deferred IDEA.
 			// MemoryOpBus is in-process; the OpBus interface keeps the

cmd/pad/redis_logger.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+)
+
+// redisSlogLogger adapts go-redis's internal logging interface onto slog
+// (BUG-2727).
+//
+// go-redis logs a handful of operational conditions through a package-level
+// logger that defaults to the standard `log` package. Two of them are
+// notification loss and neither has any other trace:
+//
+//   - "channel is full for <timeout> (message is dropped)" — a subscription's
+//     buffer stayed full past the send timeout, so go-redis discarded a
+//     message before Pad ever saw it. Pad detects the CONSEQUENCE as a
+//     sequence gap (see watchevents.Observer), but the cause is only visible
+//     here.
+//   - "unknown message type" — a protocol-level surprise on the subscription.
+//
+// Logged at WARN rather than ERROR: none of them is actionable on its own,
+// and go-redis also logs benign reconnect chatter through the same call.
+// The metrics are what to alert on; this is what to read afterwards.
+type redisSlogLogger struct{}
+
+func (redisSlogLogger) Printf(_ context.Context, format string, v ...interface{}) {
+	slog.Warn("redis: " + fmt.Sprintf(format, v...))
+}

internal/metrics/metrics.go

@@ -27,6 +27,51 @@ type Metrics struct {
 	EventBusPublishTotal *prometheus.Counter
 	EventBusSubscribers  *prometheus.Gauge
 
+	// Redis operability metrics (BUG-2727). Wired from
+	// cmd/pad/cmd_server.go and only meaningful on a deployment with
+	// PAD_REDIS_URL set — a single-process binary never touches Redis, so
+	// these stay at their zero values there, which is the honest reading
+	// rather than an absence.
+	//
+	// RedisUp is written by internal/server's health prober, NOT sampled
+	// on scrape: a collector that dials on every scrape turns a monitoring
+	// system into a load generator and makes the metric's value depend on
+	// who is asking. It is 1 when the last probe succeeded, 0 when it
+	// failed, and absent entirely when the deployment has no Redis.
+	RedisUp prometheus.Gauge
+
+	// WatchNotificationsDroppedTotal counts notifications this instance
+	// RECEIVED and could not hand to one of its own subscribers. It does
+	// not count go-redis discarding messages before we see them — that
+	// shows up as a sequence gap instead; see watchevents.Observer for why
+	// the distinction matters when reading these two together.
+	WatchNotificationsDroppedTotal *prometheus.CounterVec
+
+	// WatchSequenceGapsTotal counts gap EVENTS;
+	// WatchNotificationsMissedTotal counts the notifications those gaps
+	// span. Both, because one gap of 500 and 500 gaps of one are very
+	// different incidents and a single counter cannot tell them apart.
+	WatchSequenceGapsTotal        prometheus.Counter
+	WatchNotificationsMissedTotal prometheus.Counter
+
+	// WatchSequenceResetsTotal counts id-space changes (epoch change or
+	// the shared counter going backwards) — each one drops this
+	// instance's replay buffer, so resumes across it answer
+	// sync_required.
+	WatchSequenceResetsTotal *prometheus.CounterVec
+
+	// WatchReceiveLoopExitsTotal counts the receive loop stopping. Any
+	// non-zero value outside a shutdown means an instance that publishes
+	// fine and receives nothing.
+	WatchReceiveLoopExitsTotal prometheus.Counter
+
+	// SessionPresenceFailuresTotal counts failed presence operations by
+	// op. A sustained non-zero rate means sessions are missing from
+	// GET /api/v1/sessions and untargetable by a session-directed push,
+	// which is invisible to the user beyond a push that quietly reaches
+	// nobody.
+	SessionPresenceFailuresTotal *prometheus.CounterVec
+
 	// MCP traffic metrics (PLAN-943 TASK-961). Wired from
 	// internal/server/middleware_mcp_audit.go (per-request seam) and
 	// internal/server/middleware_mcp_auth.go + middleware_auth.go
@@ -190,7 +235,49 @@ func New() *Metrics {
 		Buckets: []float64{10, 60, 300, 900, 3600, 21600, 86400, 7 * 86400, 30 * 86400, 60 * 86400},
 	})
 
+	redisUp := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "pad_redis_up",
+		Help: "1 if the last Redis health probe succeeded, 0 if it failed. Not exported when the deployment has no Redis.",
+	})
+
+	watchNotificationsDroppedTotal := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "pad_watchevents_notifications_dropped_total",
+		Help: "Notifications received by this instance but not delivered to a local subscriber, by reason.",
+	}, []string{"reason"})
+
+	watchSequenceGapsTotal := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pad_watchevents_sequence_gaps_total",
+		Help: "Times this instance detected a forward jump in the notification id sequence, i.e. it missed at least one notification.",
+	})
+
+	watchNotificationsMissedTotal := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pad_watchevents_notifications_missed_total",
+		Help: "Notifications spanned by detected sequence gaps — how many were missed, where pad_watchevents_sequence_gaps_total counts how often.",
+	})
+
+	watchSequenceResetsTotal := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "pad_watchevents_sequence_resets_total",
+		Help: "Times the notification id space changed under this instance, dropping its replay buffer, by reason.",
+	}, []string{"reason"})
+
+	watchReceiveLoopExitsTotal := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pad_watchevents_receive_loop_exits_total",
+		Help: "Times the Redis subscription receive loop stopped. Non-zero outside shutdown means this instance receives no notifications at all.",
+	})
+
+	sessionPresenceFailuresTotal := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "pad_session_presence_failures_total",
+		Help: "Failed session-presence operations by op. Failures leave sessions unlisted and untargetable.",
+	}, []string{"op"})
+
 	reg.MustRegister(
+		redisUp,
+		watchNotificationsDroppedTotal,
+		watchSequenceGapsTotal,
+		watchNotificationsMissedTotal,
+		watchSequenceResetsTotal,
+		watchReceiveLoopExitsTotal,
+		sessionPresenceFailuresTotal,
 		httpRequestsTotal,
 		httpRequestDuration,
 		httpResponseSize,
@@ -208,7 +295,16 @@ func New() *Metrics {
 	)
 
 	return &Metrics{
-		Registry:                   reg,
+		Registry: reg,
+
+		RedisUp:                        redisUp,
+		WatchNotificationsDroppedTotal: watchNotificationsDroppedTotal,
+		WatchSequenceGapsTotal:         watchSequenceGapsTotal,
+		WatchNotificationsMissedTotal:  watchNotificationsMissedTotal,
+		WatchSequenceResetsTotal:       watchSequenceResetsTotal,
+		WatchReceiveLoopExitsTotal:     watchReceiveLoopExitsTotal,
+		SessionPresenceFailuresTotal:   sessionPresenceFailuresTotal,
+
 		HTTPRequestsTotal:          httpRequestsTotal,
 		HTTPRequestDuration:        httpRequestDuration,
 		HTTPResponseSize:           httpResponseSize,

internal/metrics/watchevents_observer.go

@@ -0,0 +1,50 @@
+package metrics
+
+import "github.com/PerpetualSoftware/pad/internal/watchevents"
+
+// WatchEventsObserver adapts a Metrics into a watchevents.Observer
+// (BUG-2727), so internal/watchevents reports its operational events
+// without importing Prometheus.
+//
+// This is an ADAPTER rather than a bus wrapper — the shape
+// NewInstrumentedBus uses for events.EventBus — because every condition it
+// reports is detected on the RECEIVE path, inside the bus, and is invisible
+// from outside the Bus interface. A wrapper can count publishes and
+// subscribers; it cannot see a notification that never arrived.
+type WatchEventsObserver struct {
+	m *Metrics
+}
+
+// NewWatchEventsObserver returns an observer writing into m.
+func NewWatchEventsObserver(m *Metrics) *WatchEventsObserver {
+	return &WatchEventsObserver{m: m}
+}
+
+// Compile-time proof this satisfies the seam it exists for. Without it, a
+// later signature change in the interface would surface as a wiring error
+// in cmd/pad rather than here.
+var _ watchevents.Observer = (*WatchEventsObserver)(nil)
+
+func (o *WatchEventsObserver) NotificationDropped(reason string) {
+	o.m.WatchNotificationsDroppedTotal.WithLabelValues(reason).Inc()
+}
+
+// SequenceGap increments both counters: the gap event, and the number of
+// notifications it spans. A non-positive `missing` still counts as an
+// event — the bus only calls this on a forward jump, so a value that is
+// not positive means the caller's arithmetic changed, and swallowing it
+// would hide that regression rather than the gap.
+func (o *WatchEventsObserver) SequenceGap(missing int64) {
+	o.m.WatchSequenceGapsTotal.Inc()
+	if missing > 0 {
+		o.m.WatchNotificationsMissedTotal.Add(float64(missing))
+	}
+}
+
+func (o *WatchEventsObserver) SequenceReset(reason string) {
+	o.m.WatchSequenceResetsTotal.WithLabelValues(reason).Inc()
+}
+
+func (o *WatchEventsObserver) ReceiveLoopExited() {
+	o.m.WatchReceiveLoopExitsTotal.Inc()
+}

internal/server/handlers_workspaces.go

@@ -130,6 +130,39 @@ func (s *Server) handleHealthReady(w http.ResponseWriter, r *http.Request) {
 		"driver":           string(s.store.D().Driver()),
 	}
 
+	// Redis reachability, INFORMATIONAL — it never changes the 200/503
+	// (BUG-2727). Readiness stays database-only because every write path,
+	// the REST API and the web UI work with Redis down; gating on it would
+	// pull healthy replicas out of the load balancer over a degraded
+	// feature. What Redis being down actually costs is named on the
+	// `degrades` field so a reader does not have to know the architecture
+	// to interpret the flag.
+	//
+	// Absent when no prober is wired, which means no Redis is configured.
+	// A `false` there would say "Redis is down" about a deployment that
+	// has none.
+	if s.redisHealth != nil {
+		status := s.redisHealth.Status()
+		redisResp := map[string]interface{}{
+			"reachable": status.Reachable,
+			"probed":    status.Probed,
+		}
+		if status.Error != "" {
+			redisResp["error"] = status.Error
+		}
+		if !status.LastCheck.IsZero() {
+			redisResp["last_check"] = status.LastCheck.UTC().Format(time.RFC3339)
+		}
+		if status.Probed && !status.Reachable {
+			redisResp["degrades"] = []string{
+				"cross-instance activity events",
+				"watch notifications",
+				"session presence and session-targeted push",
+			}
+		}
+		resp["redis"] = redisResp
+	}
+
 	writeJSON(w, http.StatusOK, resp)
 }
 

internal/server/redis_health.go

@@ -0,0 +1,203 @@
+package server
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	// redisProbeInterval is how often the prober pings Redis. Chosen to be
+	// a few times finer than the presence TTL (90s) so a loss of Redis is
+	// visible in the health payload and in pad_redis_up well before the
+	// first user-visible consequence — a session expiring out of the
+	// picker — rather than at the same moment.
+	redisProbeInterval = 15 * time.Second
+
+	// redisProbeTimeout bounds one ping ONCE A CONNECTION EXISTS.
+	// Deliberately short: this is a liveness question, and a probe that
+	// hangs is itself the answer.
+	//
+	// It does NOT bound the whole probe — see RedisHealth's doc comment.
+	// Establishing a connection is governed by the client's DialTimeout
+	// (go-redis v9 default: 5s), which this context cannot shorten, so the
+	// real worst case for one probe is that dial rather than this value.
+	// redisProbeInterval is comfortably longer than the dial bound, so a
+	// probe against a black-holed server still finishes before the next
+	// tick is due.
+	redisProbeTimeout = 2 * time.Second
+)
+
+// RedisHealth periodically probes Redis and caches the result (BUG-2727).
+//
+// WHY A PROBER AND NOT A SCRAPE-TIME COLLECTOR. Both /health/ready and
+// /metrics need to answer "is Redis reachable", and both are called by
+// automated systems on their own schedules. Dialling from inside either
+// one makes Pad's Redis load a function of how many monitors are pointed
+// at it, and makes the answer depend on WHO ASKED — a scrape and a
+// readiness check a millisecond apart could disagree. One prober on a
+// fixed interval, two readers of its cached result, and the reported value
+// is the same for everyone.
+//
+// WHAT IT DOES NOT DO. It does not gate readiness. /health/ready stays
+// database-only, deliberately: the REST API, the web UI and every write
+// path work with Redis down, so failing readiness on a Redis blip would
+// pull healthy replicas out of the load balancer and convert a degraded
+// feature into an outage. This type exists to make the degradation
+// VISIBLE, not to act on it.
+//
+// TIMEOUT NOTE, measured on BUG-2698's day-50 run rather than assumed:
+// go-redis does NOT apply a command context to connection ESTABLISHMENT.
+// A call with a 150ms context against a server that accepts connections
+// and never answers took 5.0s — the client's DialTimeout. So the probe's
+// context bounds the PING once a connection exists and nothing else; the
+// dial is bounded by the client's own DialTimeout (go-redis v9 default
+// 5s, and cmd/pad/cmd_server.go takes the default). That is why the probe
+// interval is set well above the dial bound rather than just above the
+// probe timeout.
+type RedisHealth struct {
+	client   *redis.Client
+	interval time.Duration
+	timeout  time.Duration
+
+	// onProbe, when non-nil, is called after every probe with its result.
+	// It is how the Prometheus gauge is written without this type
+	// importing the metrics package.
+	onProbe func(ok bool)
+
+	mu        sync.RWMutex
+	probed    bool
+	ok        bool
+	lastErr   string
+	lastCheck time.Time
+
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+}
+
+// RedisHealthStatus is the cached probe result, as reported by
+// /health/ready.
+type RedisHealthStatus struct {
+	// Reachable is the last probe's verdict. Meaningless until Probed.
+	Reachable bool `json:"reachable"`
+	// Probed is false in the window between the server accepting traffic
+	// and the first probe completing. Reported rather than defaulted:
+	// "not measured yet" and "measured, and it is down" are different
+	// facts, and only one of them is worth waking someone for.
+	Probed bool `json:"probed"`
+	// Error is the last probe's failure text, empty when Reachable.
+	Error string `json:"error,omitempty"`
+	// LastCheck is when the last probe completed, zero before the first.
+	LastCheck time.Time `json:"last_check,omitempty"`
+}
+
+// NewRedisHealth builds a prober for client. onProbe may be nil.
+func NewRedisHealth(client *redis.Client, onProbe func(ok bool)) *RedisHealth {
+	return &RedisHealth{
+		client:   client,
+		interval: redisProbeInterval,
+		timeout:  redisProbeTimeout,
+		onProbe:  onProbe,
+	}
+}
+
+// Start begins probing in the background. It runs one probe SYNCHRONOUSLY
+// before returning, so a server that has finished starting has a real
+// answer rather than a Probed=false window that only closes an interval
+// later — the boot case is exactly when an operator is watching.
+func (h *RedisHealth) Start() {
+	if h == nil || h.client == nil {
+		return
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	h.cancel = cancel
+
+	h.probe(ctx)
+
+	h.wg.Add(1)
+	go func() {
+		defer h.wg.Done()
+		ticker := time.NewTicker(h.interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				h.probe(ctx)
+			}
+		}
+	}()
+}
+
+// Stop ends probing and waits for the goroutine to exit. Idempotent.
+func (h *RedisHealth) Stop() {
+	if h == nil || h.cancel == nil {
+		return
+	}
+	h.cancel()
+	h.wg.Wait()
+}
+
+func (h *RedisHealth) probe(ctx context.Context) {
+	probeCtx, cancel := context.WithTimeout(ctx, h.timeout)
+	err := h.client.Ping(probeCtx).Err()
+	cancel()
+
+	// A cancelled parent means we are shutting down, not that Redis is
+	// gone. Recording it as a failure would flip pad_redis_up to 0 on
+	// every clean shutdown and teach whoever alerts on it to ignore the
+	// signal.
+	if ctx.Err() != nil {
+		return
+	}
+
+	ok := err == nil
+
+	h.mu.Lock()
+	wasProbed, was := h.probed, h.ok
+	h.probed = true
+	h.ok = ok
+	h.lastCheck = time.Now()
+	if ok {
+		h.lastErr = ""
+	} else {
+		h.lastErr = err.Error()
+	}
+	h.mu.Unlock()
+
+	// Log only on TRANSITIONS (and on the first probe), not every tick: a
+	// Redis outage lasting an hour would otherwise write 240 identical
+	// lines, which is the same log-flood shape the presence renewal
+	// warning had to be throttled for.
+	switch {
+	case !wasProbed && !ok:
+		slog.Error("redis: unreachable at startup; watch notifications, cross-instance events and session presence are degraded", "error", err)
+	case wasProbed && was && !ok:
+		slog.Error("redis: became unreachable; watch notifications, cross-instance events and session presence are degraded", "error", err)
+	case wasProbed && !was && ok:
+		slog.Info("redis: reachable again")
+	}
+
+	if h.onProbe != nil {
+		h.onProbe(ok)
+	}
+}
+
+// Status returns the cached probe result.
+func (h *RedisHealth) Status() RedisHealthStatus {
+	if h == nil {
+		return RedisHealthStatus{}
+	}
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return RedisHealthStatus{
+		Reachable: h.ok,
+		Probed:    h.probed,
+		Error:     h.lastErr,
+		LastCheck: h.lastCheck,
+	}
+}

internal/server/server.go

@@ -48,6 +48,7 @@ type Server struct {
 	events                events.EventBus      // real-time event bus (optional)
 	watchEvents           watchevents.Bus      // watch/nudge notification bus (optional, TASK-2533)
 	sessionPresence       SessionPresence      // live event-stream connections per user (optional, PLAN-2558 S1)
+	redisHealth           *RedisHealth         // cached Redis reachability, reported by /health/ready and pad_redis_up (optional, BUG-2727)
 	collab                *collab.RoomManager  // Yjs collab room manager (PLAN-1248); optional
 	webhooks              *webhooks.Dispatcher // webhook dispatcher (optional)
 	email                 *email.Sender        // transactional email sender (optional)
@@ -795,6 +796,18 @@ func (s *Server) SetSessionPresence(p SessionPresence) {
 	s.sessionPresence = p
 }
 
+// SetRedisHealth attaches the Redis reachability prober (BUG-2727). Nil
+// on a deployment with no Redis, in which case /health/ready reports no
+// redis block at all — the honest shape, since "no Redis configured" and
+// "Redis configured and down" are different states and a `false` would
+// merge them.
+//
+// It does NOT make readiness depend on Redis; see RedisHealth's doc
+// comment for why that is deliberate.
+func (s *Server) SetRedisHealth(h *RedisHealth) {
+	s.redisHealth = h
+}
+
 // SetCollabRoomManager attaches a Yjs collab RoomManager (PLAN-1248).
 // When set, the /api/v1/collab/{itemID} WebSocket endpoint hands new
 // connections to the manager for op-log replay + fan-out. When nil,

internal/server/session_presence_observer.go

@@ -0,0 +1,91 @@
+package server
+
+import (
+	"sync"
+
+	"github.com/PerpetualSoftware/pad/internal/metrics"
+)
+
+// PresenceObserver receives operational events from the Redis-backed
+// session-presence registry (BUG-2727).
+//
+// The registry is deliberately FAIL-SOFT everywhere: a failed write leaves
+// a live session unlisted rather than tearing down the SSE connection it
+// belongs to, and a failed read answers the sessions endpoint with a 503
+// rather than an empty list. That posture is right, and it is exactly why
+// the failures need a counter — a fail-soft subsystem reports nothing to
+// the user beyond a push that quietly reaches fewer sessions than it
+// should, so a log line is the only trace and log lines are not alertable
+// without someone already looking.
+//
+// Implementations must be safe for concurrent use and must not block.
+type PresenceObserver interface {
+	// PresenceOpFailed reports one failed Redis operation. op is a bounded
+	// label from the PresenceOp* constants, never free text — it becomes a
+	// metric label.
+	PresenceOpFailed(op string)
+}
+
+// Presence operation labels. Bounded by construction so they are safe as
+// metric label values.
+const (
+	// PresenceOpRegister — the initial write for a newly connected
+	// session. It leaves the session out of the picker until the first
+	// successful renewal.
+	PresenceOpRegister = "register"
+	// PresenceOpRenew — the keepalive write. Sustained failure lets the
+	// entry expire on its TTL while the connection is still live.
+	PresenceOpRenew = "renew"
+	// PresenceOpDeregister — the explicit removal on disconnect. Failure
+	// leaves a dead session listed until its TTL expires, so the picker
+	// over-reports.
+	PresenceOpDeregister = "deregister"
+	// PresenceOpList — reading a user's sessions. Failure surfaces to the
+	// caller as a 503 rather than as a short list.
+	PresenceOpList = "list"
+	// PresenceOpPrune — removing index members whose session key is gone.
+	// Failure is the most benign: the index carries a stale member that
+	// the next read skips and the next prune retries.
+	PresenceOpPrune = "prune"
+)
+
+// presenceObservable is the nil-safe holder RedisSessionPresence embeds.
+type presenceObservable struct {
+	obsMu sync.RWMutex
+	obs   PresenceObserver
+}
+
+// SetObserver attaches a PresenceObserver; nil detaches. Safe to call
+// after the registry is running.
+func (o *presenceObservable) SetObserver(obs PresenceObserver) {
+	o.obsMu.Lock()
+	o.obs = obs
+	o.obsMu.Unlock()
+}
+
+func (o *presenceObservable) reportOpFailed(op string) {
+	o.obsMu.RLock()
+	obs := o.obs
+	o.obsMu.RUnlock()
+	if obs != nil {
+		obs.PresenceOpFailed(op)
+	}
+}
+
+// metricsPresenceObserver adapts the Prometheus metrics into a
+// PresenceObserver. It lives here rather than in internal/metrics because
+// internal/server already imports internal/metrics; the reverse would be
+// an import cycle.
+type metricsPresenceObserver struct {
+	m *metrics.Metrics
+}
+
+func (o metricsPresenceObserver) PresenceOpFailed(op string) {
+	o.m.SessionPresenceFailuresTotal.WithLabelValues(op).Inc()
+}
+
+// NewMetricsPresenceObserver returns a PresenceObserver that writes into
+// m's Prometheus counters. Exported because the wiring lives in cmd/pad.
+func NewMetricsPresenceObserver(m *metrics.Metrics) PresenceObserver {
+	return metricsPresenceObserver{m: m}
+}

internal/server/session_presence_redis.go

@@ -129,6 +129,10 @@ import (
 // busy. Consumers must keep treating this list as "connected as far as the
 // server can tell", never as a delivery guarantee.
 type RedisSessionPresence struct {
+	// presenceObservable carries the optional operational-event seam
+	// (BUG-2727). Embedded so SetObserver is part of the type's surface.
+	presenceObservable
+
 	client *redis.Client
 
 	// sessionKeyTTL and renewInterval are fields rather than constants so
@@ -447,6 +451,7 @@ func (p *RedisSessionPresence) Add(userID string, ident SessionIdentity) string
 		// registration that failed on a blip.
 		slog.Warn("session presence: failed to register session; it will be missing from the picker until the next renewal",
 			"error", err, "user_id", userID)
+		p.reportOpFailed(PresenceOpRegister)
 	}
 
 	go p.renewLoop(ctx, rn, userID, id, string(payload))
@@ -512,6 +517,14 @@ func (p *RedisSessionPresence) renewLoop(ctx context.Context, rn *renewal, userI
 			}
 			if err := p.write(ctx, userID, sessionID, payload); err != nil && ctx.Err() == nil {
 				p.warnRenewFailure(err, userID, sessionID)
+				// Counted on EVERY failure, while the log line above is
+				// rate-limited (that throttle exists because a Redis
+				// outage otherwise logs once per session per renewal —
+				// roughly 33 lines a second per replica at a thousand
+				// sessions). A counter has no such problem and wants the
+				// true rate: throttling it too would make the metric
+				// under-report exactly during the incident it exists for.
+				p.reportOpFailed(PresenceOpRenew)
 			}
 		}
 	}
@@ -588,6 +601,7 @@ func (p *RedisSessionPresence) Remove(userID string, sessionID string) {
 	if _, err := pipe.Exec(ctx); err != nil {
 		slog.Warn("session presence: failed to deregister session; it will expire on its TTL",
 			"error", err, "user_id", userID)
+		p.reportOpFailed(PresenceOpDeregister)
 	}
 }
 
@@ -617,6 +631,7 @@ func (p *RedisSessionPresence) ListForUser(userID string) ([]LiveSession, error)
 		// anything. Collapsing the two here is the same defect this type
 		// was written to remove, one layer down.
 		slog.Warn("session presence: failed to read session index", "error", err, "user_id", userID)
+		p.reportOpFailed(PresenceOpList)
 		return nil, fmt.Errorf("session presence: read index: %w", err)
 	}
 	if len(ids) == 0 {
@@ -638,6 +653,7 @@ func (p *RedisSessionPresence) ListForUser(userID string) ([]LiveSession, error)
 	values, err := p.client.MGet(ctx, keys...).Result()
 	if err != nil {
 		slog.Warn("session presence: failed to read session entries", "error", err, "user_id", userID)
+		p.reportOpFailed(PresenceOpList)
 		return nil, fmt.Errorf("session presence: read entries: %w", err)
 	}
 
@@ -701,6 +717,7 @@ func (p *RedisSessionPresence) ListForUser(userID string) ([]LiveSession, error)
 		).Err(); err != nil && !errors.Is(err, redis.Nil) {
 			slog.Debug("session presence: failed to prune expired index members",
 				"error", err, "user_id", userID)
+			p.reportOpFailed(PresenceOpPrune)
 		}
 	}
 

internal/watchevents/observer.go

@@ -0,0 +1,121 @@
+package watchevents
+
+import "sync"
+
+// Observer receives OPERATIONAL events from a Bus — the conditions an
+// operator would want to alert on, as opposed to the notifications the bus
+// exists to deliver. It is the seam internal/metrics implements so this
+// package can stay free of a Prometheus dependency, mirroring the reason
+// events.EventBus is instrumented by a wrapper rather than by importing
+// metrics.
+//
+// A wrapper does not work HERE, though, and that asymmetry is the whole
+// reason this interface exists: every condition below is detected INSIDE
+// the bus, on the receive path, and is invisible at the Bus interface. A
+// caller sees Publish succeed and Subscribe return a channel whether or
+// not this instance is quietly missing notifications.
+//
+// WHAT THIS DOES AND DOES NOT SEE (BUG-2727). These callbacks report what
+// PAD detects. They do NOT hook go-redis, which has its own silent drop:
+// it buffers 100 messages per subscription and discards further ones after
+// a 60s send timeout, logging only through its own internal logger. That
+// drop is not reported here directly — it is reported by its CONSEQUENCE,
+// as a SequenceGap, because a discarded message leaves a hole in the id
+// sequence exactly like a message lost across a reconnect does. So
+// SequenceGap counts "this instance missed something", not "Redis was at
+// fault"; do not read a gap as evidence of any particular cause.
+//
+// Implementations must be safe for concurrent use and must not block —
+// they are called while the bus holds its mutex on the receive path.
+type Observer interface {
+	// NotificationDropped reports a notification this instance received but
+	// could not deliver to one of its own subscribers. reason is a bounded
+	// label ("slow_subscriber"), never free text — it becomes a metric
+	// label.
+	NotificationDropped(reason string)
+
+	// SequenceGap reports that the received id sequence skipped forward:
+	// this instance missed `missing` notifications. Resumes across the gap
+	// are answered with sync_required (see fanOutLocally); subscribers
+	// holding a stream open across it are told nothing, which is BUG-2730
+	// and not this seam's job to fix.
+	SequenceGap(missing int64)
+
+	// SequenceReset reports that the id space itself changed under this
+	// instance and the replay buffer was dropped. reason is bounded:
+	// "epoch_change" (the shared epoch key changed) or "counter_backwards"
+	// (an id arrived at or below the high-water mark).
+	SequenceReset(reason string)
+
+	// ReceiveLoopExited reports that the single consumer of the shared
+	// Redis channel has stopped. After this the instance publishes fine and
+	// receives nothing — including its own publishes, which are delivered
+	// through Redis like everyone else's.
+	//
+	// Reachability, verified against go-redis v9.22.0 rather than assumed:
+	// PubSub.Channel's message channel is closed ONLY on pool.ErrClosed,
+	// i.e. the client or the PubSub was closed. Every other receive error
+	// is retried indefinitely, and a health-check goroutine pings every 3s
+	// and reconnects (re-subscribing) on failure. So in practice this fires
+	// at shutdown, and a firing OUTSIDE shutdown means the client was
+	// closed underneath the bus — which is why it is worth a counter and an
+	// ERROR log rather than a silent return.
+	ReceiveLoopExited()
+}
+
+// observable is the shared, nil-safe Observer holder both bus
+// implementations embed. Reporting through it before SetObserver is
+// called — which is every bus in every test that does not opt in, and
+// every single-process deployment — is a no-op.
+type observable struct {
+	obsMu sync.RWMutex
+	obs   Observer
+}
+
+// SetObserver attaches an Observer. Passing nil detaches. Safe to call
+// after the bus is running; cmd/pad/cmd_server.go calls it at wiring time.
+func (o *observable) SetObserver(obs Observer) {
+	o.obsMu.Lock()
+	o.obs = obs
+	o.obsMu.Unlock()
+}
+
+func (o *observable) observer() Observer {
+	o.obsMu.RLock()
+	obs := o.obs
+	o.obsMu.RUnlock()
+	return obs
+}
+
+func (o *observable) reportDropped(reason string) {
+	if obs := o.observer(); obs != nil {
+		obs.NotificationDropped(reason)
+	}
+}
+
+func (o *observable) reportGap(missing int64) {
+	if obs := o.observer(); obs != nil {
+		obs.SequenceGap(missing)
+	}
+}
+
+func (o *observable) reportReset(reason string) {
+	if obs := o.observer(); obs != nil {
+		obs.SequenceReset(reason)
+	}
+}
+
+func (o *observable) reportReceiveLoopExited() {
+	if obs := o.observer(); obs != nil {
+		obs.ReceiveLoopExited()
+	}
+}
+
+// Drop reasons and reset reasons. Bounded by construction so they can be
+// metric labels without a cardinality risk.
+const (
+	DropReasonSlowSubscriber = "slow_subscriber"
+
+	ResetReasonEpochChange     = "epoch_change"
+	ResetReasonCounterBackward = "counter_backwards"
+)

internal/watchevents/observer_test.go

@@ -0,0 +1,267 @@
+package watchevents
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+// recordingObserver is the instrument under which the tests below run. It
+// counts, rather than merely noting, so a test can distinguish "reported
+// once" from "reported on every subscriber in the loop" — a real
+// difference, since the drop site lives inside a per-subscriber loop.
+type recordingObserver struct {
+	mu sync.Mutex
+
+	dropped     map[string]int
+	gaps        int
+	missed      int64
+	resets      map[string]int
+	loopExits   int
+	totalEvents int
+}
+
+func newRecordingObserver() *recordingObserver {
+	return &recordingObserver{
+		dropped: map[string]int{},
+		resets:  map[string]int{},
+	}
+}
+
+func (o *recordingObserver) NotificationDropped(reason string) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.dropped[reason]++
+	o.totalEvents++
+}
+
+func (o *recordingObserver) SequenceGap(missing int64) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.gaps++
+	o.missed += missing
+	o.totalEvents++
+}
+
+func (o *recordingObserver) SequenceReset(reason string) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.resets[reason]++
+	o.totalEvents++
+}
+
+func (o *recordingObserver) ReceiveLoopExited() {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.loopExits++
+	o.totalEvents++
+}
+
+func (o *recordingObserver) snapshot() recordingObserver {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	cp := recordingObserver{
+		dropped:     map[string]int{},
+		resets:      map[string]int{},
+		gaps:        o.gaps,
+		missed:      o.missed,
+		loopExits:   o.loopExits,
+		totalEvents: o.totalEvents,
+	}
+	for k, v := range o.dropped {
+		cp.dropped[k] = v
+	}
+	for k, v := range o.resets {
+		cp.resets[k] = v
+	}
+	return cp
+}
+
+// TestMemoryBusReportsSlowSubscriberDrop asserts the DROP increments the
+// observer — not merely that an observer can be attached.
+//
+// It asserts its own premise first: a subscriber with room receives the
+// notification and produces NO drop report. Without that leg the test
+// would pass against a bus that reported a drop on every publish, which
+// is a differently-broken instrument that looks identical from the
+// assertion that only checks the full case.
+func TestMemoryBusReportsSlowSubscriberDrop(t *testing.T) {
+	t.Parallel()
+
+	b := New()
+	defer b.Close()
+	obs := newRecordingObserver()
+	b.SetObserver(obs)
+
+	ch := b.Subscribe()
+
+	// PREMISE: a healthy subscriber is delivered to and reports nothing.
+	if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-1"}); err != nil {
+		t.Fatalf("publish: %v", err)
+	}
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatal("premise failed: a subscriber with room did not receive the notification")
+	}
+	if got := obs.snapshot(); got.totalEvents != 0 {
+		t.Fatalf("premise failed: healthy delivery reported %d observer events, want 0 (%+v)", got.totalEvents, got)
+	}
+
+	// Fill the subscriber's 64-deep buffer without draining it.
+	for i := 0; i < 64; i++ {
+		if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-2"}); err != nil {
+			t.Fatalf("publish %d: %v", i, err)
+		}
+	}
+	if got := obs.snapshot(); got.totalEvents != 0 {
+		t.Fatalf("premise failed: filling the buffer to capacity reported %d events, want 0 (%+v)", got.totalEvents, got)
+	}
+
+	// The 65th has nowhere to go.
+	if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-3"}); err != nil {
+		t.Fatalf("overflow publish: %v", err)
+	}
+
+	got := obs.snapshot()
+	if got.dropped[DropReasonSlowSubscriber] != 1 {
+		t.Fatalf("dropped[%s] = %d, want 1 (%+v)", DropReasonSlowSubscriber, got.dropped[DropReasonSlowSubscriber], got)
+	}
+	if got.totalEvents != 1 {
+		t.Fatalf("total observer events = %d, want exactly 1 — the drop and nothing else (%+v)", got.totalEvents, got)
+	}
+}
+
+// TestRedisBusReportsSequenceGap drives fanOutLocally directly, which is
+// where the detection lives, and pins BOTH numbers: that a gap was
+// reported once, and how many notifications it spanned. One counter
+// cannot distinguish one gap of 500 from 500 gaps of one, which is why
+// the observer reports both and why this asserts both.
+func TestRedisBusReportsSequenceGap(t *testing.T) {
+	t.Parallel()
+
+	b, _ := newMiniredisBus(t, 64)
+	obs := newRecordingObserver()
+	b.SetObserver(obs)
+
+	// PREMISE: contiguous ids report nothing. Without this leg a bus that
+	// reported a gap on every notification would pass the assertion below.
+	b.fanOutLocally(Notification{ID: 1, Kind: KindPush})
+	b.fanOutLocally(Notification{ID: 2, Kind: KindPush})
+	if got := obs.snapshot(); got.totalEvents != 0 {
+		t.Fatalf("premise failed: contiguous ids 1,2 reported %d observer events, want 0 (%+v)", got.totalEvents, got)
+	}
+
+	// 3 and 4 are missing.
+	b.fanOutLocally(Notification{ID: 5, Kind: KindPush})
+
+	got := obs.snapshot()
+	if got.gaps != 1 {
+		t.Fatalf("gaps = %d, want 1 (%+v)", got.gaps, got)
+	}
+	if got.missed != 2 {
+		t.Fatalf("missed = %d, want 2 — ids 3 and 4 (%+v)", got.missed, got)
+	}
+}
+
+// TestRedisBusReportsSequenceResets covers both reset reasons, which are
+// detected in different functions (fanOutFromRedis for the epoch,
+// fanOutLocally for the counter) and would be easy to wire one of and
+// call done.
+func TestRedisBusReportsSequenceResets(t *testing.T) {
+	t.Parallel()
+
+	t.Run("epoch change", func(t *testing.T) {
+		b, _ := newMiniredisBus(t, 64)
+		obs := newRecordingObserver()
+		b.SetObserver(obs)
+
+		b.fanOutFromRedis("epoch-a", Notification{ID: 1, Kind: KindPush})
+		if got := obs.snapshot(); got.totalEvents != 0 {
+			t.Fatalf("premise failed: the first epoch seen reported %d events, want 0 (%+v)", got.totalEvents, got)
+		}
+
+		b.fanOutFromRedis("epoch-b", Notification{ID: 1, Kind: KindPush})
+
+		got := obs.snapshot()
+		if got.resets[ResetReasonEpochChange] != 1 {
+			t.Fatalf("resets[%s] = %d, want 1 (%+v)", ResetReasonEpochChange, got.resets[ResetReasonEpochChange], got)
+		}
+	})
+
+	t.Run("counter backwards", func(t *testing.T) {
+		b, _ := newMiniredisBus(t, 64)
+		obs := newRecordingObserver()
+		b.SetObserver(obs)
+
+		b.fanOutLocally(Notification{ID: 100, Kind: KindPush})
+		if got := obs.snapshot(); got.totalEvents != 0 {
+			t.Fatalf("premise failed: a cold start reported %d events, want 0 (%+v)", got.totalEvents, got)
+		}
+
+		// The shared counter restarted: an id at or below the high-water
+		// mark.
+		b.fanOutLocally(Notification{ID: 1, Kind: KindPush})
+
+		got := obs.snapshot()
+		if got.resets[ResetReasonCounterBackward] != 1 {
+			t.Fatalf("resets[%s] = %d, want 1 (%+v)", ResetReasonCounterBackward, got.resets[ResetReasonCounterBackward], got)
+		}
+		if got.gaps != 0 {
+			t.Fatalf("gaps = %d, want 0 — a backwards counter is a reset, not a gap (%+v)", got.gaps, got)
+		}
+	})
+}
+
+// TestRedisBusReportsReceiveLoopExit closes the client underneath a
+// running bus, which is the ONE condition go-redis actually closes a
+// subscription's message channel for (pool.ErrClosed — every other
+// receive error is retried indefinitely and a health-check goroutine
+// reconnects). Driving the real condition rather than calling the
+// reporter directly is the difference between testing the wiring and
+// testing the constant.
+func TestRedisBusReportsReceiveLoopExit(t *testing.T) {
+	t.Parallel()
+
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+
+	b := NewRedisBusWithReplaySize(client, 64)
+	defer b.Close()
+	obs := newRecordingObserver()
+	b.SetObserver(obs)
+
+	// PREMISE: the bus is genuinely receiving before we break it —
+	// otherwise a bus that never started would pass by reporting an exit
+	// for the wrong reason.
+	ch := b.Subscribe()
+	if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-1"}); err != nil {
+		t.Fatalf("premise publish: %v", err)
+	}
+	select {
+	case <-ch:
+	case <-time.After(3 * time.Second):
+		t.Fatal("premise failed: the bus never delivered a round-tripped notification")
+	}
+	if got := obs.snapshot(); got.loopExits != 0 {
+		t.Fatalf("premise failed: %d loop exits before closing the client (%+v)", got.loopExits, got)
+	}
+
+	// Close the CLIENT, not the bus: b.ctx stays live, so the receive
+	// loop takes the closed-channel branch rather than the ctx branch.
+	if err := client.Close(); err != nil {
+		t.Fatalf("close client: %v", err)
+	}
+
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if obs.snapshot().loopExits == 1 {
+			return
+		}
+		time.Sleep(2 * time.Millisecond)
+	}
+	t.Fatalf("receive loop exit was never reported (%+v)", obs.snapshot())
+}

internal/watchevents/redis_bus.go

@@ -172,6 +172,11 @@ const (
 // is bounded: DefaultReplayBufferSize entries, after which a resume answers
 // "gap too large" and the consumer resyncs.
 type RedisBus struct {
+	// observable carries the optional operational-event seam (BUG-2727).
+	// Embedded first so SetObserver is part of the type's own surface
+	// rather than something a caller reaches through a field.
+	observable
+
 	client *redis.Client
 
 	// mu guards subscribers AND replay together — see the type comment (3)
@@ -695,6 +700,22 @@ func (b *RedisBus) receiveMessages() {
 			return
 		case msg, ok := <-ch:
 			if !ok {
+				// The subscription's message channel closed. go-redis
+				// (v9.22.0) closes it ONLY on pool.ErrClosed — the client
+				// or the PubSub was closed — and retries every other
+				// receive error indefinitely while a health-check
+				// goroutine reconnects on ping failure. So the ordinary
+				// cause is our own shutdown, which cancels b.ctx and is
+				// caught by the case above; arriving HERE instead means
+				// the client went away underneath a bus that is still
+				// running, and from this point the instance publishes
+				// fine and receives nothing at all — including its own
+				// publishes, which come back through Redis like everyone
+				// else's. Silence was the original behaviour and made
+				// that state indistinguishable from a quiet workspace.
+				slog.Error("watchevents: Redis subscription closed; this instance will receive no further notifications " +
+					"(publishes still succeed, so nothing else will report this)")
+				b.reportReceiveLoopExited()
 				return
 			}
 			epoch, n, err := decodePayload(msg.Payload)
@@ -768,6 +789,7 @@ func (b *RedisBus) fanOutFromRedis(epoch string, n Notification) {
 		b.lastAppendedID = 0
 		b.knownFrom = 0
 		b.epochJustChanged = true
+		b.reportReset(ResetReasonEpochChange)
 	}
 	b.mu.Unlock()
 
@@ -837,11 +859,14 @@ func (b *RedisBus) fanOutLocally(n Notification) {
 			"previous", b.lastAppendedID, "got", n.ID)
 		b.replay = newReplayBuffer(b.replaySize)
 		b.knownFrom = n.ID
+		b.reportReset(ResetReasonCounterBackward)
 
 	case n.ID != b.lastAppendedID+1:
 		slog.Warn("watchevents: gap in the received notification sequence; resumes across it will report sync_required",
-			"expected", b.lastAppendedID+1, "got", n.ID)
+			"expected", b.lastAppendedID+1, "got", n.ID,
+			"missed", n.ID-b.lastAppendedID-1)
 		b.knownFrom = n.ID
+		b.reportGap(n.ID - b.lastAppendedID - 1)
 	}
 	b.lastAppendedID = n.ID
 
@@ -853,6 +878,7 @@ func (b *RedisBus) fanOutLocally(n Notification) {
 		default:
 			slog.Warn("watchevents: dropping notification for slow subscriber",
 				"kind", n.Kind, "item_ref", n.ItemRef)
+			b.reportDropped(DropReasonSlowSubscriber)
 		}
 	}
 }

internal/watchevents/watchevents.go

@@ -328,6 +328,13 @@ func (rb *replayBuffer) since(sinceID int64) []Notification {
 // (codex round 2 finding 3 — see Publish's doc comment for why an
 // earlier "send after releasing the lock" version could panic).
 type MemoryBus struct {
+	// observable carries the optional operational-event seam (BUG-2727).
+	// A single-process deployment never wires one, so every report below
+	// is a nil check; it is here so the drop condition is reported by the
+	// SAME name on both implementations rather than existing only on the
+	// one that happens to have Redis behind it.
+	observable
+
 	mu          sync.Mutex
 	subscribers map[chan Notification]struct{}
 	seq         int64
@@ -412,6 +419,7 @@ func (b *MemoryBus) Publish(n Notification) error {
 			// a delivery it cannot confirm either way had definitely
 			// failed — exactly the false precision BUG-2699 is about.
 			slog.Warn("watchevents: dropping notification for slow subscriber", "kind", n.Kind, "item_ref", n.ItemRef)
+			b.reportDropped(DropReasonSlowSubscriber)
 		}
 	}
 	return nil