feat(api): operator-visible stream kill list with unrevoke

The kill switch had no operator surface. streamrevoke.Store.List() existed and
was called from nowhere, so an admin could only terminate one session by id or
revoke a user's streams as a side effect of editing their account — with no way
to see what was revoked, choose a TTL, or undo a mistake. Because expiry is
deliberately monotonic, a wrong 24h kill was irreversible.

- GET/POST/DELETE /api/v1/admin/streams/revocations, admin-only, additive.
- Explicit wire-to-internal kind mapping: the wire accepts "session" (and
  "sess"), the store key is "sess". Passing the wire string straight into a Key
  would create a revocation IsRevoked never consults.
- Validation: non-empty bounded session ids; canonical positive user ids
  (strconv.Itoa round-trip, so "01" is rejected — the cache key is the
  canonical form); ttl_seconds bounded to 30d so the duration cannot overflow;
  bounded reason and request body. DELETE of an absent key is idempotent.
- Store.Unrevoke, guarded by a bounded in-memory tombstone: the tombstone is
  installed and the local entry dropped BEFORE the slow durable/Redis deletes,
  so a concurrent poll reconcile cannot re-apply the row it just read. A newer
  Revoke on the same key clears the tombstone, so an unrevoke never suppresses
  a later legitimate kill. Tombstones age out with the kill they replaced.
- maintain() takes the same operation lock as Revoke/Unrevoke around its
  durable block, closing the window where a poll tick could re-Upsert a row
  Unrevoke had just deleted — invisible until a restart resurrected the kill.
- Durable self-heal now compares expiry, not mere presence, so a failed Revoke
  mirror leaves a stale shorter row that the next tick repairs.
- A failed unrevoke publish fails safe: other processes keep the kill until it
  expires. Propagation failures surface as warnings rather than weakening the
  local result.

Unrevoking an over-cap victim is legal but the async enforcer will re-revoke it
on its next pass while the user is still over cap; that is documented at the
endpoint.

Part of the stream monitoring & kill-switch epic.

internal/api/handlers/admin_stream_revocations.go

@@ -0,0 +1,178 @@
+package handlers
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/streamrevoke"
+	"github.com/go-chi/chi/v5"
+)
+
+const (
+	maxStreamRevocationBodyBytes = 16 << 10
+	maxStreamRevocationIDLength  = 256
+	maxStreamRevocationReason    = 512
+	maxStreamRevocationTTL       = 30 * 24 * 60 * 60
+)
+
+// AdminStreamRevocationHandler exposes the operator-visible stream kill list.
+// Unrevoking an over-cap victim is legal, but the async enforcer may revoke it
+// again on its next pass while the owning user remains over cap.
+type AdminStreamRevocationHandler struct {
+	store *streamrevoke.Store
+}
+
+func NewAdminStreamRevocationHandler(store *streamrevoke.Store) *AdminStreamRevocationHandler {
+	return &AdminStreamRevocationHandler{store: store}
+}
+
+type streamRevocationRequest struct {
+	Kind       string `json:"kind"`
+	ID         string `json:"id"`
+	Reason     string `json:"reason"`
+	TTLSeconds *int64 `json:"ttl_seconds"`
+}
+
+type streamRevocationResponse struct {
+	streamrevoke.Revocation
+	Warnings []string `json:"warnings,omitempty"`
+}
+
+func (h *AdminStreamRevocationHandler) HandleList(w http.ResponseWriter, _ *http.Request) {
+	revocations := h.store.List()
+	sort.Slice(revocations, func(i, j int) bool {
+		return revocations[i].RevokedAt.After(revocations[j].RevokedAt)
+	})
+	writeJSON(w, http.StatusOK, map[string]any{"revocations": revocations})
+}
+
+func (h *AdminStreamRevocationHandler) HandleCreate(w http.ResponseWriter, r *http.Request) {
+	var req streamRevocationRequest
+	r.Body = http.MaxBytesReader(w, r.Body, maxStreamRevocationBodyBytes)
+	decoder := json.NewDecoder(r.Body)
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
+		return
+	}
+	if err := ensureJSONEOF(decoder); err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
+		return
+	}
+
+	key, err := revocationKeyFromWire(req.Kind, req.ID)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", err.Error())
+		return
+	}
+	reason := strings.TrimSpace(req.Reason)
+	if len(reason) > maxStreamRevocationReason {
+		writeError(w, http.StatusBadRequest, "bad_request", "Reason is too long")
+		return
+	}
+
+	var until time.Time
+	if req.TTLSeconds != nil {
+		if *req.TTLSeconds <= 0 || *req.TTLSeconds > maxStreamRevocationTTL {
+			writeError(w, http.StatusBadRequest, "bad_request", "ttl_seconds must be between 1 and 2592000")
+			return
+		}
+		until = time.Now().Add(time.Duration(*req.TTLSeconds) * time.Second)
+	} else {
+		// A zero expiry asks the store to use its default via the public helper.
+		var warnings []string
+		switch key.Kind {
+		case streamrevoke.KindSession:
+			warnings, err = h.store.RevokeSessionWithWarnings(r.Context(), key.ID, reason)
+		case streamrevoke.KindUser:
+			userID, _ := strconv.Atoi(key.ID)
+			warnings, err = h.store.RevokeUserWithWarnings(r.Context(), userID, reason)
+		}
+		if err != nil {
+			writeError(w, http.StatusInternalServerError, "internal_error", "Failed to create revocation")
+			return
+		}
+		writeJSON(w, http.StatusCreated, streamRevocationResponse{Revocation: findRevocation(h.store, key), Warnings: warnings})
+		return
+	}
+
+	warnings, err := h.store.RevokeWithWarnings(r.Context(), key, reason, until)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to create revocation")
+		return
+	}
+	writeJSON(w, http.StatusCreated, streamRevocationResponse{Revocation: findRevocation(h.store, key), Warnings: warnings})
+}
+
+func (h *AdminStreamRevocationHandler) HandleDelete(w http.ResponseWriter, r *http.Request) {
+	key, err := revocationKeyFromWire(chi.URLParam(r, "kind"), chi.URLParam(r, "id"))
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", err.Error())
+		return
+	}
+	warnings, err := h.store.UnrevokeWithWarnings(r.Context(), key)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "internal_error", "Failed to remove revocation")
+		return
+	}
+	if len(warnings) > 0 {
+		writeJSON(w, http.StatusOK, map[string]any{"warnings": warnings})
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func revocationKeyFromWire(kind, id string) (streamrevoke.Key, error) {
+	var internalKind streamrevoke.Kind
+	switch kind {
+	case "session", string(streamrevoke.KindSession):
+		internalKind = streamrevoke.KindSession
+	case string(streamrevoke.KindUser):
+		internalKind = streamrevoke.KindUser
+	default:
+		return streamrevoke.Key{}, errors.New("kind must be session, sess, or user")
+	}
+
+	if internalKind == streamrevoke.KindSession {
+		id = strings.TrimSpace(id)
+		if id == "" {
+			return streamrevoke.Key{}, errors.New("session id is required")
+		}
+		if len(id) > maxStreamRevocationIDLength {
+			return streamrevoke.Key{}, errors.New("session id is too long")
+		}
+		return streamrevoke.Key{Kind: internalKind, ID: id}, nil
+	}
+
+	userID, err := strconv.Atoi(id)
+	if err != nil || userID <= 0 || strconv.Itoa(userID) != id {
+		return streamrevoke.Key{}, errors.New("user id must be a canonical positive integer")
+	}
+	return streamrevoke.Key{Kind: internalKind, ID: id}, nil
+}
+
+func ensureJSONEOF(decoder *json.Decoder) error {
+	var extra any
+	if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) {
+		if err == nil {
+			return errors.New("multiple JSON values")
+		}
+		return err
+	}
+	return nil
+}
+
+func findRevocation(store *streamrevoke.Store, key streamrevoke.Key) streamrevoke.Revocation {
+	for _, revocation := range store.List() {
+		if revocation.Kind == key.Kind && revocation.ID == key.ID {
+			return revocation
+		}
+	}
+	return streamrevoke.Revocation{Kind: key.Kind, ID: key.ID}
+}

