diff --git a/adapter/redis.go b/adapter/redis.go
index 928f37aee..cddb548b8 100644
--- a/adapter/redis.go
+++ b/adapter/redis.go
@@ -221,10 +221,15 @@ type RedisServer struct {
 	// Exposed via HELLO / CLIENT ID.
 	connIDSeq atomic.Uint64
 
-	// streamWaiters lets XADD wake an XREAD BLOCK waiting on the same
+	// streamWaiters lets XADD wake an XREAD BLOCK waiter on the same
 	// node, replacing what was a 10 ms time.Sleep busy-poll. See
-	// redis_stream_waiters.go.
-	streamWaiters *streamWaiterRegistry
+	// redis_key_waiters.go.
+	streamWaiters *keyWaiterRegistry
+
+	// zsetWaiters lets ZADD / ZINCRBY wake a BZPOPMIN waiter on the
+	// same node, replacing what was a 10 ms time.Sleep busy-poll. See
+	// redis_key_waiters.go.
+	zsetWaiters *keyWaiterRegistry
 
 	route map[string]func(conn redcon.Conn, cmd redcon.Command)
 }
@@ -353,7 +358,8 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 		traceCommands:   os.Getenv("ELASTICKV_REDIS_TRACE") == "1",
 		baseCtx:         baseCtx,
 		baseCancel:      baseCancel,
-		streamWaiters:   newStreamWaiterRegistry(),
+		streamWaiters:   newKeyWaiterRegistry(),
+		zsetWaiters:     newKeyWaiterRegistry(),
 	}
 	r.relay.Bind(r.publishLocal)
 
