diff --git a/cmd/silo/main.go b/cmd/silo/main.go
index c0c63bf9..1caded2c 100644
--- a/cmd/silo/main.go
+++ b/cmd/silo/main.go
@@ -1708,6 +1708,7 @@ func main() {
 	// Build the reconciler early enough that playback handlers can trigger
 	// immediate session syncs after start/stop events.
 	nodeIdentity := resolveNodeIdentity()
+	instanceID := nodeIdentity + ":" + uuid.NewString()
 
 	var reconciler *worker.Reconciler
 	var heartbeatWriter *worker.HeartbeatWriter
@@ -1752,6 +1753,15 @@ func main() {
 		})
 		var monitorSource streammonitor.Source = localSource
 		if apiRedisClient != nil {
+			publisher := nodesessions.NewPublisher(apiRedisClient, instanceID, nodeIdentity)
+			publisher.Start(appCtx, 10*time.Second, func() []nodesessions.SessionInfo {
+				return streammonitor.LiveLocalSessions(sessionMgr, nodeIdentity)
+			})
+			defer func() {
+				cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 5*time.Second)
+				defer cleanupCancel()
+				publisher.Cleanup(cleanupCtx)
+			}()
 			monitorSource = streammonitor.NewMultiSource(localSource, streammonitor.NewRedisSource(apiRedisClient))
 		}
 		// Resolve caps through the session manager so the enforcer and synchronous
@@ -1768,13 +1778,19 @@ func main() {
 				"error", ttlErr)
 			overCapRevocationTTL = streamenforcer.DefaultRevocationTTL
 		}
-		streamenforcer.NewWithRevocationTTL(monitorSource, func(ctx context.Context, userID int) (int, error) {
+		enforcer := streamenforcer.NewWithRevocationTTL(monitorSource, func(ctx context.Context, userID int) (int, error) {
 			limits, err := sessionMgr.EffectiveLimits(ctx, userID)
 			if err != nil {
 				return 0, err
 			}
 			return limits.MaxStreams, nil
-		}, streamRevocation, 0, overCapRevocationTTL).Start(appCtx)
+		}, streamRevocation, 0, overCapRevocationTTL)
+		if apiRedisClient != nil {
+			enforcer.SetCoordinator(streamenforcer.NewRedisCoordinator(
+				apiRedisClient, instanceID, streamenforcer.DefaultInterval,
+			))
+		}
+		enforcer.Start(appCtx)
 
 		nodeURL := fmt.Sprintf("http://%s%s", nodeIdentity, cfg.Server.Listen)
 		heartbeatWriter = worker.NewHeartbeatWriter(deps.DB, nodeIdentity, mode, nodeURL)