internal/api/handlers/admin_stream_revocations_test.go

@@ -0,0 +1,124 @@
+package handlers
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
+	"github.com/Silo-Server/silo-server/internal/auth"
+	"github.com/Silo-Server/silo-server/internal/streamrevoke"
+	"github.com/go-chi/chi/v5"
+)
+
+func streamRevocationTestRouter(store *streamrevoke.Store) http.Handler {
+	h := NewAdminStreamRevocationHandler(store)
+	r := chi.NewRouter()
+	r.Get("/revocations", h.HandleList)
+	r.Post("/revocations", h.HandleCreate)
+	r.Delete("/revocations/{kind}/{id}", h.HandleDelete)
+	return r
+}
+
+func TestAdminStreamRevocationRequiresAdminMiddleware(t *testing.T) {
+	const adminRole = "admin"
+	protected := apimw.RequireActingAdmin(nil)(streamRevocationTestRouter(streamrevoke.New(streamrevoke.Options{})))
+	for _, tc := range []struct {
+		name   string
+		claims *auth.Claims
+		status int
+	}{
+		{name: "unauthenticated", status: http.StatusUnauthorized},
+		{name: "non admin", claims: &auth.Claims{UserID: 1, Role: string(scopeUser)}, status: http.StatusForbidden},
+		{name: adminRole, claims: &auth.Claims{UserID: 1, Role: adminRole}, status: http.StatusOK},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			req := httptest.NewRequest(http.MethodGet, "/revocations", nil)
+			if tc.claims != nil {
+				req = req.WithContext(apimw.SetClaims(req.Context(), tc.claims))
+			}
+			rec := httptest.NewRecorder()
+			protected.ServeHTTP(rec, req)
+			if rec.Code != tc.status {
+				t.Fatalf("status=%d, want %d", rec.Code, tc.status)
+			}
+		})
+	}
+}
+
+func TestAdminStreamRevocationSessionMappingAndListDeleteRoundTrip(t *testing.T) {
+	store := streamrevoke.New(streamrevoke.Options{})
+	router := streamRevocationTestRouter(store)
+	req := httptest.NewRequest(http.MethodPost, "/revocations", bytes.NewBufferString(`{"kind":"session","id":" s1 ","reason":" test "}`))
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, req)
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("create status=%d body=%s", rec.Code, rec.Body.String())
+	}
+	var created map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &created); err != nil {
+		t.Fatal(err)
+	}
+	if created["kind"] != "sess" || created["id"] != "s1" {
+		t.Fatalf("created=%v, want internal sess kind and trimmed id", created)
+	}
+
+	listReq := httptest.NewRequest(http.MethodGet, "/revocations", nil)
+	listRec := httptest.NewRecorder()
+	router.ServeHTTP(listRec, listReq)
+	var listed struct {
+		Revocations []streamrevoke.Revocation `json:"revocations"`
+	}
+	if err := json.Unmarshal(listRec.Body.Bytes(), &listed); err != nil {
+		t.Fatal(err)
+	}
+	if len(listed.Revocations) != 1 {
+		t.Fatalf("list=%+v", listed.Revocations)
+	}
+	deletePath := "/revocations/" + string(listed.Revocations[0].Kind) + "/" + listed.Revocations[0].ID
+	deleteReq := httptest.NewRequest(http.MethodDelete, deletePath, nil)
+	deleteRec := httptest.NewRecorder()
+	router.ServeHTTP(deleteRec, deleteReq)
+	if deleteRec.Code != http.StatusNoContent || len(store.List()) != 0 {
+		t.Fatalf("delete status=%d remaining=%v", deleteRec.Code, store.List())
+	}
+}
+
+func TestAdminStreamRevocationValidation(t *testing.T) {
+	router := streamRevocationTestRouter(streamrevoke.New(streamrevoke.Options{}))
+	tests := []struct {
+		name string
+		body string
+	}{
+		{name: "noncanonical user", body: `{"kind":"user","id":"01"}`},
+		{name: "zero user", body: `{"kind":"user","id":"0"}`},
+		{name: "unknown kind", body: `{"kind":"stream","id":"x"}`},
+		{name: "missing session id", body: `{"kind":"session","id":" "}`},
+		{name: "zero ttl", body: `{"kind":"sess","id":"x","ttl_seconds":0}`},
+		{name: "negative ttl", body: `{"kind":"sess","id":"x","ttl_seconds":-1}`},
+		{name: "ttl above max", body: `{"kind":"sess","id":"x","ttl_seconds":2592001}`},
+		{name: "ttl integer overflow", body: `{"kind":"sess","id":"x","ttl_seconds":9223372036854775808}`},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			req := httptest.NewRequest(http.MethodPost, "/revocations", bytes.NewBufferString(tt.body))
+			rec := httptest.NewRecorder()
+			router.ServeHTTP(rec, req)
+			if rec.Code != http.StatusBadRequest {
+				t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String())
+			}
+		})
+	}
+}
+
+func TestAdminStreamRevocationDeleteAbsentIsIdempotent(t *testing.T) {
+	router := streamRevocationTestRouter(streamrevoke.New(streamrevoke.Options{}))
+	req := httptest.NewRequest(http.MethodDelete, "/revocations/sess/missing", nil)
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, req)
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String())
+	}
+}