diff --git a/adapter/redis_bzpopmin_wake_test.go b/adapter/redis_bzpopmin_wake_test.go
new file mode 100644
index 000000000..96d3fb3ea
--- /dev/null
+++ b/adapter/redis_bzpopmin_wake_test.go
@@ -0,0 +1,137 @@
+package adapter
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRedis_BZPopMinWakesOnZAdd verifies the event-driven wake path:
+// an in-process ZADD on the leader's redis adapter must wake a
+// BZPOPMIN waiter on the same node so the reader returns the new
+// entry before its BLOCK deadline. The wake comes through
+// keyWaiterRegistry's signal channel — the prior 10 ms time.Sleep
+// busy-poll loop would have exhibited the same end-to-end behaviour,
+// so this is an end-to-end sanity test rather than a wall-clock
+// latency gate (the latency gate is impractical under -race +
+// parallel CI load, where tryBZPopMin's Pebble seek alone can
+// exceed any tight budget).
+//
+// Both client connections target the same node so they share the
+// same keyWaiterRegistry — the signal path is intentionally
+// in-process only (Lua and follower-side applies fall through to
+// the fallback timer; see bzpopminWaitLoop).
+func TestRedis_BZPopMinWakesOnZAdd(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	rdbReader := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbReader.Close() }()
+	rdbWriter := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbWriter.Close() }()
+	ctx := context.Background()
+
+	type popResult struct {
+		zwk *redis.ZWithKey
+		err error
+	}
+	resultCh := make(chan popResult, 1)
+	go func() {
+		zwk, err := rdbReader.BZPopMin(ctx, 5*time.Second, "zset-wake").Result()
+		resultCh <- popResult{zwk: zwk, err: err}
+	}()
+
+	// Give the reader a moment to enter bzpopminWaitLoop and register
+	// a waiter on keyWaiterRegistry before ZADD. If ZADD landed first
+	// the entry would already be visible by the time the reader runs
+	// tryBZPopMin, so the registration race is benign — but waiting
+	// also gates out a different source of flake where the goroutine
+	// has not yet dialed redis.
+	time.Sleep(50 * time.Millisecond)
+
+	_, err := rdbWriter.ZAdd(ctx, "zset-wake",
+		redis.Z{Score: 1, Member: "first"},
+	).Result()
+	require.NoError(t, err)
+
+	select {
+	case res := <-resultCh:
+		require.NoError(t, res.err)
+		require.NotNil(t, res.zwk)
+		require.Equal(t, "zset-wake", res.zwk.Key)
+		require.Equal(t, "first", res.zwk.Member)
+		require.InDelta(t, 1.0, res.zwk.Score, 1e-9)
+	case <-time.After(6 * time.Second):
+		t.Fatal("BZPOPMIN did not return after ZADD signal")
+	}
+}
+
+// TestRedis_BZPopMinWakesOnZIncrBy verifies the same wake path but
+// driven by ZINCRBY rather than ZADD. ZINCRBY on a missing member
+// is functionally equivalent to ZADD for the BZPOPMIN consumer:
+// the score appears under a new member, which makes the zset
+// non-empty and BZPOPMIN should return it.
+func TestRedis_BZPopMinWakesOnZIncrBy(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	rdbReader := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbReader.Close() }()
+	rdbWriter := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbWriter.Close() }()
+	ctx := context.Background()
+
+	type popResult struct {
+		zwk *redis.ZWithKey
+		err error
+	}
+	resultCh := make(chan popResult, 1)
+	go func() {
+		zwk, err := rdbReader.BZPopMin(ctx, 5*time.Second, "zset-incr-wake").Result()
+		resultCh <- popResult{zwk: zwk, err: err}
+	}()
+
+	time.Sleep(50 * time.Millisecond)
+
+	_, err := rdbWriter.ZIncrBy(ctx, "zset-incr-wake", 7.5, "alpha").Result()
+	require.NoError(t, err)
+
+	select {
+	case res := <-resultCh:
+		require.NoError(t, res.err)
+		require.NotNil(t, res.zwk)
+		require.Equal(t, "zset-incr-wake", res.zwk.Key)
+		require.Equal(t, "alpha", res.zwk.Member)
+		require.InDelta(t, 7.5, res.zwk.Score, 1e-9)
+	case <-time.After(6 * time.Second):
+		t.Fatal("BZPOPMIN did not return after ZINCRBY signal")
+	}
+}
+
+// TestRedis_BZPopMinTimesOutOnEmptyKey locks down the BLOCK-timeout
+// contract: when no ZADD arrives within the BLOCK window, BZPOPMIN
+// returns redis.Nil rather than a protocol error. This guards a
+// regression in the wait-loop refactor where the new
+// waitForBlockedCommandUpdate timer or context-cancel branch could otherwise
+// leak a -ERR reply.
+func TestRedis_BZPopMinTimesOutOnEmptyKey(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdb.Close() }()
+	ctx := context.Background()
+
+	// 250 ms BLOCK on a key that never receives a write. The fallback
+	// timer (100 ms) fires twice, then the deadline branch writes
+	// nil. Total budget: redisDispatchTimeout caps each iter; we
+	// expect ~250 ms total wall time and a redis.Nil reply.
+	zwk, err := rdb.BZPopMin(ctx, 250*time.Millisecond, "zset-empty").Result()
+	require.ErrorIs(t, err, redis.Nil, "BLOCK timeout must return redis.Nil, got zwk=%v err=%v", zwk, err)
+}
diff --git a/adapter/redis_compat_commands.go b/adapter/redis_compat_commands.go
index 53a304383..da798eb38 100644
--- a/adapter/redis_compat_commands.go
+++ b/adapter/redis_compat_commands.go
@@ -23,21 +23,23 @@ import (
 )
 
 const (
-	redisPairWidth       = 2
-	redisTripletWidth    = 3
-	pubsubPatternArgMin  = 3
-	pubsubFirstChannel   = 2
-	redisBusyPollBackoff = 10 * time.Millisecond
-	// redisStreamWaitFallback is the safety-net poll interval that fires
-	// in xreadBusyPoll when no XADD signal arrives. The signal path covers
-	// all in-process XADDs on the same node; the fallback only matters
-	// when the entry was applied via a path that does not call Signal
-	// (e.g. follower-side FSM apply, future direct mutation paths).
+	redisPairWidth      = 2
+	redisTripletWidth   = 3
+	pubsubPatternArgMin = 3
+	pubsubFirstChannel  = 2
+	// redisBlockWaitFallback is the safety-net poll interval that fires
+	// in blocking-command wait loops (XREAD BLOCK, BZPOPMIN — and the
+	// future BLPOP / BRPOP / BLMOVE) when no in-process write signal
+	// arrives. The signal path covers all in-process XADD / ZADD /
+	// ZINCRBY on the same node; the fallback covers paths that bypass
+	// Signal (Lua flush, follower-applied entries — both addressed by
+	// the FSM ApplyObserver follow-up tracked in
+	// docs/design/2026_04_26_proposed_fsm_apply_observer.md).
 	// 100 ms keeps the fallback CPU at roughly 1/10th of the prior
 	// busy-poll, while bounding stale-poll latency to a value clients
 	// already tolerate from network round-trips.
-	redisStreamWaitFallback = 100 * time.Millisecond
-	redisKeywordCount       = "COUNT"
+	redisBlockWaitFallback = 100 * time.Millisecond
+	redisKeywordCount      = "COUNT"
 
 	// setWideColOverhead is the number of extra elements reserved in a set
 	// wide-column mutation slice beyond the per-member elements: one for the
@@ -3328,13 +3330,34 @@ func (r *RedisServer) zaddTxn(ctx context.Context, key []byte, flags zaddFlags,
 		})
 	}
 
-	_, dispatchErr := r.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
+	return added, r.dispatchAndSignalZSet(ctx, readTS, commitTS, elems, key)
+}
+
+// dispatchAndSignalZSet dispatches the elems through the coordinator
+// and, on success, wakes any BZPOPMIN waiter on the same node.
+// coordinator.Dispatch blocks until the FSM applies locally, so by
+// the time Signal fires the new members are visible at the readTS
+// the woken waiter will pick on its next iteration. Pulled out of
+// zaddTxn / zincrbyTxn so the parents stay under the cyclop budget
+// — the signal step would otherwise add an extra branch on the
+// dispatch error path.
+func (r *RedisServer) dispatchAndSignalZSet(
+	ctx context.Context,
+	readTS, commitTS uint64,
+	elems []*kv.Elem[kv.OP],
+	zsetKey []byte,
+) error {
+	_, err := r.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
 		IsTxn:    true,
 		StartTS:  normalizeStartTS(readTS),
 		CommitTS: commitTS,
 		Elems:    elems,
 	})