diff --git a/internal/nodesessions/publisher.go b/internal/nodesessions/publisher.go
new file mode 100644
index 00000000..c280cfd6
--- /dev/null
+++ b/internal/nodesessions/publisher.go
@@ -0,0 +1,124 @@
+package nodesessions
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// Publisher mirrors the integrated API process's live sessions into the
+// shared node-session keyspace.
+type Publisher struct {
+	rdb      *redis.Client
+	nodeName string
+	nodeHash string
+	ttl      time.Duration
+
+	mu  sync.Mutex
+	ids map[string]struct{}
+}
+
+// NewPublisher creates a publisher whose key namespace is derived from the
+// process-unique instanceID. nodeName remains display metadata only.
+func NewPublisher(rdb *redis.Client, instanceID, nodeName string) *Publisher {
+	return &Publisher{
+		rdb:      rdb,
+		nodeName: nodeName,
+		nodeHash: namespaceHash(instanceID),
+		ttl:      sessionTTL,
+		ids:      make(map[string]struct{}),
+	}
+}
+
+// Publish refreshes every currently-live record and immediately removes only
+// records which disappeared since the preceding successful snapshot.
+func (p *Publisher) Publish(ctx context.Context, infos []SessionInfo) {
+	if p == nil || p.rdb == nil {
+		return
+	}
+
+	type record struct {
+		id   string
+		data []byte
+	}
+	records := make([]record, 0, len(infos))
+	current := make(map[string]struct{}, len(infos))
+	for _, info := range infos {
+		if info.SessionID == "" {
+			continue
+		}
+		info.NodeName = p.nodeName
+		data, err := json.Marshal(info)
+		if err != nil {
+			slog.DebugContext(ctx, "integrated session marshal failed",
+				"component", "nodesessions", "session", info.SessionID, "error", err)
+			continue
+		}
+		records = append(records, record{id: info.SessionID, data: data})
+		current[info.SessionID] = struct{}{}
+	}
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	pipe := p.rdb.Pipeline()
+	for _, rec := range records {
+		pipe.Set(ctx, sessionRedisKey(p.nodeHash, rec.id), rec.data, p.ttl)
+	}
+	for id := range p.ids {
+		if _, live := current[id]; !live {
+			pipe.Del(ctx, sessionRedisKey(p.nodeHash, id))
+		}
+	}
+	if _, err := pipe.Exec(ctx); err != nil {
+		slog.DebugContext(ctx, "integrated session publish failed",
+			"component", "nodesessions", "error", err)
+		return
+	}
+	p.ids = current
+}
+
+// Start publishes snapshots at interval until ctx is canceled.
+func (p *Publisher) Start(ctx context.Context, interval time.Duration, snapshot func() []SessionInfo) {
+	if p == nil || p.rdb == nil || interval <= 0 || snapshot == nil {
+		return
+	}
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				p.Publish(ctx, snapshot())
+			}
+		}
+	}()
+}
+
+// Cleanup removes all records owned by this process.
+func (p *Publisher) Cleanup(ctx context.Context) {
+	if p == nil || p.rdb == nil {
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if len(p.ids) == 0 {
+		return
+	}
+	pipe := p.rdb.Pipeline()
+	for id := range p.ids {
+		pipe.Del(ctx, sessionRedisKey(p.nodeHash, id))
+	}
+	if _, err := pipe.Exec(ctx); err != nil {
+		slog.DebugContext(ctx, "integrated session cleanup failed",
+			"component", "nodesessions", "error", err)
+		return
+	}
+	p.ids = make(map[string]struct{})
+}
diff --git a/internal/nodesessions/publisher_test.go b/internal/nodesessions/publisher_test.go
new file mode 100644
index 00000000..50fb8b30
--- /dev/null
+++ b/internal/nodesessions/publisher_test.go
@@ -0,0 +1,146 @@
+package nodesessions
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+type expiringRedisHook struct {
+	mu      sync.Mutex
+	expires map[string]time.Time
+}
+
+func (h *expiringRedisHook) DialHook(next redis.DialHook) redis.DialHook {
+	return next
+}
+
+func (h *expiringRedisHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
+	return func(ctx context.Context, cmd redis.Cmder) error {
+		return h.process(ctx, cmd)
+	}
+}
+
+func (h *expiringRedisHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
+	return func(ctx context.Context, cmds []redis.Cmder) error {
+		for _, cmd := range cmds {
+			if err := h.process(ctx, cmd); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
+func (h *expiringRedisHook) process(_ context.Context, cmd redis.Cmder) error {
+	args := cmd.Args()
+	if len(args) < 2 {
+		return nil
+	}
+	key, _ := args[1].(string)
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	switch strings.ToLower(cmd.Name()) {
+	case "set":
+		expiry := time.Time{}
+		for i := 3; i+1 < len(args); i++ {
+			if strings.EqualFold(strings.TrimSpace(toString(args[i])), "px") {
+				ms, _ := strconv.ParseInt(toString(args[i+1]), 10, 64)
+				expiry = time.Now().Add(time.Duration(ms) * time.Millisecond)
+				break
+			}
+		}
+		h.expires[key] = expiry
+	case "del":
+		delete(h.expires, key)
+	}
+	return nil
+}
+
+func toString(value any) string {
+	switch value := value.(type) {
+	case string:
+		return value
+	case []byte:
+		return string(value)
+	case int:
+		return strconv.Itoa(value)
+	case int64:
+		return strconv.FormatInt(value, 10)
+	case time.Duration:
+		return strconv.FormatInt(value.Milliseconds(), 10)
+	default:
+		return ""
+	}
+}
+
+func (h *expiringRedisHook) has(key string) bool {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	expiry, ok := h.expires[key]
+	if ok && !expiry.IsZero() && !time.Now().Before(expiry) {
+		delete(h.expires, key)
+		return false
+	}
+	return ok
+}
+
+func testPublisherRedis(t *testing.T) (*redis.Client, *expiringRedisHook) {
+	t.Helper()
+	hook := &expiringRedisHook{expires: make(map[string]time.Time)}
+	rdb := redis.NewClient(&redis.Options{Addr: "unused.invalid:6379"})
+	rdb.AddHook(hook)
+	t.Cleanup(func() { _ = rdb.Close() })
+	return rdb, hook
+}
+
+func TestPublishersWithSharedNodeNameDoNotClobberEachOther(t *testing.T) {
+	ctx := context.Background()
+	rdb, redisState := testPublisherRedis(t)
+	suffix := time.Now().UTC().Format("20060102150405.000000000")
+	first := NewPublisher(rdb, "first:"+suffix, "shared-name")
+	second := NewPublisher(rdb, "second:"+suffix, "shared-name")
+	t.Cleanup(func() {
+		first.Cleanup(ctx)
+		second.Cleanup(ctx)
+	})
+
+	info := SessionInfo{SessionID: "same-session"}
+	first.Publish(ctx, []SessionInfo{info})
+	second.Publish(ctx, []SessionInfo{info})
+	first.Publish(ctx, nil)
+
+	firstKey := sessionRedisKey(first.nodeHash, info.SessionID)
+	secondKey := sessionRedisKey(second.nodeHash, info.SessionID)
+	if redisState.has(firstKey) {
+		t.Fatal("departed first key is still present")
+	}
+	if !redisState.has(secondKey) {
+		t.Fatal("live second key is absent")
+	}
+}
+
+func TestPublisherRefreshesStableRecordPastTTL(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	rdb, redisState := testPublisherRedis(t)
+	suffix := time.Now().UTC().Format("20060102150405.000000000")
+	publisher := NewPublisher(rdb, "refresh:"+suffix, "node")
+	publisher.ttl = 80 * time.Millisecond
+	t.Cleanup(func() { publisher.Cleanup(context.Background()) })
+
+	publisher.Start(ctx, 20*time.Millisecond, func() []SessionInfo {
+		return []SessionInfo{{SessionID: "stable"}}
+	})
+	time.Sleep(220 * time.Millisecond)
+
+	key := sessionRedisKey(publisher.nodeHash, "stable")
+	if !redisState.has(key) {
+		t.Fatal("stable key is absent after > TTL")
+	}
+}
diff --git a/internal/nodesessions/tracker.go b/internal/nodesessions/tracker.go
index 398d31ce..c00a5e2a 100644
--- a/internal/nodesessions/tracker.go
+++ b/internal/nodesessions/tracker.go
@@ -108,13 +108,12 @@ type Tracker struct {
 // NewTracker creates a session tracker for the given node.
 // rdb may be nil, in which case all operations are no-ops.
 func NewTracker(rdb *redis.Client, nodeURL, nodeName, nodeType string) *Tracker {
-	h := sha256.Sum256([]byte(nodeURL))
 	return &Tracker{
 		rdb:       rdb,
 		nodeURL:   nodeURL,
 		nodeName:  nodeName,
 		nodeType:  nodeType,
-		nodeHash:  hex.EncodeToString(h[:4]), // 8 hex chars
+		nodeHash:  namespaceHash(nodeURL),
 		sessions:  make(map[string]sessionRefs),
 		ephemeral: make(map[string]time.Time),
 		touched:   make(map[string]time.Time),
@@ -125,7 +124,16 @@ func NewTracker(rdb *redis.Client, nodeURL, nodeName, nodeType string) *Tracker
 
 // redisKey returns the full Redis key for a session.
 func (tr *Tracker) redisKey(sessionID string) string {
-	return KeyPrefix + tr.nodeHash + ":" + sessionID
+	return sessionRedisKey(tr.nodeHash, sessionID)
+}
+
+func namespaceHash(namespace string) string {
+	h := sha256.Sum256([]byte(namespace))
+	return hex.EncodeToString(h[:4])
+}
+
+func sessionRedisKey(nodeHash, sessionID string) string {
+	return KeyPrefix + nodeHash + ":" + sessionID
 }
 
 // NodeHash returns the node's hash prefix used in Redis keys.
diff --git a/internal/streamenforcer/coordinator.go b/internal/streamenforcer/coordinator.go
new file mode 100644
index 00000000..f244e667
--- /dev/null
+++ b/internal/streamenforcer/coordinator.go
@@ -0,0 +1,58 @@
+package streamenforcer
+
+import (
+	"context"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const coordinatorKey = "silo:enforcer:leader"
+
+var acquireLeaseScript = redis.NewScript(`
+local v = redis.call('GET', KEYS[1])
+if v == false or v == ARGV[1] then
+	redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])
+	return 1
+end
+return 0
+`)
+
+// Coordinator elects one enforcer replica for an evaluation pass.
+type Coordinator interface {
+	Acquire(ctx context.Context) (bool, error)
+}
+
+// RedisCoordinator uses a renewable lease to select one evaluator.
+type RedisCoordinator struct {
+	rdb        *redis.Client
+	instanceID string
+	leaseTTL   time.Duration
+}
+
+// NewRedisCoordinator creates a coordinator whose lease lasts twice the
+// evaluation interval.
+func NewRedisCoordinator(rdb *redis.Client, instanceID string, interval time.Duration) *RedisCoordinator {
+	if interval <= 0 {
+		interval = DefaultInterval
+	}
+	return &RedisCoordinator{rdb: rdb, instanceID: instanceID, leaseTTL: 2 * interval}
+}
+
+// Acquire claims or renews this instance's lease.
+func (c *RedisCoordinator) Acquire(ctx context.Context) (bool, error) {
+	if c == nil || c.rdb == nil {
+		return true, nil
+	}
+	result, err := acquireLeaseScript.Run(
+		ctx,
+		c.rdb,
+		[]string{coordinatorKey},
+		c.instanceID,
+		c.leaseTTL.Milliseconds(),
+	).Int()
+	if err != nil {
+		return false, err
+	}
+	return result == 1, nil
+}
diff --git a/internal/streamenforcer/enforcer.go b/internal/streamenforcer/enforcer.go
index f9039f56..1b152e57 100644
--- a/internal/streamenforcer/enforcer.go
+++ b/internal/streamenforcer/enforcer.go
@@ -60,6 +60,15 @@ type Enforcer struct {
 	interval      time.Duration
 	revocationTTL time.Duration
 	now           func() time.Time
+	coordinator   Coordinator
+}
+
+// SetCoordinator installs the optional cross-replica evaluator coordinator.
+// Without one, every tick evaluates as before.
+func (e *Enforcer) SetCoordinator(c Coordinator) {
+	if e != nil {
+		e.coordinator = c
+	}
 }
 
 // New builds an enforcer. interval <= 0 uses DefaultInterval.
@@ -131,7 +140,18 @@ func (e *Enforcer) Start(ctx context.Context) {
 // evaluate runs one pass: snapshot → per-user over-cap check → revoke victims.
 // Exported behavior is covered by EvaluateOnce for tests.
 func (e *Enforcer) evaluate(ctx context.Context) {
-	if err := e.EvaluateOnce(ctx); err != nil {
+	if e.coordinator != nil {
+		acquired, err := e.coordinator.Acquire(ctx)
+		if err != nil {
+			// Coordination failure must not disable enforcement.
+			slog.DebugContext(ctx, "stream enforcer coordination failed; evaluating anyway", "error", err)
+		} else if !acquired {
+			return
+		}
+	}
+	passCtx, cancel := context.WithTimeout(ctx, e.interval)
+	defer cancel()
+	if err := e.EvaluateOnce(passCtx); err != nil {
 		slog.DebugContext(ctx, "stream enforcer evaluate failed", "error", err)
 	}
 }
diff --git a/internal/streamenforcer/enforcer_test.go b/internal/streamenforcer/enforcer_test.go
index e8d590a8..f9f77e37 100644
--- a/internal/streamenforcer/enforcer_test.go
+++ b/internal/streamenforcer/enforcer_test.go
@@ -3,13 +3,16 @@ package streamenforcer
 import (
 	"context"
 	"errors"
+	"os"
 	"sort"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/Silo-Server/silo-server/internal/nodesessions"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/streammonitor"
+	"github.com/redis/go-redis/v9"
 )
 
 type fakeSource struct {
@@ -246,3 +249,158 @@ func TestParseRevocationTTL(t *testing.T) {
 		}
 	}
 }
+
+type firstCoordinator struct {
+	mu       sync.Mutex
+	acquired bool
+}
+
+func (c *firstCoordinator) Acquire(context.Context) (bool, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.acquired {
+		return false, nil
+	}
+	c.acquired = true
+	return true, nil
+}
+
+type errorCoordinator struct{}
+
+func (errorCoordinator) Acquire(context.Context) (bool, error) {
+	return false, errors.New("redis unavailable")
+}
+
+type countingSource struct {
+	mu    sync.Mutex
+	snap  streammonitor.Snapshot
+	calls int
+}
+
+func (s *countingSource) Snapshot(context.Context) (streammonitor.Snapshot, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.calls++
+	return s.snap, nil
+}
+
+func TestSharedCoordinatorAllowsExactlyOneEnforcerEvaluation(t *testing.T) {
+	base := time.Now()
+	source := &countingSource{snap: streammonitor.Snapshot{Streams: []streammonitor.LiveStream{
+		stream("keep", 7, base.Add(time.Minute)),
+		stream("victim", 7, base),
+	}}}
+	revoker := &fakeRevoker{}
+	coordinator := &firstCoordinator{}
+	enforcers := []*Enforcer{
+		New(source, func(context.Context, int) (int, error) { return 1, nil }, revoker, time.Second),
+		New(source, func(context.Context, int) (int, error) { return 1, nil }, revoker, time.Second),
+	}
+	for _, enforcer := range enforcers {
+		enforcer.SetCoordinator(coordinator)
+	}
+
+	var wg sync.WaitGroup
+	for _, enforcer := range enforcers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			enforcer.evaluate(context.Background())
+		}()
+	}
+	wg.Wait()
+	source.mu.Lock()
+	snapshotCalls := source.calls
+	source.mu.Unlock()
+	if snapshotCalls != 1 {
+		t.Fatalf("snapshot calls = %d, want exactly 1", snapshotCalls)
+	}
+	if len(revoker.revoked) != 1 || revoker.revoked[0] != "victim" {
+		t.Fatalf("revoked = %v, want exactly [victim]", revoker.revoked)
+	}
+}
+
+func TestCoordinatorErrorStillEvaluates(t *testing.T) {
+	base := time.Now()
+	revoker := &fakeRevoker{}
+	enforcer := New(fakeSource{snap: streammonitor.Snapshot{Streams: []streammonitor.LiveStream{
+		stream("keep", 7, base.Add(time.Minute)),
+		stream("victim", 7, base),
+	}}}, func(context.Context, int) (int, error) { return 1, nil }, revoker, time.Second)
+	enforcer.SetCoordinator(errorCoordinator{})
+	enforcer.evaluate(context.Background())
+	if len(revoker.revoked) != 1 || revoker.revoked[0] != "victim" {
+		t.Fatalf("revoked = %v, want [victim]", revoker.revoked)
+	}
+}
+
+type blockingSource struct {
+	deadlineObserved chan struct{}
+}
+
+func (s blockingSource) Snapshot(ctx context.Context) (streammonitor.Snapshot, error) {
+	if _, ok := ctx.Deadline(); !ok {
+		close(s.deadlineObserved)
+		return streammonitor.Snapshot{}, errors.New("evaluation context has no deadline")
+	}
+	<-ctx.Done()
+	close(s.deadlineObserved)
+	return streammonitor.Snapshot{}, ctx.Err()
+}
+
+func TestEvaluationPassIsBoundedByInterval(t *testing.T) {
+	observed := make(chan struct{})
+	enforcer := New(
+		blockingSource{deadlineObserved: observed},
+		func(context.Context, int) (int, error) { return 1, nil },
+		&fakeRevoker{},
+		25*time.Millisecond,
+	)
+	start := time.Now()
+	enforcer.evaluate(context.Background())
+	elapsed := time.Since(start)
+	select {
+	case <-observed:
+	default:
+		t.Fatal("source did not observe the evaluation deadline")
+	}
+	if elapsed < 20*time.Millisecond || elapsed > 250*time.Millisecond {
+		t.Fatalf("evaluation elapsed %v, want bounded near 25ms", elapsed)
+	}
+}
+
+func TestRedisCoordinatorElectsOneHolderAndRenews(t *testing.T) {
+	addr := os.Getenv("SILO_TEST_REDIS_ADDR")
+	if addr == "" {
+		t.Skip("SILO_TEST_REDIS_ADDR is not set")
+	}
+	ctx := context.Background()
+	rdb := redis.NewClient(&redis.Options{Addr: addr})
+	t.Cleanup(func() {
+		_ = rdb.Del(ctx, coordinatorKey).Err()
+		_ = rdb.Close()
+	})
+	if err := rdb.Ping(ctx).Err(); err != nil {
+		t.Fatalf("ping test Redis: %v", err)
+	}
+	if err := rdb.Del(ctx, coordinatorKey).Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	const interval = 80 * time.Millisecond
+	first := NewRedisCoordinator(rdb, "first", interval)
+	second := NewRedisCoordinator(rdb, "second", interval)
+	if acquired, err := first.Acquire(ctx); err != nil || !acquired {
+		t.Fatalf("first acquire=%v err=%v, want true", acquired, err)
+	}
+	if acquired, err := second.Acquire(ctx); err != nil || acquired {
+		t.Fatalf("second acquire=%v err=%v, want false", acquired, err)
+	}
+	time.Sleep(90 * time.Millisecond)
+	if acquired, err := first.Acquire(ctx); err != nil || !acquired {
+		t.Fatalf("renew acquire=%v err=%v, want true", acquired, err)
+	}
+	if ttl, err := rdb.PTTL(ctx, coordinatorKey).Result(); err != nil || ttl < 120*time.Millisecond {
+		t.Fatalf("renewed lease TTL=%v err=%v, want >=120ms", ttl, err)
+	}
+}
diff --git a/internal/streamrevoke/store.go b/internal/streamrevoke/store.go
index b92a61c8..6ee0b2ed 100644
--- a/internal/streamrevoke/store.go
+++ b/internal/streamrevoke/store.go
@@ -51,6 +51,67 @@ const (
 	defaultTTL = 24 * time.Hour
 )
 
+var mergeMirrorScript = redis.NewScript(`
+local incoming = cjson.decode(ARGV[1])
+local existing_raw = redis.call('GET', KEYS[1])
+local merged = incoming
+
+if existing_raw ~= false then
+	local ok, existing = pcall(cjson.decode, existing_raw)
+	if ok
+		and type(existing.revoked_at_sec) == 'number'
+		and type(existing.revoked_at_nsec) == 'number'
+		and type(existing.expires_at_sec) == 'number'
+		and type(existing.expires_at_nsec) == 'number' then
+		merged = existing
+
+		local incoming_permanent = incoming.expires_at_sec == 0
+		local existing_permanent = existing.expires_at_sec == 0
+		local incoming_expiry_later = incoming.expires_at_sec > existing.expires_at_sec
+			or (incoming.expires_at_sec == existing.expires_at_sec
+				and incoming.expires_at_nsec > existing.expires_at_nsec)
+		if incoming_permanent or (not existing_permanent and incoming_expiry_later) then
+			merged.expires_at = incoming.expires_at
+			merged.expires_at_sec = incoming.expires_at_sec
+			merged.expires_at_nsec = incoming.expires_at_nsec
+		end
+
+		local incoming_cutoff_later = incoming.revoked_at_sec > existing.revoked_at_sec
+			or (incoming.revoked_at_sec == existing.revoked_at_sec
+				and incoming.revoked_at_nsec > existing.revoked_at_nsec)
+		if incoming_cutoff_later then
+			merged.revoked_at = incoming.revoked_at
+			merged.revoked_at_sec = incoming.revoked_at_sec
+			merged.revoked_at_nsec = incoming.revoked_at_nsec
+			merged.reason = incoming.reason
+		end
+	end
+end
+
+local permanent = merged.expires_at_sec == 0
+local expired = not permanent and (
+	merged.expires_at_sec < tonumber(ARGV[2])
+	or (merged.expires_at_sec == tonumber(ARGV[2])
+		and merged.expires_at_nsec <= tonumber(ARGV[3]))
+)
+if expired then
+	redis.call('DEL', KEYS[1])
+	return 0
+end
+
+local encoded = cjson.encode(merged)
+if permanent then
+	redis.call('SET', KEYS[1], encoded)
+else
+	local ttl_ms = math.ceil(
+		(merged.expires_at_sec - tonumber(ARGV[2])) * 1000
+		+ (merged.expires_at_nsec - tonumber(ARGV[3])) / 1000000
+	)
+	redis.call('SET', KEYS[1], encoded, 'PX', math.max(1, ttl_ms))
+end
+return 1
+`)
+
 // Kind identifies what a revocation targets.
 type Kind string
 
@@ -143,6 +204,7 @@ type Store struct {
 	items            map[Key]Revocation
 	tombstones       map[Key]time.Time
 	tombstoneExpires map[Key]time.Time
+	mirrorWarnOnce   sync.Once
 }
 
 // New builds a Store, applying defaults for any unset Options.
@@ -311,6 +373,19 @@ func (r Revocation) effectiveExpiry() time.Time {
 	return r.ExpiresAt
 }
 
+// mergeRevocations is the Go definition of the two independent monotonic
+// dimensions also applied atomically by mergeMirrorScript.
+func mergeRevocations(existing, incoming Revocation) Revocation {
+	if incoming.effectiveExpiry().After(existing.effectiveExpiry()) {
+		existing.ExpiresAt = incoming.ExpiresAt
+	}
+	if incoming.RevokedAt.After(existing.RevokedAt) {
+		existing.RevokedAt = incoming.RevokedAt
+		existing.Reason = incoming.Reason
+	}
+	return existing
+}
+
 // applyLocal independently merges a revocation's two monotonic dimensions:
 // expiry never moves earlier, while the user-cutoff RevokedAt never moves
 // backward. Reason follows the record that supplied the newer cutoff. Keeping
@@ -331,14 +406,7 @@ func (s *Store) applyLocal(r Revocation) {
 		}
 	}
 	if existing, ok := s.items[r.key()]; ok {
-		if r.effectiveExpiry().After(existing.effectiveExpiry()) {
-			existing.ExpiresAt = r.ExpiresAt
-		}
-		if r.RevokedAt.After(existing.RevokedAt) {
-			existing.RevokedAt = r.RevokedAt
-			existing.Reason = r.Reason
-		}
-		s.items[r.key()] = existing
+		s.items[r.key()] = mergeRevocations(existing, r)
 	} else {
 		s.items[r.key()] = r
 	}
@@ -405,10 +473,8 @@ func (s *Store) revokeWithWarningsLocked(ctx context.Context, key Key, reason st
 	merged := s.effective(r.key())
 
 	// Redis mirror for multi-node propagation to edges. Mirror the merged copy
-	// applyLocal kept. This SET is deliberately serialized by opMu in this
-	// process. Separate central replicas can still race and overwrite a newer
-	// value because Redis does not atomically merge records; that is deferred to
-	// the multi-replica work, not addressed by narrowing this lock.
+	// applyLocal kept; Redis atomically merges it with writes from other central
+	// replicas.
 	if err := s.mirrorToRedis(ctx, merged); err != nil {
 		warnings = append(warnings, "Redis mirror was not updated")
 	}
@@ -442,8 +508,29 @@ func (s *Store) revokeWithWarningsLocked(ctx context.Context, key Key, reason st
 	return warnings, nil
 }
 
-// mirrorToRedis writes (or deletes) the Redis mirror key for a revocation with a
-// TTL of the time remaining until r.ExpiresAt. It is the single shared
+type mirrorPayload struct {
+	Revocation
+	RevokedAtSec  int64 `json:"revoked_at_sec"`
+	RevokedAtNSec int64 `json:"revoked_at_nsec"`
+	ExpiresAtSec  int64 `json:"expires_at_sec"`
+	ExpiresAtNSec int64 `json:"expires_at_nsec"`
+}
+
+func newMirrorPayload(r Revocation) mirrorPayload {
+	payload := mirrorPayload{
+		Revocation:    r,
+		RevokedAtSec:  r.RevokedAt.Unix(),
+		RevokedAtNSec: int64(r.RevokedAt.Nanosecond()),
+	}
+	if !r.ExpiresAt.IsZero() {
+		payload.ExpiresAtSec = r.ExpiresAt.Unix()
+		payload.ExpiresAtNSec = int64(r.ExpiresAt.Nanosecond())
+	}
+	return payload
+}
+
+// mirrorToRedis atomically merges the Redis mirror with a revocation and sets
+// its TTL to the time remaining until the merged expiry. It is the single shared
 // Redis-arming path, used both by Revoke and by the durable warm loops: edge
 // nodes (proxy/transcode) have no durable store and learn kills ONLY from
 // Redis, so re-arming Redis from the durable source of truth after a Redis flush
@@ -454,15 +541,32 @@ func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) error {
 	if s.rdb == nil {
 		return nil
 	}
-	data, err := json.Marshal(r)
+	data, err := json.Marshal(newMirrorPayload(r))
 	if err != nil {
 		slog.WarnContext(ctx, "streamrevoke marshal failed", "error", err, "kind", r.Kind, "id", r.ID)
 		return err
 	}
+	now := time.Now()
+	if err := mergeMirrorScript.Run(
+		ctx,
+		s.rdb,
+		[]string{redisKey(r.key())},
+		data,
+		now.Unix(),
+		now.Nanosecond(),
+	).Err(); err == nil {
+		return nil
+	} else {
+		s.mirrorWarnOnce.Do(func() {
+			slog.WarnContext(ctx, "streamrevoke atomic Redis merge unavailable; falling back to plain mirror writes",
+				"error", err)
+		})
+	}
+	return s.mirrorToRedisFallback(ctx, r, data)
+}
+
+func (s *Store) mirrorToRedisFallback(ctx context.Context, r Revocation, data []byte) error {
 	if r.ExpiresAt.IsZero() {
-		// Permanent kill: set without a TTL, matching applyLocal/effectiveExpiry
-		// treating a zero ExpiresAt as "never". A TTL of time.Until(zero) would be
-		// hugely negative and drop the key, so late-joining edges would miss it.
 		if err := s.rdb.Set(ctx, redisKey(r.key()), data, 0).Err(); err != nil {
 			slog.WarnContext(ctx, "streamrevoke redis set failed", "error", err, "kind", r.Kind, "id", r.ID)
 			return err
@@ -471,7 +575,6 @@ func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) error {
 	}
 	ttl := time.Until(r.ExpiresAt)
 	if ttl <= 0 {
-		// Already expired: nothing to mirror in Redis.
 		if err := s.rdb.Del(ctx, redisKey(r.key())).Err(); err != nil {
 			slog.DebugContext(ctx, "streamrevoke redis del failed", "error", err, "kind", r.Kind, "id", r.ID)
 			return err
diff --git a/internal/streamrevoke/store_redis_test.go b/internal/streamrevoke/store_redis_test.go
new file mode 100644
index 00000000..3590783e
--- /dev/null
+++ b/internal/streamrevoke/store_redis_test.go
@@ -0,0 +1,208 @@
+package streamrevoke
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+func realRedis(t *testing.T) *redis.Client {
+	t.Helper()
+	addr := os.Getenv("SILO_TEST_REDIS_ADDR")
+	if addr == "" {
+		t.Skip("SILO_TEST_REDIS_ADDR is not set")
+	}
+	rdb := redis.NewClient(&redis.Options{Addr: addr})
+	t.Cleanup(func() { _ = rdb.Close() })
+	if err := rdb.Ping(context.Background()).Err(); err != nil {
+		t.Fatalf("ping test Redis: %v", err)
+	}
+	return rdb
+}
+
+func uniqueRedisRevocation(t *testing.T) (Key, string) {
+	t.Helper()
+	id := t.Name() + ":" + time.Now().UTC().Format("20060102150405.000000000")
+	key := Key{Kind: KindSession, ID: id}
+	return key, redisKey(key)
+}
+
+func readMirror(t *testing.T, rdb *redis.Client, key string) mirrorPayload {
+	t.Helper()
+	data, err := rdb.Get(context.Background(), key).Bytes()
+	if err != nil {
+		t.Fatalf("get mirror: %v", err)
+	}
+	var payload mirrorPayload
+	if err := json.Unmarshal(data, &payload); err != nil {
+		t.Fatalf("decode mirror %q: %v", data, err)
+	}
+	return payload
+}
+
+func TestRedisMirrorConvergesConcurrentIndependentDimensions(t *testing.T) {
+	ctx := context.Background()
+	rdb := realRedis(t)
+	key, redisMirrorKey := uniqueRedisRevocation(t)
+	t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() })
+	base := time.Unix(time.Now().Add(time.Hour).Unix(), 123456789).UTC()
+	longer := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "older-cutoff",
+		RevokedAt: base, ExpiresAt: base.Add(2 * time.Hour),
+	}
+	later := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "later-cutoff",
+		RevokedAt: base.Add(time.Nanosecond), ExpiresAt: base.Add(time.Hour),
+	}
+
+	stores := []*Store{New(Options{Redis: rdb}), New(Options{Redis: rdb})}
+	revocations := []Revocation{longer, later}
+	errs := make(chan error, len(stores))
+	var wg sync.WaitGroup
+	for i := range stores {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			errs <- stores[i].mirrorToRedis(ctx, revocations[i])
+		}(i)
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	got := readMirror(t, rdb, redisMirrorKey)
+	if !got.RevokedAt.Equal(later.RevokedAt) || got.Reason != later.Reason {
+		t.Fatalf("cutoff=%v reason=%q, want %v %q", got.RevokedAt, got.Reason, later.RevokedAt, later.Reason)
+	}
+	if !got.ExpiresAt.Equal(longer.ExpiresAt) {
+		t.Fatalf("expiry=%v, want %v", got.ExpiresAt, longer.ExpiresAt)
+	}
+}
+
+func TestRedisMirrorPermanentKillIsNeverDowngraded(t *testing.T) {
+	ctx := context.Background()
+	rdb := realRedis(t)
+	key, redisMirrorKey := uniqueRedisRevocation(t)
+	t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() })
+	base := time.Now().UTC()
+	store := New(Options{Redis: rdb})
+	if err := store.mirrorToRedis(ctx, Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "permanent",
+		RevokedAt: base,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	bounded := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "new-cutoff",
+		RevokedAt: base.Add(time.Nanosecond), ExpiresAt: base.Add(time.Hour),
+	}
+	if err := store.mirrorToRedis(ctx, bounded); err != nil {
+		t.Fatal(err)
+	}
+
+	got := readMirror(t, rdb, redisMirrorKey)
+	if !got.ExpiresAt.IsZero() {
+		t.Fatalf("expiry=%v, want permanent", got.ExpiresAt)
+	}
+	if !got.RevokedAt.Equal(bounded.RevokedAt) || got.Reason != bounded.Reason {
+		t.Fatalf("cutoff=%v reason=%q, want %v %q", got.RevokedAt, got.Reason, bounded.RevokedAt, bounded.Reason)
+	}
+}
+
+func TestRedisMirrorLapsedIncomingDoesNotDeleteStrongerKill(t *testing.T) {
+	ctx := context.Background()
+	rdb := realRedis(t)
+	key, redisMirrorKey := uniqueRedisRevocation(t)
+	t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() })
+	now := time.Now().UTC()
+	store := New(Options{Redis: rdb})
+	live := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "live",
+		RevokedAt: now.Add(-time.Minute), ExpiresAt: now.Add(time.Hour),
+	}
+	if err := store.mirrorToRedis(ctx, live); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.mirrorToRedis(ctx, Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "lapsed-new-cutoff",
+		RevokedAt: now, ExpiresAt: now.Add(-time.Second),
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	got := readMirror(t, rdb, redisMirrorKey)
+	if !got.ExpiresAt.Equal(live.ExpiresAt) || got.Reason != "lapsed-new-cutoff" {
+		t.Fatalf("merged mirror=%+v, want live expiry and newer cutoff reason", got)
+	}
+}
+
+func TestRedisMirrorOverwritesLegacyPayload(t *testing.T) {
+	ctx := context.Background()
+	rdb := realRedis(t)
+	key, redisMirrorKey := uniqueRedisRevocation(t)
+	t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() })
+	legacy := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "legacy",
+		RevokedAt: time.Now().Add(-time.Hour), ExpiresAt: time.Now().Add(time.Hour),
+	}
+	legacyData, err := json.Marshal(legacy)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := rdb.Set(ctx, redisMirrorKey, legacyData, time.Hour).Err(); err != nil {
+		t.Fatal(err)
+	}
+	incoming := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "replacement",
+		RevokedAt: time.Now().UTC(), ExpiresAt: time.Now().Add(2 * time.Hour).UTC(),
+	}
+	if err := New(Options{Redis: rdb}).mirrorToRedis(ctx, incoming); err != nil {
+		t.Fatal(err)
+	}
+
+	got := readMirror(t, rdb, redisMirrorKey)
+	if !got.RevokedAt.Equal(incoming.RevokedAt) || !got.ExpiresAt.Equal(incoming.ExpiresAt) ||
+		got.RevokedAtSec == 0 || got.ExpiresAtSec == 0 {
+		t.Fatalf("mirror=%+v, want numeric replacement payload for %+v", got, incoming)
+	}
+}
+
+func TestRedisMirrorEqualNanosecondCutoffKeepsExistingReason(t *testing.T) {
+	ctx := context.Background()
+	rdb := realRedis(t)
+	key, redisMirrorKey := uniqueRedisRevocation(t)
+	t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() })
+	cutoff := time.Unix(time.Now().Unix(), 123456789).UTC()
+	first := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "original-reason",
+		RevokedAt: cutoff, ExpiresAt: cutoff.Add(time.Hour),
+	}
+	second := Revocation{
+		Kind: key.Kind, ID: key.ID, Reason: "second",
+		RevokedAt: cutoff, ExpiresAt: cutoff.Add(2 * time.Hour),
+	}
+	store := New(Options{Redis: rdb})
+	if err := store.mirrorToRedis(ctx, first); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.mirrorToRedis(ctx, second); err != nil {
+		t.Fatal(err)
+	}
+
+	got := readMirror(t, rdb, redisMirrorKey)
+	if !got.RevokedAt.Equal(cutoff) || got.Reason != first.Reason {
+		t.Fatalf("cutoff=%v reason=%q, want stable %v %q", got.RevokedAt, got.Reason, cutoff, first.Reason)
+	}
+	if !got.ExpiresAt.Equal(second.ExpiresAt) {
+		t.Fatalf("expiry=%v, want independently merged %v", got.ExpiresAt, second.ExpiresAt)
+	}
+}
diff --git a/internal/streamrevoke/store_test.go b/internal/streamrevoke/store_test.go
index 46ae4d51..6a767569 100644
--- a/internal/streamrevoke/store_test.go
+++ b/internal/streamrevoke/store_test.go
@@ -181,6 +181,9 @@ func (h *recordingRedisHook) DialHook(next redis.DialHook) redis.DialHook {
 
 func (h *recordingRedisHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
 	return func(ctx context.Context, cmd redis.Cmder) error {
+		if cmd.Name() == "eval" || cmd.Name() == "evalsha" {
+			return errors.New("lua disabled")
+		}
 		if cmd.Name() == "set" {
 			select {
 			case h.set <- struct{}{}:
@@ -196,6 +199,33 @@ func (h *recordingRedisHook) ProcessPipelineHook(next redis.ProcessPipelineHook)
 	return next
 }
 
+func TestMirrorFallsBackToPlainSetWhenLuaFails(t *testing.T) {
+	rdb := redis.NewClient(&redis.Options{Addr: "unused.invalid:6379"})
+	t.Cleanup(func() { _ = rdb.Close() })
+	set := make(chan struct{}, 2)
+	rdb.AddHook(&recordingRedisHook{set: set})
+	store := New(Options{Redis: rdb})
+	var logs bytes.Buffer
+	previousLogger := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil)))
+	t.Cleanup(func() { slog.SetDefault(previousLogger) })
+	revocation := Revocation{
+		Kind: KindSession, ID: "fallback", RevokedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour),
+	}
+	for range 2 {
+		if err := store.mirrorToRedis(context.Background(), revocation); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if got := len(set); got != 2 {
+		t.Fatalf("plain SET fallback attempts = %d, want 2", got)
+	}
+	const warning = "streamrevoke atomic Redis merge unavailable; falling back to plain mirror writes"
+	if got := strings.Count(logs.String(), warning); got != 1 {
+		t.Fatalf("fallback warnings = %d, want 1; logs: %s", got, logs.String())
+	}
+}
+
 func (f *fakeDurable) UpsertTombstone(_ context.Context, tombstone Tombstone) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()