internal/api/router.go

@@ -3103,6 +3103,12 @@ func NewRouter(deps Dependencies) chi.Router {
 								r.Post("/sessions/{session_id}/terminate", adminPlaybackControlHandler.HandleTerminateSession)
 								r.Post("/sessions/{session_id}/message", adminPlaybackControlHandler.HandleMessageSession)
 							}
+							if deps.RevocationStore != nil {
+								streamRevocationHandler := handlers.NewAdminStreamRevocationHandler(deps.RevocationStore)
+								r.Get("/streams/revocations", streamRevocationHandler.HandleList)
+								r.Post("/streams/revocations", streamRevocationHandler.HandleCreate)
+								r.Delete("/streams/revocations/{kind}/{id}", streamRevocationHandler.HandleDelete)
+							}
 
 							if deps.TaskManager != nil {
 								taskHistoryRepo := repository.NewPgExecutionRepository(deps.DB)

internal/streamrevoke/durable_postgres.go

@@ -69,6 +69,14 @@ func (s *PostgresDurableStore) Upsert(ctx context.Context, r Revocation) error {
 	return nil
 }
 
+// Delete removes a revocation from the durable mirror.
+func (s *PostgresDurableStore) Delete(ctx context.Context, kind Kind, id string) error {
+	if _, err := s.pool.Exec(ctx, `DELETE FROM stream_revocations WHERE kind = $1 AND id = $2`, string(kind), id); err != nil {
+		return fmt.Errorf("streamrevoke delete: %w", err)
+	}
+	return nil
+}
+
 // ListActive returns every revocation not yet expired, for warming the hot-path
 // cache on startup and re-warming after a Redis flush.
 func (s *PostgresDurableStore) ListActive(ctx context.Context) ([]Revocation, error) {

internal/streamrevoke/store.go

@@ -36,6 +36,8 @@ const (
 	// EventStreamRevoked is the cache.Event.Type published on cache.ChannelAdmin
 	// when a revocation is created so other nodes apply it immediately.
 	EventStreamRevoked = "stream_revoked"
+	// EventStreamUnrevoked is published when an operator removes a revocation.
+	EventStreamUnrevoked = "stream_unrevoked"
 
 	defaultPollInterval = 60 * time.Second
 	// defaultTTL is intentionally >= the playback recipe-card lifetime
@@ -87,6 +89,7 @@ func (r Revocation) expired(now time.Time) bool {
 // the interface when one is provided.
 type DurableStore interface {
 	Upsert(ctx context.Context, r Revocation) error
+	Delete(ctx context.Context, kind Kind, id string) error
 	ListActive(ctx context.Context) ([]Revocation, error)
 	Prune(ctx context.Context) error
 }
@@ -108,8 +111,11 @@ type Store struct {
 	pollInterval time.Duration
 	defaultTTL   time.Duration
 
-	mu    sync.RWMutex
-	items map[Key]Revocation
+	opMu             sync.Mutex
+	mu               sync.RWMutex
+	items            map[Key]Revocation
+	tombstones       map[Key]time.Time
+	tombstoneExpires map[Key]time.Time
 }
 
 // New builds a Store, applying defaults for any unset Options.
@@ -121,12 +127,14 @@ func New(opts Options) *Store {
 		opts.DefaultTTL = defaultTTL
 	}
 	return &Store{
-		rdb:          opts.Redis,
-		bus:          opts.Bus,
-		durable:      opts.Durable,
-		pollInterval: opts.PollInterval,
-		defaultTTL:   opts.DefaultTTL,
-		items:        make(map[Key]Revocation),
+		rdb:              opts.Redis,
+		bus:              opts.Bus,
+		durable:          opts.Durable,
+		pollInterval:     opts.PollInterval,
+		defaultTTL:       opts.DefaultTTL,
+		items:            make(map[Key]Revocation),
+		tombstones:       make(map[Key]time.Time),
+		tombstoneExpires: make(map[Key]time.Time),
 	}
 }
 
@@ -207,14 +215,15 @@ func (s *Store) WatchAndCut(w http.ResponseWriter, sessionID string, userID int,
 //
 // startedAt is when the request's stream credential was issued: the stream
 // token's iat on token-bearing surfaces (edge proxy, transcode node, native
-// ?st=), or the request entry time on freshly-authenticated surfaces (native
-// session auth, jellycompat login). A user revocation is a CUTOFF, not a ban:
+// ?st=, ABS bearer), the persisted playback-session start for public ABS
+// tracks, or the feed creation time for public RSS capabilities. A user
+// revocation is a CUTOFF, not a ban:
 // it kills streams whose credential predates it, while a stream authorized
 // after it (which required passing auth that the revocation just reset) plays
 // normally. Without the cutoff, the OnUserSessionsRevoked hook — fired by any
 // admin edit of password/role/enabled/permissions/quality — would 403 the
-// user's playback for the full 24h TTL even after they re-authenticate, with
-// no unrevoke path (expiry is deliberately monotonic). A zero startedAt never
+// user's playback for the full 24h TTL even after they re-authenticate unless
+// an operator explicitly unrevoked it. A zero startedAt never
 // matches a user revocation (fail open, matching the enforcer's "never kill on
 // uncertainty"); session revocations are exact-id kills and ignore startedAt.
 func (s *Store) IsRevoked(sessionID string, userID int, startedAt time.Time) bool {
@@ -258,6 +267,14 @@ func (r Revocation) effectiveExpiry() time.Time {
 // not fight a freshly-applied one. The caller must not hold s.mu.
 func (s *Store) applyLocal(r Revocation) {
 	s.mu.Lock()
+	if issuedAt, ok := s.tombstones[r.key()]; ok {
+		if !r.RevokedAt.After(issuedAt) {
+			s.mu.Unlock()
+			return
+		}
+		delete(s.tombstones, r.key())
+		delete(s.tombstoneExpires, r.key())
+	}
 	if existing, ok := s.items[r.key()]; !ok || !existing.effectiveExpiry().After(r.effectiveExpiry()) {
 		s.items[r.key()] = r
 	}
@@ -273,11 +290,29 @@ func (s *Store) effective(k Key) Revocation {
 	return s.items[k]
 }
 
+func (s *Store) isTombstoned(k Key) bool {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	_, ok := s.tombstones[k]
+	return ok
+}
+
 // Revoke adds a revocation to the local cache, writes it to Redis (with a TTL
 // of until-now) when configured, mirrors it to the durable store when
 // configured, and publishes a pub/sub event so other nodes apply it
 // immediately.
 func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.Time) error {
+	_, err := s.RevokeWithWarnings(ctx, key, reason, until)
+	return err
+}
+
+// RevokeWithWarnings applies a revocation locally and reports propagation
+// failures without weakening the local kill. Revoke is the compatibility
+// wrapper used by existing callers that only need local-success semantics.
+func (s *Store) RevokeWithWarnings(ctx context.Context, key Key, reason string, until time.Time) ([]string, error) {
+	s.opMu.Lock()
+	defer s.opMu.Unlock()
+
 	now := time.Now()
 	r := Revocation{
 		Kind:      key.Kind,
@@ -290,6 +325,7 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T
 	// Local cache first: the hot path must reflect the kill immediately even if
 	// downstream propagation fails.
 	s.applyLocal(r)
+	var warnings []string
 
 	// Propagation must not die with the caller: an admin terminate rides its
 	// HTTP request's context, and an abort after applyLocal would strand the
@@ -301,6 +337,7 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T
 	if s.durable != nil {
 		if err := s.durable.Upsert(ctx, r); err != nil {
 			slog.Warn("streamrevoke durable upsert failed", "error", err, "kind", r.Kind, "id", r.ID)
+			warnings = append(warnings, "durable mirror was not updated")
 		}
 	}
 
@@ -308,7 +345,9 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T
 	// applyLocal kept (monotonic): a short over-cap re-revoke must not shorten a
 	// longer admin kill's Redis TTL, matching applyLocal and the durable upsert's
 	// GREATEST. Edges that warm from Redis in that window must see the later kill.
-	s.mirrorToRedis(ctx, s.effective(r.key()))
+	if err := s.mirrorToRedis(ctx, s.effective(r.key())); err != nil {
+		warnings = append(warnings, "Redis mirror was not updated")
+	}
 
 	// Pub/sub push so already-connected nodes apply the kill immediately (the
 	// warm loops deliberately skip this — SCAN reconcile handles late-joiners).
@@ -316,15 +355,16 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T
 		data, err := json.Marshal(r)
 		if err != nil {
 			slog.Warn("streamrevoke marshal failed", "error", err, "kind", r.Kind, "id", r.ID)
-			return err
+			return warnings, err
 		}
 		evt := cache.Event{Type: EventStreamRevoked, Payload: string(data)}
 		if err := s.bus.Publish(ctx, cache.ChannelAdmin, evt); err != nil {
 			slog.Warn("streamrevoke publish failed", "error", err, "kind", r.Kind, "id", r.ID)
+			warnings = append(warnings, "revocation event was not published")
 		}
 	}
 
-	return nil
+	return warnings, nil
 }
 
 // mirrorToRedis writes (or deletes) the Redis mirror key for a revocation with a
@@ -334,14 +374,14 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T
 // Redis, so re-arming Redis from the durable source of truth after a Redis flush
 // + central restart is what lets edges reconverge via their SCAN reconcile.
 // Best-effort: failures are logged, never returned. No-op when Redis is absent.
-func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) {
+func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) error {
 	if s.rdb == nil {
-		return
+		return nil
 	}
 	data, err := json.Marshal(r)
 	if err != nil {
 		slog.Warn("streamrevoke marshal failed", "error", err, "kind", r.Kind, "id", r.ID)
-		return
+		return err
 	}
 	if r.ExpiresAt.IsZero() {
 		// Permanent kill: set without a TTL, matching applyLocal/effectiveExpiry
@@ -349,20 +389,105 @@ func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) {
 		// 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.Warn("streamrevoke redis set failed", "error", err, "kind", r.Kind, "id", r.ID)
+			return err
 		}
-		return
+		return nil
 	}
 	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.Debug("streamrevoke redis del failed", "error", err, "kind", r.Kind, "id", r.ID)
+			return err
 		}
-		return
+		return nil
 	}
 	if err := s.rdb.Set(ctx, redisKey(r.key()), data, ttl).Err(); err != nil {
 		slog.Warn("streamrevoke redis set failed", "error", err, "kind", r.Kind, "id", r.ID)
+		return err
 	}
+	return nil
+}
+
+type unrevocation struct {
+	Kind      Kind      `json:"kind"`
+	ID        string    `json:"id"`
+	IssuedAt  time.Time `json:"issued_at"`
+	ExpiresAt time.Time `json:"expires_at"`
+}
+
+func (u unrevocation) key() Key {
+	return Key{Kind: u.Kind, ID: u.ID}
+}
+
+// applyUnrevocation installs a bounded tombstone before deleting the local
+// kill. A delayed unrevoke event never removes a newer revocation.
+func (s *Store) applyUnrevocation(u unrevocation) {
+	k := u.key()
+	s.mu.Lock()
+	if existing, ok := s.items[k]; ok && existing.RevokedAt.After(u.IssuedAt) {
+		s.mu.Unlock()
+		return
+	}
+	if previous, ok := s.tombstones[k]; ok && previous.After(u.IssuedAt) {
+		s.mu.Unlock()
+		return
+	}
+	expiresAt := u.ExpiresAt
+	maxExpiry := u.IssuedAt.Add(s.defaultTTL)
+	if expiresAt.IsZero() || expiresAt.After(maxExpiry) {
+		expiresAt = maxExpiry
+	}
+	s.tombstones[k] = u.IssuedAt
+	s.tombstoneExpires[k] = expiresAt
+	delete(s.items, k)
+	s.mu.Unlock()
+}
+
+// Unrevoke removes a kill locally and from every configured mirror. A failed
+// publish fails safe: other processes may retain the kill until expiry.
+func (s *Store) Unrevoke(ctx context.Context, key Key) error {
+	_, err := s.UnrevokeWithWarnings(ctx, key)
+	return err
+}
+
+// UnrevokeWithWarnings is the admin-facing form of Unrevoke.
+func (s *Store) UnrevokeWithWarnings(ctx context.Context, key Key) ([]string, error) {
+	s.opMu.Lock()
+	defer s.opMu.Unlock()
+
+	issuedAt := time.Now()
+	s.mu.RLock()
+	existing := s.items[key]
+	s.mu.RUnlock()
+	u := unrevocation{Kind: key.Kind, ID: key.ID, IssuedAt: issuedAt, ExpiresAt: existing.ExpiresAt}
+	s.applyUnrevocation(u)
+
+	ctx = context.WithoutCancel(ctx)
+	var warnings []string
+	if s.durable != nil {
+		if err := s.durable.Delete(ctx, key.Kind, key.ID); err != nil {
+			slog.WarnContext(ctx, "streamrevoke durable delete failed", "error", err, "kind", key.Kind, "id", key.ID)
+			warnings = append(warnings, "durable mirror was not deleted")
+		}
+	}
+	if s.rdb != nil {
+		if err := s.rdb.Del(ctx, redisKey(key)).Err(); err != nil {
+			slog.WarnContext(ctx, "streamrevoke redis delete failed", "error", err, "kind", key.Kind, "id", key.ID)
+			warnings = append(warnings, "Redis mirror was not deleted")
+		}
+	}
+	if s.bus != nil {
+		data, err := json.Marshal(u)
+		if err != nil {
+			return warnings, err
+		}
+		if err := s.bus.Publish(ctx, cache.ChannelAdmin, cache.Event{Type: EventStreamUnrevoked, Payload: string(data)}); err != nil {
+			slog.WarnContext(ctx, "streamrevoke unrevoke publish failed", "error", err, "kind", key.Kind, "id", key.ID)
+			warnings = append(warnings, "unrevocation event was not published; other processes may retain the kill until it expires")
+		}
+	}
+	return warnings, nil
 }
 
 // RevokeSession revokes a single session id for DefaultTTL.
@@ -370,6 +495,11 @@ func (s *Store) RevokeSession(ctx context.Context, sessionID, reason string) err
 	return s.Revoke(ctx, Key{Kind: KindSession, ID: sessionID}, reason, time.Now().Add(s.defaultTTL))
 }
 
+// RevokeSessionWithWarnings is the admin-facing form of RevokeSession.
+func (s *Store) RevokeSessionWithWarnings(ctx context.Context, sessionID, reason string) ([]string, error) {
+	return s.RevokeWithWarnings(ctx, Key{Kind: KindSession, ID: sessionID}, reason, time.Now().Add(s.defaultTTL))
+}
+
 // RevokeSessionFor revokes a single session id for a caller-chosen TTL. The
 // async enforcer uses a short TTL so a transient over-count (e.g. a ghost
 // session lingering next to a fresh reconnect) self-heals; a persistent abuser
@@ -398,6 +528,15 @@ func (s *Store) RevokeUser(ctx context.Context, userID int, reason string) error
 	return s.Revoke(ctx, userKey(userID), reason, time.Now().Add(s.defaultTTL))
 }
 