-	return added, cockerrors.WithStack(dispatchErr)
+	if err != nil {
+		return cockerrors.WithStack(err)
+	}
+	r.zsetWaiters.Signal(zsetKey)
+	return nil
 }
 
 // zincrbyTxn performs one attempt of ZINCRBY in wide-column format.
@@ -3386,13 +3409,10 @@ func (r *RedisServer) zincrbyTxn(ctx context.Context, key []byte, member string,
 			Value: deltaVal,
 		})
 	}
-	_, dispatchErr := r.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
-		IsTxn:    true,
-		StartTS:  normalizeStartTS(readTS),
-		CommitTS: commitTS,
-		Elems:    elems,
-	})
-	return newScore, cockerrors.WithStack(dispatchErr)
+	if err := r.dispatchAndSignalZSet(ctx, readTS, commitTS, elems, key); err != nil {
+		return 0, err
+	}
+	return newScore, nil
 }
 
 func (r *RedisServer) zincrby(conn redcon.Conn, cmd redcon.Command) {
@@ -3727,29 +3747,93 @@ func (r *RedisServer) bzpopmin(conn redcon.Conn, cmd redcon.Command) {
 	}
 	deadline := time.Now().Add(time.Duration(timeoutSeconds * float64(time.Second)))
 
-	for {
-		for _, key := range cmd.Args[1 : len(cmd.Args)-1] {
-			result, err := r.tryBZPopMin(key)
-			if err != nil {
-				conn.WriteError(err.Error())
-				return
-			}
-			if result == nil {
-				continue
-			}
+	keys := cmd.Args[1 : len(cmd.Args)-1]
+	r.bzpopminWaitLoop(conn, keys, deadline)
+}
 
-			conn.WriteArray(redisTripletWidth)
-			conn.WriteBulk(result.key)
-			conn.WriteBulkString(result.entry.Member)
-			conn.WriteBulkString(formatRedisFloat(result.entry.Score))
+// bzpopminWaitLoop runs the BLOCK-window wait loop. Extracted from
+// bzpopmin so the parent function stays under the cyclop budget.
+// Uses an event-driven signal from the in-process ZADD / ZINCRBY
+// path with a fallback timer for paths that bypass the signal.
+//
+// Registration happens BEFORE the first tryBZPopMin so a signal that
+// fires between the check and the wait cannot be lost: the buffered
+// channel holds it, and the next select wakes immediately.
+func (r *RedisServer) bzpopminWaitLoop(conn redcon.Conn, keys [][]byte, deadline time.Time) {
+	handlerCtx := r.handlerContext()
+	w, release := r.zsetWaiters.Register(keys)
+	defer release()
+	for {
+		if handlerCtx.Err() != nil {
+			conn.WriteNull()
+			return
+		}
+		if r.bzpopminTryAllKeys(conn, keys) {
 			return
 		}
-
 		if !time.Now().Before(deadline) {
 			conn.WriteNull()
 			return
 		}
-		time.Sleep(redisBusyPollBackoff)
+		waitForBlockedCommandUpdate(handlerCtx, w.C, deadline)
+	}
+}
+
+// bzpopminTryAllKeys runs one tryBZPopMin pass across keys. Returns
+// true when a result was written (success or terminal error) and the
+// caller should stop the loop, false to continue waiting.
+func (r *RedisServer) bzpopminTryAllKeys(conn redcon.Conn, keys [][]byte) bool {
+	for _, key := range keys {
+		result, err := r.tryBZPopMin(key)
+		if err != nil {
+			conn.WriteError(err.Error())
+			return true
+		}
+		if result == nil {
+			continue
+		}
+		conn.WriteArray(redisTripletWidth)
+		conn.WriteBulk(result.key)
+		conn.WriteBulkString(result.entry.Member)
+		conn.WriteBulkString(formatRedisFloat(result.entry.Score))
+		return true
+	}
+	return false
+}
+
+// waitForBlockedCommandUpdate blocks until one of: a write signal
+// arrives, the fallback poll tick fires, the parent handlerCtx is
+// cancelled, or the BLOCK deadline elapses — whichever happens first.
+// The fallback bounds latency for write paths that do not signal (Lua
+// flush, follower-applied entries); it cannot exceed the remaining
+// BLOCK window so the deadline branch in the caller's loop top always
+// gets a chance to fire when the BLOCK expires. Shared by every
+// blocking-command wait loop (XREAD BLOCK, BZPOPMIN today; BLPOP /
+// BRPOP / BLMOVE in follow-ups) — the keyWaiterRegistry that produces
+// waiterC is per-domain (streamWaiters vs zsetWaiters), but the
+// timer-and-select shape is identical.
+func waitForBlockedCommandUpdate(handlerCtx context.Context, waiterC <-chan struct{}, deadline time.Time) {
+	fallback := redisBlockWaitFallback
+	if remaining := time.Until(deadline); remaining < fallback {
+		fallback = remaining
+	}
+	timer := time.NewTimer(fallback)
+	defer func() {
+		if !timer.Stop() {
+			// The timer either fired (its case won and the channel
+			// was drained inline by select) or is still buffering
+			// the tick (waiter / handlerCtx won the race); drain
+			// the channel non-blocking so timer GC is clean.
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+	}()
+	select {
+	case <-waiterC:
+	case <-timer.C:
+	case <-handlerCtx.Done():
 	}
 }
 
@@ -4874,7 +4958,7 @@ func (r *RedisServer) xreadBusyPoll(conn redcon.Conn, req xreadRequest, deadline
 		// in handlerCtx, so it would cancel-on-call too — but routing
 		// through isXReadIterCtxError silently translates that into an
 		// empty iteration and the loop would otherwise wait at
-		// redisStreamWaitFallback cadence until the deadline.
+		// redisBlockWaitFallback cadence until the deadline.
 		if handlerCtx.Err() != nil {
 			conn.WriteNull()
 			return
@@ -4926,44 +5010,7 @@ func (r *RedisServer) xreadBusyPoll(conn redcon.Conn, req xreadRequest, deadline
 			conn.WriteNull()
 			return
 		}
-		waitForStreamUpdate(handlerCtx, w.C, deadline)
-	}
-}
-
-// waitForStreamUpdate blocks until one of: an XADD signal arrives,
-// the fallback poll tick fires, the parent handlerCtx is cancelled,
-// or the BLOCK deadline elapses — whichever happens first. The
-// fallback bounds latency for write paths that do not signal (Lua
-// flush, follower-applied entries); it cannot exceed the remaining
-// BLOCK window so the deadline branch in the caller's loop top
-// always gets a chance to fire when the BLOCK expires.
-//
-// Extracted to keep xreadBusyPoll under the cyclop budget — the
-// timer cleanup pattern (Stop + drain on miss) is awkward inline
-// and the helper lets the caller stay focused on the poll-result
-// branch.
-func waitForStreamUpdate(handlerCtx context.Context, waiterC <-chan struct{}, deadline time.Time) {
-	fallback := redisStreamWaitFallback
-	if remaining := time.Until(deadline); remaining < fallback {
-		fallback = remaining
-	}
-	timer := time.NewTimer(fallback)
-	defer func() {
-		if !timer.Stop() {
-			// The timer either fired (its case won and the channel
-			// was drained inline by select) or is still buffering
-			// the tick (waiter / handlerCtx won the race); drain
-			// the channel non-blocking so timer GC is clean.
-			select {
-			case <-timer.C:
-			default:
-			}
-		}
-	}()
-	select {
-	case <-waiterC:
-	case <-timer.C:
-	case <-handlerCtx.Done():
+		waitForBlockedCommandUpdate(handlerCtx, w.C, deadline)
 	}
 }
 
diff --git a/adapter/redis_compat_commands_stream_test.go b/adapter/redis_compat_commands_stream_test.go
index 5417e4f39..28fb98252 100644
--- a/adapter/redis_compat_commands_stream_test.go
+++ b/adapter/redis_compat_commands_stream_test.go
@@ -30,14 +30,14 @@ import (
 // path: an in-process XADD on the leader's redis adapter must wake an
 // XREAD BLOCK waiter on the same node so the reader returns the new
 // entry before its BLOCK deadline. The wake comes through
-// streamWaiterRegistry's signal channel — the prior 10 ms time.Sleep
+// keyWaiterRegistry's signal channel — the prior 10 ms time.Sleep
 // busy-poll loop would have exhibited the same end-to-end behaviour, so
 // this is an end-to-end sanity test rather than a wall-clock latency
 // gate (the latency gate is impractical under -race + parallel CI load,
 // where xreadOnce's Pebble seek alone can exceed any tight budget).
 //
 // Both client connections target the same node so they share the same
-// streamWaiterRegistry — the signal path is intentionally in-process
+// keyWaiterRegistry — the signal path is intentionally in-process
 // only (Lua and follower-side applies fall through to the fallback
 // timer; see xreadBusyPoll).
 func TestRedis_StreamXReadBlockWakesOnXAdd(t *testing.T) {
@@ -75,14 +75,14 @@ func TestRedis_StreamXReadBlockWakesOnXAdd(t *testing.T) {
 	}()
 
 	// Give the reader a moment to enter xreadBusyPoll and register a
-	// waiter on streamWaiterRegistry before XADD. If XADD landed first
+	// waiter on keyWaiterRegistry before XADD. If XADD landed first
 	// the entry would already be visible by the time the reader runs
 	// xreadOnce, so the registration race is benign — but waiting also
 	// gates out a different source of flake where the goroutine has not
 	// yet dialed redis.
 	//
 	// TODO: replace the time.Sleep with explicit synchronization (e.g.
-	// poll streamWaiterRegistry until the test's stream key shows up,
+	// poll keyWaiterRegistry until the test's stream key shows up,
 	// or expose a hook from RedisServer that fires when registration
 	// completes). Under -race on a slow CI runner the 50 ms pause may
 	// be insufficient — the test then exercises the
@@ -917,7 +917,7 @@ func TestRedis_StreamXReadShutdownShortCircuits(t *testing.T) {
 	// pre-fix this would happily run for the full 5 s after Close()
 	// because iterCtx was rooted in context.Background(). Post-fix the
 	// handlerCtx.Err() guard at the top of each loop iteration kicks
-	// in within ~one redisBusyPollBackoff (10 ms) and we reply null.
+	// in within ~one redisBlockWaitFallback (100 ms) and we reply null.
 	_, err := rdb.XAdd(ctx, &redis.XAddArgs{
 		Stream: "stream-shutdown",
 		ID:     "1-0",
diff --git a/adapter/redis_key_waiters.go b/adapter/redis_key_waiters.go
new file mode 100644
index 000000000..40aeaea4d
--- /dev/null
+++ b/adapter/redis_key_waiters.go
@@ -0,0 +1,149 @@
+package adapter
+
+import "sync"
+
+// keyWaiterRegistry is a multi-key channel-based wakeup primitive for
+// blocking Redis commands. It tracks goroutines blocked inside a wait
+// loop (BZPOPMIN today; BLPOP / BRPOP / BLMOVE in follow-ups) so a
+// later in-process write to one of the registered keys can wake them
+// up immediately instead of letting the poll-fallback timer run.
+//
+// Lookup is keyed by the user key (the same byte slice the client
+// passed to ZADD/BZPOPMIN). Waiters list every key the caller is
+// interested in. A signal on any registered key wakes the waiter, and
+// the caller is expected to re-check via the command's "try" helper
+// after waking — the channel only carries "something might have
+// changed".
+//
+// A buffered channel of size 1 is sufficient because the caller always
+// re-checks state after waking; coalescing multiple in-flight signals
+// into one wake-up is correct. Send is non-blocking (default branch
+// on send) so ZADD never stalls when a waiter is already pending.
+//
+// The mutex is held only briefly: collecting a snapshot of the waiter
+// set in Signal lets the slow path (channel sends) run lock-free, and
+// avoids holding the mutex across user code. Register's release fn is
+// idempotent (sync.Once) so deferred releases survive panic-style
+// early returns.
+//
+// The registry is intentionally generic — the same type instance can
+// be reused for any Redis blocking command. RedisServer holds one
+// registry per command domain (streamWaiters for XREAD BLOCK,
+// zsetWaiters for BZPOPMIN; BLPOP / BRPOP / BLMOVE in follow-ups)
+// so a signal on a zset key does not iterate over stream waiters and
+// vice versa.
+type keyWaiterRegistry struct {
+	mu      sync.Mutex
+	waiters map[string]map[*keyWaiter]struct{}
+}
+
+type keyWaiter struct {
+	C    chan struct{}
+	keys []string
+}
+
+func newKeyWaiterRegistry() *keyWaiterRegistry {
+	return &keyWaiterRegistry{
+		waiters: map[string]map[*keyWaiter]struct{}{},
+	}
+}
+
+// Register subscribes a waiter to every key in keys. Duplicate keys
+// are deduplicated so a multi-key BZPOPMIN where the client repeats
+// the same key does not signal the same waiter twice. Returns the
+// waiter handle and a release fn the caller MUST defer; the release
+// fn is safe to invoke multiple times. Nil-safe: returns a never-
+// fires waiter and a no-op release if the registry is nil (the
+// caller's select still gets a well-typed channel and the fallback
+// timer drives the loop). The nil path uses a buffered-1 channel
+// for consistency with the non-nil path so any code that hand-sends
+// into w.C on a nil-registry waiter (test stubs only, in practice)
+// coalesces the same way instead of deadlocking.
+func (reg *keyWaiterRegistry) Register(keys [][]byte) (*keyWaiter, func()) {
+	if reg == nil {
+		return &keyWaiter{C: make(chan struct{}, 1)}, func() {}
+	}
+	w := &keyWaiter{
+		C:    make(chan struct{}, 1),
+		keys: dedupKeyWaiterKeys(keys),
+	}
+	reg.mu.Lock()
+	for _, s := range w.keys {
+		set := reg.waiters[s]
+		if set == nil {
+			set = map[*keyWaiter]struct{}{}
+			reg.waiters[s] = set
+		}
+		set[w] = struct{}{}
+	}
+	reg.mu.Unlock()
+	var releaseOnce sync.Once
+	return w, func() {
+		releaseOnce.Do(func() { reg.unregister(w) })
+	}
+}
+
+func (reg *keyWaiterRegistry) unregister(w *keyWaiter) {
+	reg.mu.Lock()
+	for _, s := range w.keys {
+		set := reg.waiters[s]
+		if set == nil {
+			continue
+		}
+		delete(set, w)
+		if len(set) == 0 {
+			delete(reg.waiters, s)
+		}
+	}
+	reg.mu.Unlock()
+}
+
+// Signal wakes every waiter registered for key. Caller is the local
+// write command (ZADD / ZINCRBY) after coordinator.Dispatch returns;
+// by that point the FSM has applied the entry and a re-running
+// "try" helper on the leader is guaranteed to see it. Nil-safe:
+// test stubs that construct a RedisServer literal directly may
+// leave the registry unset, in which case Signal is a no-op.
+func (reg *keyWaiterRegistry) Signal(key []byte) {
+	if reg == nil {
+		return
+	}
+	reg.mu.Lock()
+	// staticcheck SA6001: indexing the map directly with the byte
+	// conversion lets the compiler skip the temporary string
+	// allocation on the lookup-only path.
+	set := reg.waiters[string(key)]
+	if len(set) == 0 {
+		reg.mu.Unlock()
+		return
+	}
+	waiters := make([]*keyWaiter, 0, len(set))
+	for w := range set {
+		waiters = append(waiters, w)
+	}
+	reg.mu.Unlock()
+	for _, w := range waiters {
+		select {
+		case w.C <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// dedupKeyWaiterKeys returns a new slice of stringified keys with
+// duplicates removed, preserving first-seen order. Empty input
+// returns an empty, non-nil slice so the caller can range over it
+// without a nil-guard.
+func dedupKeyWaiterKeys(keys [][]byte) []string {
+	out := make([]string, 0, len(keys))
+	seen := make(map[string]struct{}, len(keys))
+	for _, k := range keys {
+		s := string(k)
+		if _, ok := seen[s]; ok {
+			continue
+		}
+		seen[s] = struct{}{}
+		out = append(out, s)
+	}
+	return out
+}
diff --git a/adapter/redis_stream_waiters_test.go b/adapter/redis_key_waiters_test.go
similarity index 57%
rename from adapter/redis_stream_waiters_test.go
rename to adapter/redis_key_waiters_test.go
index 86f6eba3d..778173910 100644
--- a/adapter/redis_stream_waiters_test.go
+++ b/adapter/redis_key_waiters_test.go
@@ -7,13 +7,13 @@ import (
 	"time"
 )
 
-func TestStreamWaiterRegistry_SignalWakesRegisteredWaiter(t *testing.T) {
+func TestKeyWaiterRegistry_SignalWakesRegisteredWaiter(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
-	w, release := reg.Register([][]byte{[]byte("stream-a")})
+	reg := newKeyWaiterRegistry()
+	w, release := reg.Register([][]byte{[]byte("z-a")})
 	defer release()
 
-	reg.Signal([]byte("stream-a"))
+	reg.Signal([]byte("z-a"))
 
 	select {
 	case <-w.C:
@@ -22,13 +22,13 @@ func TestStreamWaiterRegistry_SignalWakesRegisteredWaiter(t *testing.T) {
 	}
 }
 
-func TestStreamWaiterRegistry_SignalUnrelatedKeyDoesNotWake(t *testing.T) {
+func TestKeyWaiterRegistry_SignalUnrelatedKeyDoesNotWake(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
-	w, release := reg.Register([][]byte{[]byte("stream-a")})
+	reg := newKeyWaiterRegistry()
+	w, release := reg.Register([][]byte{[]byte("z-a")})
 	defer release()
 
-	reg.Signal([]byte("stream-b"))
+	reg.Signal([]byte("z-b"))
 
 	select {
 	case <-w.C:
@@ -37,9 +37,9 @@ func TestStreamWaiterRegistry_SignalUnrelatedKeyDoesNotWake(t *testing.T) {
 	}
 }
 
-func TestStreamWaiterRegistry_MultiKeyWaiterWokenByAnyKey(t *testing.T) {
+func TestKeyWaiterRegistry_MultiKeyWaiterWokenByAnyKey(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
+	reg := newKeyWaiterRegistry()
 	w, release := reg.Register([][]byte{[]byte("a"), []byte("b"), []byte("c")})
 	defer release()
 
@@ -52,38 +52,25 @@ func TestStreamWaiterRegistry_MultiKeyWaiterWokenByAnyKey(t *testing.T) {
 	}
 }
 
-func TestStreamWaiterRegistry_DuplicateKeysDeduplicated(t *testing.T) {
+func TestKeyWaiterRegistry_DuplicateKeysDeduplicated(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
-	// Same key twice in the request; the dedup pass should record one
-	// registration so a single Signal does not double-send into the
-	// waiter's channel.
+	reg := newKeyWaiterRegistry()
 	w, release := reg.Register([][]byte{[]byte("dup"), []byte("dup")})
 	defer release()
 
 	reg.Signal([]byte("dup"))
 
-	// First select drains the (single) buffered signal.
 	select {
 	case <-w.C:
 	case <-time.After(time.Second):
 		t.Fatal("dedup waiter not signaled")
 	}
-	// Without dedup, the same Signal would have tried twice to send into
-	// a buffer of size 1 — the second send would still drop (default
-	// branch) so the channel would have only one item — but the buffer
-	// would be re-filled if the first drain had not happened yet. The
-	// stronger guarantee we test here is that registry.waiters only
-	// recorded one membership, so a *second* Signal (after drain) wakes
-	// exactly once, not twice.
 	reg.Signal([]byte("dup"))
 	select {
 	case <-w.C:
 	case <-time.After(time.Second):
 		t.Fatal("post-drain Signal failed to wake waiter")
 	}
-	// After draining both signals, no further signal is in flight; the
-	// channel must be empty.
 	select {
 	case <-w.C:
 		t.Fatal("waiter received a phantom third wake")
@@ -91,16 +78,12 @@ func TestStreamWaiterRegistry_DuplicateKeysDeduplicated(t *testing.T) {
 	}
 }
 
-func TestStreamWaiterRegistry_CoalescesPreDrainSignals(t *testing.T) {
+func TestKeyWaiterRegistry_CoalescesPreDrainSignals(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
+	reg := newKeyWaiterRegistry()
 	w, release := reg.Register([][]byte{[]byte("k")})
 	defer release()
 
-	// Three signals fire before the waiter has a chance to drain — the
-	// channel buffer is 1, so the latter two must drop on the
-	// non-blocking send. A correctly-coalescing waiter sees exactly one
-	// wake.
 	reg.Signal([]byte("k"))
 	reg.Signal([]byte("k"))
 	reg.Signal([]byte("k"))
@@ -117,12 +100,11 @@ func TestStreamWaiterRegistry_CoalescesPreDrainSignals(t *testing.T) {
 	}
 }
 
-func TestStreamWaiterRegistry_ReleaseStopsFurtherSignals(t *testing.T) {
+func TestKeyWaiterRegistry_ReleaseStopsFurtherSignals(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
+	reg := newKeyWaiterRegistry()
 	w, release := reg.Register([][]byte{[]byte("k")})
 	release()
-	// Drain the channel just in case it got buffered before release.
 	select {
 	case <-w.C:
 	default:
@@ -135,24 +117,24 @@ func TestStreamWaiterRegistry_ReleaseStopsFurtherSignals(t *testing.T) {
 	}
 }
 
-func TestStreamWaiterRegistry_ReleaseIsIdempotent(t *testing.T) {
+func TestKeyWaiterRegistry_ReleaseIsIdempotent(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
+	reg := newKeyWaiterRegistry()
 	_, release := reg.Register([][]byte{[]byte("k")})
 	release()
-	release() // second call must not panic / double-delete
+	release()
 	reg.Signal([]byte("k"))
 }
 
-func TestStreamWaiterRegistry_SignalWithNoWaiterIsNoOp(t *testing.T) {
+func TestKeyWaiterRegistry_SignalWithNoWaiterIsNoOp(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
+	reg := newKeyWaiterRegistry()
 	reg.Signal([]byte("nobody-here"))
 }
 
-func TestStreamWaiterRegistry_NilRegistry(t *testing.T) {
+func TestKeyWaiterRegistry_NilRegistryIsNoOp(t *testing.T) {
 	t.Parallel()
-	var reg *streamWaiterRegistry
+	var reg *keyWaiterRegistry
 	w, release := reg.Register([][]byte{[]byte("k")})
 	defer release()
 	if w == nil {
@@ -163,7 +145,7 @@ func TestStreamWaiterRegistry_NilRegistry(t *testing.T) {
 	}
 	// Buffered-1 invariant: a non-blocking direct send must succeed
 	// without deadlocking. (In production, Signal is a no-op for nil
-	// registries; this only matters for test stubs that hand-write
+	// registries; this only matters for test stubs that hand-send
 	// into w.C — but the contract is documented as buffered-1 either
 	// way, and an unbuffered channel would deadlock here.)
 	select {
@@ -171,21 +153,19 @@ func TestStreamWaiterRegistry_NilRegistry(t *testing.T) {
 	default:
 		t.Fatal("nil-registry waiter channel must be buffered (size 1)")
 	}
-	// And it must drain.
 	select {
 	case <-w.C:
 	default:
 		t.Fatal("nil-registry waiter channel did not deliver the manual send")
 	}
-	// Signal on nil registry is a no-op.
 	reg.Signal([]byte("k"))
 }
 
-func TestStreamWaiterRegistry_ManyWaitersFanOut(t *testing.T) {
+func TestKeyWaiterRegistry_ManyWaitersFanOut(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
+	reg := newKeyWaiterRegistry()
 	const n = 10
-	waiters := make([]*streamWaiter, 0, n)
+	waiters := make([]*keyWaiter, 0, n)
 	releases := make([]func(), 0, n)
 	for range n {
 		w, rel := reg.Register([][]byte{[]byte("fan")})
@@ -209,13 +189,9 @@ func TestStreamWaiterRegistry_ManyWaitersFanOut(t *testing.T) {
 	}
 }
 
-// TestStreamWaiterRegistry_ConcurrentRegisterSignal exercises the lock to
-// catch ordering bugs in Register/Signal/release: parallel registers and
-// signals on the same key must always either reach the waiter or have
-// happened before its registration window closed.
-func TestStreamWaiterRegistry_ConcurrentRegisterSignal(t *testing.T) {
+func TestKeyWaiterRegistry_ConcurrentRegisterSignal(t *testing.T) {
 	t.Parallel()
-	reg := newStreamWaiterRegistry()
+	reg := newKeyWaiterRegistry()
 	const goroutines = 64
 	const iterations = 100
 	var wokeOrTimedOut atomic.Int64
diff --git a/adapter/redis_stream_waiters.go b/adapter/redis_stream_waiters.go
deleted file mode 100644
index aaa9956ba..000000000
--- a/adapter/redis_stream_waiters.go
+++ /dev/null
@@ -1,149 +0,0 @@
-package adapter
-
-import "sync"
-
-// streamWaiterRegistry tracks goroutines blocked inside an XREAD BLOCK loop
-// so an XADD on the same node can wake them up immediately instead of
-// letting the poll-fallback timer run. Lookup is keyed by the user stream
-// key (the same byte slice the client passed to XADD/XREAD); waiters list
-// every key the caller is interested in. A signal on any registered key
-// wakes the waiter, and the caller is expected to re-check via xreadOnce
-// after waking — the channel only carries "something might have changed".
-//
-// A buffered channel of size 1 is sufficient because the caller always
-// re-checks state after waking; coalescing multiple in-flight signals into
-// one wake-up is correct. Send is non-blocking (default branch on send) so
-// XADD never stalls when a waiter is already pending.
-//
-// The mutex is held only briefly: collecting a snapshot of the waiter set
-// in Signal lets the slow path (channel sends) run lock-free, and avoids
-// holding the mutex across user code. Register's release fn is idempotent
-// (sync.Once) so deferred releases survive panic-style early returns.
-type streamWaiterRegistry struct {
-	mu      sync.Mutex
-	waiters map[string]map[*streamWaiter]struct{}
-}
-
-type streamWaiter struct {
-	C    chan struct{}
-	keys []string
-}
-
-func newStreamWaiterRegistry() *streamWaiterRegistry {
-	return &streamWaiterRegistry{
-		waiters: map[string]map[*streamWaiter]struct{}{},
-	}
-}
-
-// Register subscribes a waiter to every key in keys. Duplicate keys are
-// deduplicated so a multi-key XREAD where the client repeats the same
-// stream does not signal the same waiter twice. Returns the waiter handle
-// and a release fn the caller MUST defer; the release fn is safe to invoke
-// multiple times. Nil-safe: returns a never-fires waiter and a no-op
-// release if the registry is nil (the caller's select still gets a
-// well-typed channel and the fallback timer drives the loop). The nil
-// path uses a buffered-1 channel for consistency with the non-nil path
-// so any code that reaches Signal on a nil-registry waiter (test stubs
-// only, in practice) coalesces the same way.
-func (reg *streamWaiterRegistry) Register(keys [][]byte) (*streamWaiter, func()) {
-	if reg == nil {
-		return &streamWaiter{C: make(chan struct{}, 1)}, func() {}
-	}
-	w := &streamWaiter{
-		C:    make(chan struct{}, 1),
-		keys: dedupStreamKeys(keys),
-	}
-	reg.mu.Lock()
-	for _, s := range w.keys {
-		set := reg.waiters[s]
-		if set == nil {
-			set = map[*streamWaiter]struct{}{}
-			reg.waiters[s] = set
-		}
-		set[w] = struct{}{}
-	}
-	reg.mu.Unlock()
-	var releaseOnce sync.Once
-	return w, func() {
-		releaseOnce.Do(func() { reg.unregister(w) })
-	}
-}
-
-func (reg *streamWaiterRegistry) unregister(w *streamWaiter) {
-	reg.mu.Lock()
-	for _, s := range w.keys {
-		set := reg.waiters[s]
-		if set == nil {
-			continue
-		}
-		delete(set, w)
-		if len(set) == 0 {
-			delete(reg.waiters, s)
-		}
-	}
-	reg.mu.Unlock()
-}
-
-// Signal wakes every waiter registered for key. Caller is XADD on the
-// local node after dispatchElems returns; by that point the FSM has
-// applied the entry and a re-running xreadOnce on the leader is
-// guaranteed to see it. Nil-safe: test stubs that construct a
-// RedisServer literal directly may leave streamWaiters unset, in
-// which case Signal is a no-op.
-//
-// Cost shape: O(N) work per Signal where N is the waiter count for
-// the key — collects the waiter snapshot under the lock, then
-// non-blocking-sends outside the lock. With R XADDs/sec on a stream
-// shared by N BLOCK waiters, total wake-up cost is O(N*R)
-// xreadOnce calls/sec across the leader. For the production
-// operating point that motivated this change (small N — typically
-// one consumer per stream — moderate R), this is strictly better
-// than the old N*100 polls/sec. A pathological fan-out scenario
-// (many BLOCK waiters watching one hot stream, or one waiter with
-// an unsatisfiable far-future afterID being woken on every XADD)
-// can exceed the old busy-poll cost. The 100 ms fallback timer
-// bounds the worst-case CPU under such patterns. AfterID-aware
-// filtering at Signal time is a follow-up — see
-// docs/design/2026_04_26_proposed_fsm_apply_observer.md.
-func (reg *streamWaiterRegistry) Signal(key []byte) {
-	if reg == nil {
-		return
-	}
-	reg.mu.Lock()
-	// staticcheck SA6001: indexing the map directly with the byte
-	// conversion lets the compiler skip the temporary string allocation
-	// on the lookup-only path.
-	set := reg.waiters[string(key)]
-	if len(set) == 0 {
-		reg.mu.Unlock()
-		return
-	}
-	waiters := make([]*streamWaiter, 0, len(set))
-	for w := range set {
-		waiters = append(waiters, w)
-	}
-	reg.mu.Unlock()
-	for _, w := range waiters {
-		select {
-		case w.C <- struct{}{}:
-		default:
-		}
-	}
-}
-
-// dedupStreamKeys returns a new slice of stringified keys with duplicates
-// removed, preserving first-seen order. Empty input returns an empty,
-// non-nil slice so the caller can range over it without a nil-guard.
-func dedupStreamKeys(keys [][]byte) []string {
-	out := make([]string, 0, len(keys))
-	seen := make(map[string]struct{}, len(keys))
-	for _, k := range keys {
-		s := string(k)
-		if _, ok := seen[s]; ok {
-			continue
-		}
-		seen[s] = struct{}{}
-		out = append(out, s)
-	}
-	return out
-}