+// RevokeUserWithWarnings applies the default TTL while preserving propagation
+// warnings for the admin API.
+func (s *Store) RevokeUserWithWarnings(ctx context.Context, userID int, reason string) ([]string, error) {
+	if userID <= 0 {
+		return nil, fmt.Errorf("streamrevoke: invalid userID %d", userID)
+	}
+	return s.RevokeWithWarnings(ctx, userKey(userID), reason, time.Now().Add(s.defaultTTL))
+}
+
 // List returns the currently-active revocations, pruning expired entries on
 // read.
 func (s *Store) List() []Revocation {
@@ -424,6 +563,12 @@ func (s *Store) pruneExpired() {
 			delete(s.items, k)
 		}
 	}
+	for k, expiresAt := range s.tombstoneExpires {
+		if !now.Before(expiresAt) {
+			delete(s.tombstones, k)
+			delete(s.tombstoneExpires, k)
+		}
+	}
 	s.mu.Unlock()
 }
 
@@ -448,7 +593,9 @@ func (s *Store) StartSync(ctx context.Context) {
 			for _, r := range revs {
 				if !r.expired(now) {
 					s.applyLocal(r)
-					s.mirrorToRedis(ctx, r)
+					if !s.isTombstoned(r.key()) {
+						_ = s.mirrorToRedis(ctx, s.effective(r.key()))
+					}
 				}
 			}
 		}
@@ -516,20 +663,43 @@ func (s *Store) maintain(ctx context.Context) {
 	// flush self-heals, re-arming Redis for edge nodes too, then physically
 	// reclaim expired rows.
 	if s.durable != nil {
+		s.opMu.Lock()
+
 		if revs, err := s.durable.ListActive(ctx); err != nil {
 			slog.Warn("streamrevoke durable reconcile failed", "error", err)
 		} else {
 			now := time.Now()
+			durable := make(map[Key]Revocation, len(revs))
 			for _, r := range revs {
 				if !r.expired(now) {
+					durable[r.key()] = r
 					s.applyLocal(r)
-					s.mirrorToRedis(ctx, r)
+					if !s.isTombstoned(r.key()) {
+						_ = s.mirrorToRedis(ctx, s.effective(r.key()))
+					}
+				}
+			}
+			s.mu.RLock()
+			local := make([]Revocation, 0, len(s.items))
+			for k, r := range s.items {
+				if _, tombstoned := s.tombstones[k]; !tombstoned && !r.expired(now) {
+					local = append(local, r)
+				}
+			}
+			s.mu.RUnlock()
+			for _, r := range local {
+				durableCopy, ok := durable[r.key()]
+				if !ok || durableCopy.effectiveExpiry().Before(r.effectiveExpiry()) {
+					if err := s.durable.Upsert(ctx, r); err != nil {
+						slog.WarnContext(ctx, "streamrevoke durable self-heal failed", "error", err, "kind", r.Kind, "id", r.ID)
+					}
 				}
 			}
 		}
 		if err := s.durable.Prune(ctx); err != nil {
 			slog.Warn("streamrevoke durable prune failed", "error", err)
 		}
+		s.opMu.Unlock()
 	}
 	s.pruneExpired()
 }
@@ -537,18 +707,24 @@ func (s *Store) maintain(ctx context.Context) {
 // handleEvent applies an EventStreamRevoked event to the local cache. Other
 // event types on the channel are ignored.
 func (s *Store) handleEvent(evt cache.Event) {
-	if evt.Type != EventStreamRevoked {
-		return
+	switch evt.Type {
+	case EventStreamRevoked:
+		var r Revocation
+		if err := json.Unmarshal([]byte(evt.Payload), &r); err != nil {
+			slog.Debug("streamrevoke event unmarshal failed", "error", err)
+			return
+		}
+		if !r.expired(time.Now()) {
+			s.applyLocal(r)
+		}
+	case EventStreamUnrevoked:
+		var u unrevocation
+		if err := json.Unmarshal([]byte(evt.Payload), &u); err != nil {
+			slog.Debug("streamrevoke unrevoke event unmarshal failed", "error", err)
+			return
+		}
+		s.applyUnrevocation(u)
 	}
-	var r Revocation
-	if err := json.Unmarshal([]byte(evt.Payload), &r); err != nil {
-		slog.Debug("streamrevoke event unmarshal failed", "error", err)
-		return
-	}
-	if r.expired(time.Now()) {
-		return
-	}
-	s.applyLocal(r)
 }
 
 // reconcileFromRedis SCANs the revocation namespace and applies every live

internal/streamrevoke/store_test.go

@@ -2,10 +2,14 @@ package streamrevoke
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/Silo-Server/silo-server/internal/cache"
+	"github.com/redis/go-redis/v9"
 )
 
 // errFakeDurable is the error fakeDurable.ListActive returns while failNext > 0.
@@ -23,12 +27,31 @@ func newMemStore() *Store {
 // mutex-guarded. failNext, when > 0, makes ListActive return an error that many
 // times before succeeding, to exercise the bounded boot-warm retry.
 type fakeDurable struct {
-	mu       sync.Mutex
-	rows     map[Key]Revocation
-	upserts  int
-	prunes   int
-	failNext int
-	lists    int
+	mu            sync.Mutex
+	rows          map[Key]Revocation
+	upserts       int
+	prunes        int
+	deletes       int
+	failNext      int
+	lists         int
+	upsertStarted chan struct{}
+	unblockUpsert chan struct{}
+}
+
+type failingBus struct{}
+
+func (failingBus) Publish(context.Context, string, cache.Event) error {
+	return errors.New("publish failed")
+}
+func (failingBus) Subscribe(context.Context, string, cache.EventHandler) error { return nil }
+func (failingBus) Close() error                                                { return nil }
+
+func (f *fakeDurable) Delete(_ context.Context, kind Kind, id string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	delete(f.rows, Key{Kind: kind, ID: id})
+	f.deletes++
+	return nil
 }
 
 func newFakeDurable() *fakeDurable {
@@ -36,6 +59,10 @@ func newFakeDurable() *fakeDurable {
 }
 
 func (f *fakeDurable) Upsert(_ context.Context, r Revocation) error {
+	if f.upsertStarted != nil {
+		close(f.upsertStarted)
+		<-f.unblockUpsert
+	}
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.rows[r.key()] = r
@@ -221,6 +248,167 @@ func TestMaintainPrunesAndReWarmsFromDurable(t *testing.T) {
 	}
 }
 
+func TestMaintainDurableSelfHealDoesNotRaceUnrevoke(t *testing.T) {
+	ctx := context.Background()
+	fake := newFakeDurable()
+	fake.upsertStarted = make(chan struct{})
+	fake.unblockUpsert = make(chan struct{})
+	s := New(Options{Durable: fake})
+	key := Key{Kind: KindSession, ID: "sess-maintain-unrevoke"}
+	s.applyLocal(Revocation{
+		Kind:      key.Kind,
+		ID:        key.ID,
+		RevokedAt: time.Now(),
+		ExpiresAt: time.Now().Add(time.Hour),
+	})
+
+	maintainDone := make(chan struct{})
+	go func() {
+		s.maintain(ctx)
+		close(maintainDone)
+	}()
+	<-fake.upsertStarted
+
+	unrevokeDone := make(chan error, 1)
+	go func() {
+		unrevokeDone <- s.Unrevoke(ctx, key)
+	}()
+
+	select {
+	case err := <-unrevokeDone:
+		t.Fatalf("Unrevoke completed before maintain released the operation lock: %v", err)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	close(fake.unblockUpsert)
+	<-maintainDone
+	if err := <-unrevokeDone; err != nil {
+		t.Fatalf("Unrevoke: %v", err)
+	}
+
+	fake.mu.Lock()
+	_, exists := fake.rows[key]
+	fake.mu.Unlock()
+	if exists {
+		t.Fatal("durable self-heal resurrected the unrevoked row")
+	}
+}
+
+func TestUnrevokeTombstoneBlocksReconcileAndNewerRevokeClearsIt(t *testing.T) {
+	ctx := context.Background()
+	s := newMemStore()
+	key := Key{Kind: KindSession, ID: "sess-unrevoke"}
+	old := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Hour)}
+	s.applyLocal(old)
+
+	if err := s.Unrevoke(ctx, key); err != nil {
+		t.Fatalf("Unrevoke: %v", err)
+	}
+	s.applyLocal(old)
+	if s.IsRevoked(key.ID, 0, time.Time{}) {
+		t.Fatal("stale reconcile resurrected an unrevoked entry")
+	}
+
+	if err := s.Revoke(ctx, key, "new kill", time.Now().Add(time.Hour)); err != nil {
+		t.Fatalf("newer Revoke: %v", err)
+	}
+	if !s.IsRevoked(key.ID, 0, time.Time{}) {
+		t.Fatal("newer revoke was suppressed by tombstone")
+	}
+	s.mu.RLock()
+	_, tombstoned := s.tombstones[key]
+	s.mu.RUnlock()
+	if tombstoned {
+		t.Fatal("newer revoke did not clear tombstone")
+	}
+}
+
+func TestConcurrentRevokeUnrevokeThenReconcileDoesNotResurrect(t *testing.T) {
+	ctx := context.Background()
+	s := newMemStore()
+	key := Key{Kind: KindSession, ID: "sess-race"}
+	var wg sync.WaitGroup
+	for range 25 {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			_ = s.Revoke(ctx, key, "race", time.Now().Add(time.Hour))
+		}()
+		go func() {
+			defer wg.Done()
+			_ = s.Unrevoke(ctx, key)
+		}()
+	}
+	wg.Wait()
+
+	stale := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Hour)}
+	if err := s.Unrevoke(ctx, key); err != nil {
+		t.Fatalf("final Unrevoke: %v", err)
+	}
+	s.applyLocal(stale)
+	if s.IsRevoked(key.ID, 0, time.Time{}) {
+		t.Fatal("stale reconcile resurrected entry after concurrent operations")
+	}
+}
+
+func TestDelayedUnrevokeEventDoesNotDeleteNewerRevoke(t *testing.T) {
+	s := newMemStore()
+	key := Key{Kind: KindSession, ID: "sess-order"}
+	issuedAt := time.Now().Add(-time.Minute)
+	s.applyLocal(Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour)})
+	payload, err := json.Marshal(unrevocation{Kind: key.Kind, ID: key.ID, IssuedAt: issuedAt, ExpiresAt: time.Now().Add(time.Hour)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.handleEvent(cache.Event{Type: EventStreamUnrevoked, Payload: string(payload)})
+	if !s.IsRevoked(key.ID, 0, time.Time{}) {
+		t.Fatal("delayed unrevoke event deleted a newer revocation")
+	}
+}
+
+func TestUnrevokePropagationFailuresLeaveLocalStateUnrevoked(t *testing.T) {
+	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1", DialTimeout: 10 * time.Millisecond})
+	t.Cleanup(func() { _ = rdb.Close() })
+	s := New(Options{Redis: rdb, Bus: failingBus{}})
+	key := Key{Kind: KindSession, ID: "sess-local"}
+	s.applyLocal(Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Hour)})
+
+	warnings, err := s.UnrevokeWithWarnings(context.Background(), key)
+	if err != nil {
+		t.Fatalf("UnrevokeWithWarnings: %v", err)
+	}
+	if len(warnings) != 2 {
+		t.Fatalf("warnings = %v, want Redis and publish failures", warnings)
+	}
+	if s.IsRevoked(key.ID, 0, time.Time{}) {
+		t.Fatal("local revocation remained after propagation failures")
+	}
+}
+
+func TestMaintainRepairsShorterDurableExpiry(t *testing.T) {
+	ctx := context.Background()
+	fake := newFakeDurable()
+	key := Key{Kind: KindSession, ID: "sess-short-durable"}
+	short := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Minute)}
+	long := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour)}
+	fake.rows[key] = short
+	s := New(Options{Durable: fake})
+	s.applyLocal(long)
+
+	s.maintain(ctx)
+
+	fake.mu.Lock()
+	got := fake.rows[key]
+	upserts := fake.upserts
+	fake.mu.Unlock()
+	if got.ExpiresAt.Before(long.ExpiresAt) {
+		t.Fatalf("durable expiry = %v, want at least %v", got.ExpiresAt, long.ExpiresAt)
+	}
+	if upserts == 0 {
+		t.Fatal("shorter durable row was not re-upserted")
+	}
+}
+
 func TestIsRevoked(t *testing.T) {
 	ctx := context.Background()