diff --git a/docs/content/docs/guides/forge-extension.mdx b/docs/content/docs/guides/forge-extension.mdx
index 9771805..916864d 100644
--- a/docs/content/docs/guides/forge-extension.mdx
+++ b/docs/content/docs/guides/forge-extension.mdx
@@ -47,6 +47,7 @@ Forge will call `ext.Start(ctx)` on application start and `ext.Stop(ctx)` on shu
 | `WithDisableMigrate()` | Skip auto-migration at startup | `false` |
 | `WithGroveDatabase(name)` | Resolve a grove.DB from DI by name | `""` |
 | `WithGroveKV(name)` | Resolve a grove KV store from DI by name | `""` |
+| `WithKVKeyPrefix(prefix)` | Namespace the Redis store's keys per tenant (grove KV path only) | `""` |
 
 ## Auto-migration
 
@@ -114,6 +115,19 @@ ext := extension.New(
 )
 ```
 
+When several dispatch instances share one Redis, give each its own key
+prefix. Every key, stream and pub/sub channel the store touches goes under
+it (`ws_acme:dispatch:job:<id>`), so instances never see each other's
+queues, cron locks or leadership. Leave it empty for a single instance and
+the keys stay `dispatch:*` as before:
+
+```go
+ext := extension.New(
+    extension.WithGroveKV("dispatch-kv"),
+    extension.WithKVKeyPrefix("ws_acme:"),
+)
+```
+
 ### Store resolution order
 
 The extension resolves its store in this order:
@@ -134,6 +148,7 @@ extensions:
     base_path: /api/dispatch
     grove_database: jobs
     grove_kv: dispatch-kv
+    key_prefix: "ws_acme:"
     concurrency: 20
     queues:
       - default
@@ -161,6 +176,7 @@ dispatch:
 | `BasePath` | `base_path` | `string` | `"/api/dispatch"` | URL prefix for all dispatch HTTP routes |
 | `GroveDatabase` | `grove_database` | `string` | `""` | Name of the grove.DB to resolve from DI; empty uses the default DB |
 | `GroveKV` | `grove_kv` | `string` | `""` | Name of the grove KV store to resolve from DI; empty uses the default KV |
+| `KeyPrefix` | `key_prefix` | `string` | `""` | Tenant prefix for every Redis key the grove KV store writes; empty keeps `dispatch:*` |
 | `Concurrency` | `concurrency` | `int` | `10` | Max concurrent jobs |
 | `Queues` | `queues` | `[]string` | `["default"]` | Queues to poll |
 | `DisableRoutes` | `disable_routes` | `bool` | `false` | Skip HTTP route registration |
diff --git a/docs/content/docs/stores/redis.mdx b/docs/content/docs/stores/redis.mdx
index 3080745..f4b3d6b 100644
--- a/docs/content/docs/stores/redis.mdx
+++ b/docs/content/docs/stores/redis.mdx
@@ -27,6 +27,22 @@ if err := s.Ping(ctx); err != nil {
 d, err := dispatch.New(dispatch.WithStore(s))
 ```
 
+## Sharing one Redis between instances
+
+All keys sit under `dispatch:`. To run more than one dispatch instance
+against the same Redis, give each a key prefix and the store puts every
+key, event stream and wake channel under it:
+
+```go
+s := redis.New(kvStore, redis.WithKeyPrefix("ws_acme:"))
+// jobs land at ws_acme:dispatch:job:<id>, the queue at ws_acme:dispatch:queue:<name>
+```
+
+An instance with a prefix never dequeues another instance's jobs, never
+sees its cron names as duplicates, and takes leadership independently. An
+empty prefix keeps the historical unprefixed keys, so existing deployments
+carry on unchanged.
+
 ## Internals
 
 | Aspect | Detail |
diff --git a/extension/config.go b/extension/config.go
index 4cb17b2..b6ad13b 100644
--- a/extension/config.go
+++ b/extension/config.go
@@ -50,6 +50,12 @@ type Config struct {
 	// (unnamed) kv.Store is used.
 	GroveKV string `json:"grove_kv" mapstructure:"grove_kv" yaml:"grove_kv"`
 
+	// KeyPrefix namespaces every key the Redis-backed store writes, so
+	// several dispatch instances can share one Redis without seeing each
+	// other's jobs, cron locks or leadership. Only the grove KV path reads
+	// it. Empty keeps the historical "dispatch:" keys.
+	KeyPrefix string `json:"key_prefix" mapstructure:"key_prefix" yaml:"key_prefix"`
+
 	// Artifacts configures the artifact plane.
 	Artifacts ArtifactConfig `json:"artifacts" mapstructure:"artifacts" yaml:"artifacts"`
 
diff --git a/extension/extension.go b/extension/extension.go
index 0c3b0ea..8f04b95 100644
--- a/extension/extension.go
+++ b/extension/extension.go
@@ -165,7 +165,7 @@ func (e *Extension) init(fapp forge.App) error {
 		if err != nil {
 			return fmt.Errorf("dispatch: %w", err)
 		}
-		e.dispatchOpts = append(e.dispatchOpts, dispatch.WithStore(redisstore.New(kvStore)))
+		e.dispatchOpts = append(e.dispatchOpts, dispatch.WithStore(e.buildStoreFromGroveKV(kvStore)))
 	} else if db, err := vessel.Inject[*grove.DB](fapp.Container()); err == nil {
 		// Auto-discover default grove.DB from container (matches authsome/cortex pattern).
 		s, err := e.buildStoreFromGroveDB(db)
@@ -498,6 +498,7 @@ func (e *Extension) loadConfiguration() error {
 		forge.F("base_path", e.config.BasePath),
 		forge.F("grove_database", e.config.GroveDatabase),
 		forge.F("grove_kv", e.config.GroveKV),
+		forge.F("key_prefix", e.config.KeyPrefix),
 	)
 
 	return nil
@@ -624,6 +625,9 @@ func (e *Extension) mergeConfigurations(yamlConfig, programmaticConfig Config) C
 	if yamlConfig.GroveKV == "" && programmaticConfig.GroveKV != "" {
 		yamlConfig.GroveKV = programmaticConfig.GroveKV
 	}
+	if yamlConfig.KeyPrefix == "" && programmaticConfig.KeyPrefix != "" {
+		yamlConfig.KeyPrefix = programmaticConfig.KeyPrefix
+	}
 	if yamlConfig.DWPBasePath == "" && programmaticConfig.DWPBasePath != "" {
 		yamlConfig.DWPBasePath = programmaticConfig.DWPBasePath
 	}
@@ -665,6 +669,12 @@ func (e *Extension) buildStoreFromGroveDB(db *grove.DB) (dispatch.Storer, error)
 	}
 }
 
+// buildStoreFromGroveKV wraps a grove KV store in the Redis backend,
+// namespaced by the configured key prefix.
+func (e *Extension) buildStoreFromGroveKV(kvStore *kv.Store) dispatch.Storer {
+	return redisstore.New(kvStore, redisstore.WithKeyPrefix(e.config.KeyPrefix))
+}
+
 // resolveGroveKV resolves a *kv.Store from the DI container.
 // If GroveKV is set, it looks up the named store; otherwise it uses the default.
 func (e *Extension) resolveGroveKV(fapp forge.App) (*kv.Store, error) {
diff --git a/extension/kvprefix_internal_test.go b/extension/kvprefix_internal_test.go
new file mode 100644
index 0000000..aa06609
--- /dev/null
+++ b/extension/kvprefix_internal_test.go
@@ -0,0 +1,66 @@
+package extension
+
+import (
+	"testing"
+
+	"github.com/xraph/grove/kv"
+	"github.com/xraph/grove/kv/drivers/redisdriver"
+
+	redisstore "github.com/xraph/dispatch/store/redis"
+)
+
+// unopenedKV returns a Redis-driver KV store that never dials. Building the
+// dispatch store needs only the driver type, so the tests below can pin the
+// wiring without a Redis on the box.
+func unopenedKV(t *testing.T) *kv.Store {
+	t.Helper()
+	s, err := kv.Open(redisdriver.New())
+	if err != nil {
+		t.Fatalf("open kv: %v", err)
+	}
+	return s
+}
+
+func TestBuildStoreFromGroveKV_appliesKeyPrefix(t *testing.T) {
+	tests := []struct {
+		name   string
+		prefix string
+	}{
+		{name: "no prefix keeps the historical keys", prefix: ""},
+		{name: "tenant prefix reaches the redis store", prefix: "ws_acme:"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := New(WithKVKeyPrefix(tt.prefix))
+			s, ok := e.buildStoreFromGroveKV(unopenedKV(t)).(*redisstore.Store)
+			if !ok {
+				t.Fatal("grove KV must build a redis store")
+			}
+			if got := s.KeyPrefix(); got != tt.prefix {
+				t.Fatalf("store key prefix = %q, want %q", got, tt.prefix)
+			}
+		})
+	}
+}
+
+func TestMergeConfigurations_keyPrefixYAMLWins(t *testing.T) {
+	e := New()
+	tests := []struct {
+		name         string
+		yaml         string
+		programmatic string
+		want         string
+	}{
+		{name: "yaml set, programmatic empty", yaml: "ws_yaml:", programmatic: "", want: "ws_yaml:"},
+		{name: "yaml empty, programmatic set", yaml: "", programmatic: "ws_go:", want: "ws_go:"},
+		{name: "both set, yaml wins", yaml: "ws_yaml:", programmatic: "ws_go:", want: "ws_yaml:"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := e.mergeConfigurations(Config{KeyPrefix: tt.yaml}, Config{KeyPrefix: tt.programmatic})
+			if got.KeyPrefix != tt.want {
+				t.Fatalf("merged KeyPrefix = %q, want %q", got.KeyPrefix, tt.want)
+			}
+		})
+	}
+}
diff --git a/extension/options.go b/extension/options.go
index 02d8cb4..1f66681 100644
--- a/extension/options.go
+++ b/extension/options.go
@@ -205,6 +205,15 @@ func WithGroveKV(name string) ExtOption {
 	}
 }
 
+// WithKVKeyPrefix namespaces the Redis-backed store's keys (see
+// Config.KeyPrefix). Pair it with WithGroveKV; the grove database path
+// ignores it.
+func WithKVKeyPrefix(prefix string) ExtOption {
+	return func(e *Extension) {
+		e.config.KeyPrefix = prefix
+	}
+}
+
 // WithDWP enables the Dispatch Wire Protocol (DWP) for real-time
 // client communication over WebSocket, SSE, and HTTP RPC.
 // Options configure authentication, codec, and server behaviour.
diff --git a/store/redis/artifact.go b/store/redis/artifact.go
index 6441917..a5df9ee 100644
--- a/store/redis/artifact.go
+++ b/store/redis/artifact.go
@@ -130,7 +130,7 @@ func linkField(name string, attempt int) string {
 // two concurrent creates at the same coordinates resolve to one winner
 // and one ErrExists.
 func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link *artifact.Link) error {
-	guard := artifactKeyGuard(a.Backend, a.Bucket, a.Key)
+	guard := s.keys.artifactGuard(a.Backend, a.Bucket, a.Key)
 
 	ok, err := s.rdb.SetNX(ctx, guard, a.ID.String(), 0).Result()
 	if err != nil {
@@ -141,14 +141,14 @@ func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link *
 		return artifact.ErrExists
 	}
 
-	if err := s.setEntity(ctx, artifactKey(a.ID.String()), toArtifactEntity(a)); err != nil {
+	if err := s.setEntity(ctx, s.keys.artifact(a.ID.String()), toArtifactEntity(a)); err != nil {
 		// Release the guard so the coordinates are not permanently burned.
 		s.rdb.Del(ctx, guard)
 
 		return fmt.Errorf("dispatch/redis: create artifact: %w", err)
 	}
 
-	if err := s.rdb.SAdd(ctx, artifactIDsKey, a.ID.String()).Err(); err != nil {
+	if err := s.rdb.SAdd(ctx, s.keys.artifactIDs(), a.ID.String()).Err(); err != nil {
 		return fmt.Errorf("dispatch/redis: index artifact: %w", err)
 	}
 
@@ -157,7 +157,7 @@ func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link *
 	// no durable artifact is ever a member.
 	if a.Lifecycle == artifact.Ephemeral {
 		score := float64(a.CreatedAt.UnixNano())
-		if err := s.rdb.ZAdd(ctx, artifactEphemeralKey, goredis.Z{Score: score, Member: a.ID.String()}).Err(); err != nil {
+		if err := s.rdb.ZAdd(ctx, s.keys.artifactEphemeral(), goredis.Z{Score: score, Member: a.ID.String()}).Err(); err != nil {
 			return fmt.Errorf("dispatch/redis: index ephemeral artifact: %w", err)
 		}
 	}
@@ -187,7 +187,7 @@ func (s *Store) GetArtifact(ctx context.Context, artifactID id.ArtifactID) (*art
 func (s *Store) loadArtifact(ctx context.Context, artifactID string) (*artifact.Artifact, error) {
 	var e artifactEntity
 
-	if err := s.getEntity(ctx, artifactKey(artifactID), &e); err != nil {
+	if err := s.getEntity(ctx, s.keys.artifact(artifactID), &e); err != nil {
 		if isNotFound(err) {
 			return nil, artifact.ErrNotFound
 		}
@@ -200,7 +200,7 @@ func (s *Store) loadArtifact(ctx context.Context, artifactID string) (*artifact.
 
 // FindArtifactByKey retrieves a live artifact by its storage coordinates.
 func (s *Store) FindArtifactByKey(ctx context.Context, backend, bucket, key string) (*artifact.Artifact, error) {
-	got, err := s.rdb.Get(ctx, artifactKeyGuard(backend, bucket, key)).Result()
+	got, err := s.rdb.Get(ctx, s.keys.artifactGuard(backend, bucket, key)).Result()
 	if err != nil {
 		return nil, artifact.ErrNotFound
 	}
@@ -221,7 +221,7 @@ func (s *Store) UpdateArtifact(ctx context.Context, a *artifact.Artifact) error
 	existing.ContentType = a.ContentType
 	existing.ExpiresAt = a.ExpiresAt
 
-	if err := s.setEntity(ctx, artifactKey(a.ID.String()), toArtifactEntity(existing)); err != nil {
+	if err := s.setEntity(ctx, s.keys.artifact(a.ID.String()), toArtifactEntity(existing)); err != nil {
 		return fmt.Errorf("dispatch/redis: update artifact: %w", err)
 	}
 
@@ -230,7 +230,7 @@ func (s *Store) UpdateArtifact(ctx context.Context, a *artifact.Artifact) error
 
 // ListArtifacts returns artifacts matching the given options, newest first.
 func (s *Store) ListArtifacts(ctx context.Context, opts artifact.ListOpts) ([]*artifact.Artifact, error) {
-	ids, err := s.rdb.SMembers(ctx, artifactIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.artifactIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list artifact ids: %w", err)
 	}
@@ -300,12 +300,12 @@ func (s *Store) LinkArtifact(ctx context.Context, link *artifact.Link) error {
 	owner := artifact.OwnerRef{Kind: link.OwnerKind, ID: link.OwnerID}
 	field := linkField(link.Name, link.Attempt)
 
-	if err := s.rdb.HSet(ctx, ownerLinksKey(string(owner.Kind), owner.ID), field, raw).Err(); err != nil {
+	if err := s.rdb.HSet(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID), field, raw).Err(); err != nil {
 		return fmt.Errorf("dispatch/redis: link artifact: %w", err)
 	}
 
 	member := string(link.OwnerKind) + "\x00" + link.OwnerID + "\x00" + field
-	if err := s.rdb.SAdd(ctx, artifactLinksKey(link.ArtifactID.String()), member).Err(); err != nil {
+	if err := s.rdb.SAdd(ctx, s.keys.artifactLinks(link.ArtifactID.String()), member).Err(); err != nil {
 		return fmt.Errorf("dispatch/redis: index artifact link: %w", err)
 	}
 
@@ -314,7 +314,7 @@ func (s *Store) LinkArtifact(ctx context.Context, link *artifact.Link) error {
 
 // ListLinks returns every link belonging to the given owner.
 func (s *Store) ListLinks(ctx context.Context, owner artifact.OwnerRef) ([]*artifact.Link, error) {
-	vals, err := s.rdb.HGetAll(ctx, ownerLinksKey(string(owner.Kind), owner.ID)).Result()
+	vals, err := s.rdb.HGetAll(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID)).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list links: %w", err)
 	}
@@ -449,7 +449,7 @@ func (s *Store) SweepEphemeral(
 		limit = defaultSweepLimit
 	}
 
-	ids, err := s.rdb.ZRange(ctx, artifactEphemeralKey, 0, -1).Result()
+	ids, err := s.rdb.ZRange(ctx, s.keys.artifactEphemeral(), 0, -1).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: sweep ephemeral: %w", err)
 	}
@@ -515,7 +515,7 @@ func (s *Store) SweepEphemeral(
 
 // linksForArtifact resolves every link pointing at an artifact.
 func (s *Store) linksForArtifact(ctx context.Context, artifactID string) ([]*artifact.Link, error) {
-	members, err := s.rdb.SMembers(ctx, artifactLinksKey(artifactID)).Result()
+	members, err := s.rdb.SMembers(ctx, s.keys.artifactLinks(artifactID)).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list artifact links: %w", err)
 	}
@@ -530,7 +530,7 @@ func (s *Store) linksForArtifact(ctx context.Context, artifactID string) ([]*art
 
 		owner := artifact.OwnerRef{Kind: artifact.OwnerKind(kind), ID: ownerID}
 
-		raw, herr := s.rdb.HGet(ctx, ownerLinksKey(string(owner.Kind), owner.ID), field).Result()
+		raw, herr := s.rdb.HGet(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID), field).Result()
 		if herr != nil {
 			continue
 		}
@@ -604,7 +604,7 @@ func (s *Store) ownerTerminalAt(ctx context.Context, l *artifact.Link) (time.Tim
 	switch l.OwnerKind {
 	case artifact.OwnerJob:
 		var e jobEntity
-		if err := s.getEntity(ctx, jobKey(l.OwnerID), &e); err != nil {
+		if err := s.getEntity(ctx, s.keys.job(l.OwnerID), &e); err != nil {
 			if isNotFound(err) {
 				return l.CreatedAt, true, nil
 			}
@@ -624,7 +624,7 @@ func (s *Store) ownerTerminalAt(ctx context.Context, l *artifact.Link) (time.Tim
 
 	case artifact.OwnerRun, artifact.OwnerStep:
 		var e runEntity
-		if err := s.getEntity(ctx, runKey(l.OwnerID), &e); err != nil {
+		if err := s.getEntity(ctx, s.keys.run(l.OwnerID), &e); err != nil {
 			if isNotFound(err) {
 				return l.CreatedAt, true, nil
 			}
@@ -668,7 +668,7 @@ func (s *Store) SweepOrphans(
 
 	// The ephemeral index is scored by creation time, so the cutoff is a
 	// range query rather than a scan.
-	ids, err := s.rdb.ZRangeByScore(ctx, artifactEphemeralKey, &goredis.ZRangeBy{
+	ids, err := s.rdb.ZRangeByScore(ctx, s.keys.artifactEphemeral(), &goredis.ZRangeBy{
 		Min: "-inf",
 		Max: strconv.FormatInt(cutoff.UnixNano(), 10),
 	}).Result()
@@ -700,7 +700,7 @@ func (s *Store) SweepOrphans(
 			continue
 		}
 
-		n, cerr := s.rdb.SCard(ctx, artifactLinksKey(a.ID.String())).Result()
+		n, cerr := s.rdb.SCard(ctx, s.keys.artifactLinks(a.ID.String())).Result()
 		if cerr != nil {
 			return nil, fmt.Errorf("dispatch/redis: count artifact links: %w", cerr)
 		}
@@ -737,20 +737,20 @@ func (s *Store) markDeleted(
 		deleted := at
 		clone.DeletedAt = &deleted
 
-		if err := s.setEntity(ctx, artifactKey(clone.ID.String()), toArtifactEntity(clone)); err != nil {
+		if err := s.setEntity(ctx, s.keys.artifact(clone.ID.String()), toArtifactEntity(clone)); err != nil {
 			return nil, fmt.Errorf("dispatch/redis: mark artifact deleted: %w", err)
 		}
 
 		// Release the live-key guard so the coordinates become reusable,
 		// and index the deletion time for the purge pass.
-		s.rdb.Del(ctx, artifactKeyGuard(clone.Backend, clone.Bucket, clone.Key))
+		s.rdb.Del(ctx, s.keys.artifactGuard(clone.Backend, clone.Bucket, clone.Key))
 
-		if err := s.rdb.ZAdd(ctx, artifactDeletedKey,
+		if err := s.rdb.ZAdd(ctx, s.keys.artifactDeleted(),
 			goredis.Z{Score: float64(at.UnixNano()), Member: clone.ID.String()}).Err(); err != nil {
 			return nil, fmt.Errorf("dispatch/redis: index deleted artifact: %w", err)
 		}
 
-		if err := s.rdb.ZRem(ctx, artifactEphemeralKey, clone.ID.String()).Err(); err != nil {
+		if err := s.rdb.ZRem(ctx, s.keys.artifactEphemeral(), clone.ID.String()).Err(); err != nil {
 			return nil, fmt.Errorf("dispatch/redis: deindex ephemeral artifact: %w", err)
 		}
 
@@ -772,7 +772,7 @@ func (s *Store) ListPurgeable(
 
 	cutoff := time.Now().UTC().Add(-grace)
 
-	ids, err := s.rdb.ZRangeByScore(ctx, artifactDeletedKey, &goredis.ZRangeBy{
+	ids, err := s.rdb.ZRangeByScore(ctx, s.keys.artifactDeleted(), &goredis.ZRangeBy{
 		Min: "-inf",
 		Max: strconv.FormatInt(cutoff.UnixNano(), 10),
 	}).Result()
@@ -817,22 +817,22 @@ func (s *Store) PurgeArtifact(ctx context.Context, artifactID id.ArtifactID) err
 
 	for _, l := range links {
 		owner := artifact.OwnerRef{Kind: l.OwnerKind, ID: l.OwnerID}
-		if herr := s.rdb.HDel(ctx, ownerLinksKey(string(owner.Kind), owner.ID), linkField(l.Name, l.Attempt)).Err(); herr != nil {
+		if herr := s.rdb.HDel(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID), linkField(l.Name, l.Attempt)).Err(); herr != nil {
 			return fmt.Errorf("dispatch/redis: purge artifact link: %w", herr)
 		}
 	}
 
 	a, err := s.loadArtifact(ctx, key)
 	if err == nil {
-		s.rdb.Del(ctx, artifactKeyGuard(a.Backend, a.Bucket, a.Key))
+		s.rdb.Del(ctx, s.keys.artifactGuard(a.Backend, a.Bucket, a.Key))
 	}
 
 	pipe := s.rdb.TxPipeline()
-	pipe.Del(ctx, artifactKey(key))
-	pipe.Del(ctx, artifactLinksKey(key))
-	pipe.SRem(ctx, artifactIDsKey, key)
-	pipe.ZRem(ctx, artifactEphemeralKey, key)
-	pipe.ZRem(ctx, artifactDeletedKey, key)
+	pipe.Del(ctx, s.keys.artifact(key))
+	pipe.Del(ctx, s.keys.artifactLinks(key))
+	pipe.SRem(ctx, s.keys.artifactIDs(), key)
+	pipe.ZRem(ctx, s.keys.artifactEphemeral(), key)
+	pipe.ZRem(ctx, s.keys.artifactDeleted(), key)
 
 	if _, err := pipe.Exec(ctx); err != nil {
 		return fmt.Errorf("dispatch/redis: purge artifact: %w", err)
diff --git a/store/redis/cluster.go b/store/redis/cluster.go
index c638299..45f0668 100644
--- a/store/redis/cluster.go
+++ b/store/redis/cluster.go
@@ -63,14 +63,14 @@ func fromWorkerEntity(e *workerEntity) (*cluster.Worker, error) {
 // RegisterWorker adds a new worker to the cluster registry.
 func (s *Store) RegisterWorker(ctx context.Context, w *cluster.Worker) error {
 	wID := w.ID.String()
-	key := workerKey(wID)
+	key := s.keys.worker(wID)
 
 	e := toWorkerEntity(w)
 	if err := s.setEntity(ctx, key, e); err != nil {
 		return fmt.Errorf("dispatch/redis: register worker set: %w", err)
 	}
 
-	if err := s.rdb.SAdd(ctx, workerIDsKey, wID).Err(); err != nil {
+	if err := s.rdb.SAdd(ctx, s.keys.workerIDs(), wID).Err(); err != nil {
 		return fmt.Errorf("dispatch/redis: register worker index: %w", err)
 	}
 	return nil
@@ -79,7 +79,7 @@ func (s *Store) RegisterWorker(ctx context.Context, w *cluster.Worker) error {
 // DeregisterWorker removes a worker from the cluster registry.
 func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) error {
 	wID := workerID.String()
-	key := workerKey(wID)
+	key := s.keys.worker(wID)
 
 	exists, err := s.entityExists(ctx, key)
 	if err != nil {
@@ -91,7 +91,7 @@ func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) erro
 
 	pipe := s.rdb.TxPipeline()
 	pipe.Del(ctx, key)
-	pipe.SRem(ctx, workerIDsKey, wID)
+	pipe.SRem(ctx, s.keys.workerIDs(), wID)
 	_, err = pipe.Exec(ctx)
 	if err != nil {
 		return fmt.Errorf("dispatch/redis: deregister worker: %w", err)
@@ -101,7 +101,7 @@ func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) erro
 
 // HeartbeatWorker updates the last-seen timestamp for a worker.
 func (s *Store) HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error {
-	key := workerKey(workerID.String())
+	key := s.keys.worker(workerID.String())
 
 	var e workerEntity
 	if err := s.getEntity(ctx, key, &e); err != nil {
@@ -117,7 +117,7 @@ func (s *Store) HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error
 
 // ListWorkers returns all registered workers.
 func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error) {
-	ids, err := s.rdb.SMembers(ctx, workerIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.workerIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list workers: %w", err)
 	}
@@ -125,7 +125,7 @@ func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error) {
 	workers := make([]*cluster.Worker, 0, len(ids))
 	for _, wID := range ids {
 		var e workerEntity
-		if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil {
 			continue
 		}
 		w, convErr := fromWorkerEntity(&e)
@@ -142,7 +142,7 @@ func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error) {
 func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) (int64, error) {
 	cutoff := now().Add(-threshold)
 
-	ids, err := s.rdb.SMembers(ctx, workerIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.workerIDs()).Result()
 	if err != nil {
 		return 0, fmt.Errorf("dispatch/redis: delete stale smembers: %w", err)
 	}
@@ -150,11 +150,11 @@ func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration)
 	var deleted int64
 	for _, wID := range ids {
 		var e workerEntity
-		if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil {
 			// Orphaned set member without a backing entity — treat as
 			// stale and remove from the index.
 			if isNotFound(getErr) {
-				if remErr := s.rdb.SRem(ctx, workerIDsKey, wID).Err(); remErr == nil {
+				if remErr := s.rdb.SRem(ctx, s.keys.workerIDs(), wID).Err(); remErr == nil {
 					deleted++
 				}
 			}
@@ -162,8 +162,8 @@ func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration)
 		}
 		if e.LastSeen.Before(cutoff) {
 			pipe := s.rdb.TxPipeline()
-			pipe.Del(ctx, workerKey(wID))
-			pipe.SRem(ctx, workerIDsKey, wID)
+			pipe.Del(ctx, s.keys.worker(wID))
+			pipe.SRem(ctx, s.keys.workerIDs(), wID)
 			if _, execErr := pipe.Exec(ctx); execErr == nil {
 				deleted++
 			}
@@ -176,7 +176,7 @@ func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration)
 func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([]*cluster.Worker, error) {
 	cutoff := now().Add(-threshold)
 
-	ids, err := s.rdb.SMembers(ctx, workerIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.workerIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: reap smembers: %w", err)
 	}
@@ -184,7 +184,7 @@ func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([
 	var dead []*cluster.Worker
 	for _, wID := range ids {
 		var e workerEntity
-		if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil {
 			continue
 		}
 		if e.LastSeen.Before(cutoff) {
@@ -201,7 +201,7 @@ func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([
 // AcquireLeadership attempts to become the cluster leader.
 func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error) {
 	wID := workerID.String()
-	wKey := workerKey(wID)
+	wKey := s.keys.worker(wID)
 
 	// Check worker exists.
 	exists, err := s.entityExists(ctx, wKey)
@@ -213,7 +213,7 @@ func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl
 	}
 
 	// Try SET NX with TTL (atomic acquire).
-	ok, err := s.rdb.SetNX(ctx, leaderKey, wID, ttl).Result()
+	ok, err := s.rdb.SetNX(ctx, s.keys.leader(), wID, ttl).Result()
 	if err != nil {
 		return false, fmt.Errorf("dispatch/redis: acquire leadership setnx: %w", err)
 	}
@@ -230,13 +230,13 @@ func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl
 	}
 
 	// Check if we already hold it.
-	current, err := s.rdb.Get(ctx, leaderKey).Result()
+	current, err := s.rdb.Get(ctx, s.keys.leader()).Result()
 	if err != nil && !isRedisNil(err) {
 		return false, fmt.Errorf("dispatch/redis: acquire leadership get: %w", err)
 	}
 	if current == wID {
 		// Re-acquire: extend TTL.
-		_ = s.rdb.Expire(ctx, leaderKey, ttl).Err() //nolint:errcheck // best-effort
+		_ = s.rdb.Expire(ctx, s.keys.leader(), ttl).Err() //nolint:errcheck // best-effort
 		until := now().Add(ttl)
 		var e workerEntity
 		if getErr := s.getEntity(ctx, wKey, &e); getErr == nil {
@@ -254,7 +254,7 @@ func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl
 func (s *Store) RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error) {
 	wID := workerID.String()
 
-	current, err := s.rdb.Get(ctx, leaderKey).Result()
+	current, err := s.rdb.Get(ctx, s.keys.leader()).Result()
 	if err != nil {
 		if isRedisNil(err) {
 			return false, nil // no leader
@@ -265,19 +265,19 @@ func (s *Store) RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl t
 		return false, nil // not the leader
 	}
 
-	_ = s.rdb.Expire(ctx, leaderKey, ttl).Err() //nolint:errcheck // best-effort
+	_ = s.rdb.Expire(ctx, s.keys.leader(), ttl).Err() //nolint:errcheck // best-effort
 	until := now().Add(ttl)
 	var e workerEntity
-	if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr == nil {
+	if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr == nil {
 		e.LeaderUntil = &until
-		_ = s.setEntity(ctx, workerKey(wID), &e) //nolint:errcheck // best-effort update
+		_ = s.setEntity(ctx, s.keys.worker(wID), &e) //nolint:errcheck // best-effort update
 	}
 	return true, nil
 }
 
 // GetLeader returns the current cluster leader, or nil if there is no leader.
 func (s *Store) GetLeader(ctx context.Context) (*cluster.Worker, error) {
-	wID, err := s.rdb.Get(ctx, leaderKey).Result()
+	wID, err := s.rdb.Get(ctx, s.keys.leader()).Result()
 	if err != nil {
 		if isRedisNil(err) {
 			return nil, nil // no leader
@@ -286,7 +286,7 @@ func (s *Store) GetLeader(ctx context.Context) (*cluster.Worker, error) {
 	}
 
 	var e workerEntity
-	if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil {
+	if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil {
 		return nil, nil // leader key exists but worker gone
 	}
 	return fromWorkerEntity(&e)
diff --git a/store/redis/cron.go b/store/redis/cron.go
index 52edf4b..1f2c388 100644
--- a/store/redis/cron.go
+++ b/store/redis/cron.go
@@ -80,10 +80,10 @@ func fromCronEntity(e *cronEntity) (*cron.Entry, error) {
 // RegisterCron persists a new cron entry.
 func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error {
 	eID := entry.ID.String()
-	key := cronKey(eID)
+	key := s.keys.cron(eID)
 
 	// Check for duplicate name.
-	existing, err := s.rdb.HGet(ctx, cronNamesKey, entry.Name).Result()
+	existing, err := s.rdb.HGet(ctx, s.keys.cronNames(), entry.Name).Result()
 	if err != nil && !isRedisNil(err) {
 		return fmt.Errorf("dispatch/redis: register cron check name: %w", err)
 	}
@@ -97,8 +97,8 @@ func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error {
 	}
 
 	pipe := s.rdb.TxPipeline()
-	pipe.SAdd(ctx, cronIDsKey, eID)
-	pipe.HSet(ctx, cronNamesKey, entry.Name, eID)
+	pipe.SAdd(ctx, s.keys.cronIDs(), eID)
+	pipe.HSet(ctx, s.keys.cronNames(), entry.Name, eID)
 	_, err = pipe.Exec(ctx)
 	if err != nil {
 		return fmt.Errorf("dispatch/redis: register cron indexes: %w", err)
@@ -109,7 +109,7 @@ func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error {
 // GetCron retrieves a cron entry by ID.
 func (s *Store) GetCron(ctx context.Context, entryID id.CronID) (*cron.Entry, error) {
 	var e cronEntity
-	if err := s.getEntity(ctx, cronKey(entryID.String()), &e); err != nil {
+	if err := s.getEntity(ctx, s.keys.cron(entryID.String()), &e); err != nil {
 		if isNotFound(err) {
 			return nil, dispatch.ErrCronNotFound
 		}
@@ -120,7 +120,7 @@ func (s *Store) GetCron(ctx context.Context, entryID id.CronID) (*cron.Entry, er
 
 // ListCrons returns all cron entries.
 func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error) {
-	ids, err := s.rdb.SMembers(ctx, cronIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.cronIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list crons: %w", err)
 	}
@@ -128,7 +128,7 @@ func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error) {
 	entries := make([]*cron.Entry, 0, len(ids))
 	for _, eID := range ids {
 		var e cronEntity
-		if getErr := s.getEntity(ctx, cronKey(eID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.cron(eID), &e); getErr != nil {
 			continue
 		}
 		entry, convErr := fromCronEntity(&e)
@@ -143,7 +143,7 @@ func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error) {
 // AcquireCronLock attempts to acquire a distributed lock for a cron entry.
 func (s *Store) AcquireCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID, ttl time.Duration) (bool, error) {
 	eID := entryID.String()
-	key := cronKey(eID)
+	key := s.keys.cron(eID)
 	wID := workerID.String()
 	t := now()
 	until := t.Add(ttl)
@@ -177,7 +177,7 @@ func (s *Store) AcquireCronLock(ctx context.Context, entryID id.CronID, workerID
 
 // ReleaseCronLock releases the distributed lock for a cron entry.
 func (s *Store) ReleaseCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID) error {
-	key := cronKey(entryID.String())
+	key := s.keys.cron(entryID.String())
 	wID := workerID.String()
 
 	var e cronEntity
@@ -200,7 +200,7 @@ func (s *Store) ReleaseCronLock(ctx context.Context, entryID id.CronID, workerID
 
 // UpdateCronLastRun records when a cron entry last fired.
 func (s *Store) UpdateCronLastRun(ctx context.Context, entryID id.CronID, at time.Time) error {
-	key := cronKey(entryID.String())
+	key := s.keys.cron(entryID.String())
 	var e cronEntity
 	if err := s.getEntity(ctx, key, &e); err != nil {
 		if isNotFound(err) {
@@ -216,7 +216,7 @@ func (s *Store) UpdateCronLastRun(ctx context.Context, entryID id.CronID, at tim
 
 // UpdateCronEntry updates a cron entry.
 func (s *Store) UpdateCronEntry(ctx context.Context, entry *cron.Entry) error {
-	key := cronKey(entry.ID.String())
+	key := s.keys.cron(entry.ID.String())
 	exists, err := s.entityExists(ctx, key)
 	if err != nil {
 		return fmt.Errorf("dispatch/redis: update cron exists: %w", err)
@@ -233,7 +233,7 @@ func (s *Store) UpdateCronEntry(ctx context.Context, entry *cron.Entry) error {
 // DeleteCron removes a cron entry by ID.
 func (s *Store) DeleteCron(ctx context.Context, entryID id.CronID) error {
 	eID := entryID.String()
-	key := cronKey(eID)
+	key := s.keys.cron(eID)
 
 	// Get name for name index cleanup.
 	var e cronEntity
@@ -246,9 +246,9 @@ func (s *Store) DeleteCron(ctx context.Context, entryID id.CronID) error {
 
 	pipe := s.rdb.TxPipeline()
 	pipe.Del(ctx, key)
-	pipe.SRem(ctx, cronIDsKey, eID)
+	pipe.SRem(ctx, s.keys.cronIDs(), eID)
 	if e.Name != "" {
-		pipe.HDel(ctx, cronNamesKey, e.Name)
+		pipe.HDel(ctx, s.keys.cronNames(), e.Name)
 	}
 	_, err := pipe.Exec(ctx)
 	if err != nil {
diff --git a/store/redis/dequeue.go b/store/redis/dequeue.go
index 59277fa..116fcdb 100644
--- a/store/redis/dequeue.go
+++ b/store/redis/dequeue.go
@@ -233,7 +233,7 @@ func (s *Store) scanQueue(
 	q string,
 	t time.Time,
 ) ([]dequeueCandidate, error) {
-	key := queueKey(q)
+	key := s.keys.queue(q)
 	full := !opts.IsUnbounded() || len(opts.PreferredHashes()) > 0
 
 	var (
@@ -353,7 +353,7 @@ func (s *Store) readJobEntities(ctx context.Context, ids []string) ([]*jobEntity
 
 	cmds := make([]*goredis.StringCmd, len(ids))
 	for i, jID := range ids {
-		cmds[i] = pipe.Get(ctx, jobKey(jID))
+		cmds[i] = pipe.Get(ctx, s.keys.job(jID))
 	}
 
 	// A missing key makes Exec report goredis.Nil for the batch as a
@@ -429,7 +429,7 @@ func (s *Store) claimCandidates(
 
 	rems := make([]*goredis.IntCmd, len(candidates))
 	for i, c := range candidates {
-		rems[i] = pipe.ZRem(ctx, queueKey(c.queue), c.id)
+		rems[i] = pipe.ZRem(ctx, s.keys.queue(c.queue), c.id)
 	}
 
 	if _, err := pipe.Exec(ctx); err != nil {
@@ -445,7 +445,7 @@ func (s *Store) claimCandidates(
 			continue // another worker removed it first
 		}
 
-		key := jobKey(c.id)
+		key := s.keys.job(c.id)
 
 		var e jobEntity
 		if getErr := s.getEntity(ctx, key, &e); getErr != nil {
diff --git a/store/redis/dlq.go b/store/redis/dlq.go
index 165f573..d79863b 100644
--- a/store/redis/dlq.go
+++ b/store/redis/dlq.go
@@ -108,14 +108,14 @@ func fromDLQEntity(e *dlqEntity) (*dlq.Entry, error) {
 // PushDLQ adds a failed job entry to the dead letter queue.
 func (s *Store) PushDLQ(ctx context.Context, entry *dlq.Entry) error {
 	eID := entry.ID.String()
-	key := dlqKey(eID)
+	key := s.keys.dlq(eID)
 
 	e := toDLQEntity(entry)
 	if err := s.setEntity(ctx, key, e); err != nil {
 		return fmt.Errorf("dispatch/redis: push dlq set: %w", err)
 	}
 
-	if err := s.rdb.SAdd(ctx, dlqIDsKey, eID).Err(); err != nil {
+	if err := s.rdb.SAdd(ctx, s.keys.dlqIDs(), eID).Err(); err != nil {
 		return fmt.Errorf("dispatch/redis: push dlq index: %w", err)
 	}
 	return nil
@@ -123,7 +123,7 @@ func (s *Store) PushDLQ(ctx context.Context, entry *dlq.Entry) error {
 
 // ListDLQ returns DLQ entries matching the given options.
 func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, error) {
-	ids, err := s.rdb.SMembers(ctx, dlqIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.dlqIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list dlq: %w", err)
 	}
@@ -131,7 +131,7 @@ func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, e
 	entries := make([]*dlq.Entry, 0, len(ids))
 	for _, eID := range ids {
 		var e dlqEntity
-		if getErr := s.getEntity(ctx, dlqKey(eID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.dlq(eID), &e); getErr != nil {
 			continue
 		}
 		if opts.Queue != "" && e.Queue != opts.Queue {
@@ -150,7 +150,7 @@ func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, e
 // GetDLQ retrieves a DLQ entry by ID.
 func (s *Store) GetDLQ(ctx context.Context, entryID id.DLQID) (*dlq.Entry, error) {
 	var e dlqEntity
-	if err := s.getEntity(ctx, dlqKey(entryID.String()), &e); err != nil {
+	if err := s.getEntity(ctx, s.keys.dlq(entryID.String()), &e); err != nil {
 		if isNotFound(err) {
 			return nil, dispatch.ErrDLQNotFound
 		}
@@ -161,7 +161,7 @@ func (s *Store) GetDLQ(ctx context.Context, entryID id.DLQID) (*dlq.Entry, error
 
 // ReplayDLQ marks a DLQ entry as replayed.
 func (s *Store) ReplayDLQ(ctx context.Context, entryID id.DLQID) error {
-	key := dlqKey(entryID.String())
+	key := s.keys.dlq(entryID.String())
 	var e dlqEntity
 	if err := s.getEntity(ctx, key, &e); err != nil {
 		if isNotFound(err) {
@@ -177,14 +177,14 @@ func (s *Store) ReplayDLQ(ctx context.Context, entryID id.DLQID) error {
 
 // PurgeDLQ removes DLQ entries with FailedAt before the given time.
 func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error) {
-	ids, err := s.rdb.SMembers(ctx, dlqIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.dlqIDs()).Result()
 	if err != nil {
 		return 0, fmt.Errorf("dispatch/redis: purge dlq smembers: %w", err)
 	}
 
 	var purged int64
 	for _, eID := range ids {
-		key := dlqKey(eID)
+		key := s.keys.dlq(eID)
 		var e dlqEntity
 		if getErr := s.getEntity(ctx, key, &e); getErr != nil {
 			continue
@@ -193,7 +193,7 @@ func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error) {
 		if e.FailedAt.Before(before) {
 			pipe := s.rdb.TxPipeline()
 			pipe.Del(ctx, key)
-			pipe.SRem(ctx, dlqIDsKey, eID)
+			pipe.SRem(ctx, s.keys.dlqIDs(), eID)
 			if _, pErr := pipe.Exec(ctx); pErr != nil {
 				return purged, fmt.Errorf("dispatch/redis: purge dlq del: %w", pErr)
 			}
@@ -205,7 +205,7 @@ func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error) {
 
 // CountDLQ returns the total number of entries in the dead letter queue.
 func (s *Store) CountDLQ(ctx context.Context) (int64, error) {
-	count, err := s.rdb.SCard(ctx, dlqIDsKey).Result()
+	count, err := s.rdb.SCard(ctx, s.keys.dlqIDs()).Result()
 	if err != nil {
 		return 0, fmt.Errorf("dispatch/redis: count dlq: %w", err)
 	}
diff --git a/store/redis/event.go b/store/redis/event.go
index 8bfb14d..678b425 100644
--- a/store/redis/event.go
+++ b/store/redis/event.go
@@ -56,7 +56,7 @@ func fromEventEntity(e *eventEntity) (*event.Event, error) {
 // PublishEvent persists a new event and adds it to the name's stream.
 func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error {
 	eID := evt.ID.String()
-	key := eventKey(eID)
+	key := s.keys.event(eID)
 
 	e := toEventEntity(evt)
 	if err := s.setEntity(ctx, key, e); err != nil {
@@ -65,7 +65,7 @@ func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error {
 
 	// Add to the named stream so subscribers get notified.
 	if err := s.rdb.XAdd(ctx, &goredis.XAddArgs{
-		Stream: eventStreamKey(evt.Name),
+		Stream: s.keys.eventStream(evt.Name),
 		Values: map[string]interface{}{
 			"event_id": eID,
 		},
@@ -78,7 +78,7 @@ func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error {
 // SubscribeEvent waits for an unacked event matching the given name.
 // Uses stream polling for efficient waiting.
 func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Duration) (*event.Event, error) {
-	stream := eventStreamKey(name)
+	stream := s.keys.eventStream(name)
 	deadline := time.Now().Add(timeout)
 
 	for {
@@ -104,7 +104,7 @@ func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Du
 				continue
 			}
 
-			key := eventKey(eID)
+			key := s.keys.event(eID)
 			var e eventEntity
 			if getErr := s.getEntity(ctx, key, &e); getErr != nil {
 				continue
@@ -132,7 +132,7 @@ func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Du
 
 // AckEvent acknowledges an event, marking it as consumed.
 func (s *Store) AckEvent(ctx context.Context, eventID id.EventID) error {
-	key := eventKey(eventID.String())
+	key := s.keys.event(eventID.String())
 
 	var e eventEntity
 	if err := s.getEntity(ctx, key, &e); err != nil {
diff --git a/store/redis/job.go b/store/redis/job.go
index adaab4f..ed325d0 100644
--- a/store/redis/job.go
+++ b/store/redis/job.go
@@ -181,7 +181,7 @@ func fromJobEntity(e *jobEntity) (*job.Job, error) {
 // EnqueueJob stores the job as a JSON entity and adds it to the queue's Sorted Set.
 func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error {
 	jID := j.ID.String()
-	key := jobKey(jID)
+	key := s.keys.job(jID)
 
 	// Check for duplicate.
 	exists, err := s.entityExists(ctx, key)
@@ -201,11 +201,11 @@ func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error {
 	}
 
 	pipe := s.rdb.TxPipeline()
-	pipe.SAdd(ctx, jobIDsKey, jID)
+	pipe.SAdd(ctx, s.keys.jobIDs(), jID)
 
 	// Add to queue sorted set: score = priority (negated for DESC) + time component.
 	score := jobScore(j.Priority, j.RunAt)
-	pipe.ZAdd(ctx, queueKey(j.Queue), goredis.Z{Score: score, Member: jID})
+	pipe.ZAdd(ctx, s.keys.queue(j.Queue), goredis.Z{Score: score, Member: jID})
 
 	_, err = pipe.Exec(ctx)
 	if err != nil {
@@ -221,7 +221,7 @@ func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error {
 // GetJob retrieves a job by ID.
 func (s *Store) GetJob(ctx context.Context, jobID id.JobID) (*job.Job, error) {
 	var e jobEntity
-	if err := s.getEntity(ctx, jobKey(jobID.String()), &e); err != nil {
+	if err := s.getEntity(ctx, s.keys.job(jobID.String()), &e); err != nil {
 		if isNotFound(err) {
 			return nil, dispatch.ErrJobNotFound
 		}
@@ -260,7 +260,7 @@ func (s *Store) GetJob(ctx context.Context, jobID id.JobID) (*job.Job, error) {
 // safe for a job that was never claimed.
 func (s *Store) UpdateJob(ctx context.Context, j *job.Job) error {
 	jID := j.ID.String()
-	key := jobKey(jID)
+	key := s.keys.job(jID)
 
 	exists, err := s.entityExists(ctx, key)
 	if err != nil {
@@ -280,7 +280,7 @@ func (s *Store) UpdateJob(ctx context.Context, j *job.Job) error {
 	// so j.Queue is the queue it was indexed under. If that ever changes,
 	// the old queue keeps a member pointing at this job, and this function
 	// has to read the stored entity to learn which queue to clear.
-	qk := queueKey(j.Queue)
+	qk := s.keys.queue(j.Queue)
 	runnable := j.State == job.StatePending || j.State == job.StateRetrying
 
 	if runnable {
@@ -306,7 +306,7 @@ func (s *Store) UpdateJob(ctx context.Context, j *job.Job) error {
 // DeleteJob removes a job by ID.
 func (s *Store) DeleteJob(ctx context.Context, jobID id.JobID) error {
 	jID := jobID.String()
-	key := jobKey(jID)
+	key := s.keys.job(jID)
 
 	// Get queue name before deleting to remove from sorted set.
 	var e jobEntity
@@ -320,8 +320,8 @@ func (s *Store) DeleteJob(ctx context.Context, jobID id.JobID) error {
 	// Delete entity via raw Redis DEL (KV store may not have Delete).
 	pipe := s.rdb.TxPipeline()
 	pipe.Del(ctx, key)
-	pipe.SRem(ctx, jobIDsKey, jID)
-	pipe.ZRem(ctx, queueKey(e.Queue), jID)
+	pipe.SRem(ctx, s.keys.jobIDs(), jID)
+	pipe.ZRem(ctx, s.keys.queue(e.Queue), jID)
 	_, err := pipe.Exec(ctx)
 	if err != nil {
 		return fmt.Errorf("dispatch/redis: delete job: %w", err)
@@ -331,7 +331,7 @@ func (s *Store) DeleteJob(ctx context.Context, jobID id.JobID) error {
 
 // ListJobsByState returns jobs matching the given state.
 func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.ListOpts) ([]*job.Job, error) {
-	ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list jobs smembers: %w", err)
 	}
@@ -339,7 +339,7 @@ func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.L
 	jobs := make([]*job.Job, 0, len(ids))
 	for _, jID := range ids {
 		var e jobEntity
-		if getErr := s.getEntity(ctx, jobKey(jID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.job(jID), &e); getErr != nil {
 			continue // skip missing
 		}
 		if job.State(e.State) != state {
@@ -360,7 +360,7 @@ func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.L
 
 // HeartbeatJob updates the heartbeat timestamp for a running job.
 func (s *Store) HeartbeatJob(ctx context.Context, jobID id.JobID, _ id.WorkerID) error {
-	key := jobKey(jobID.String())
+	key := s.keys.job(jobID.String())
 	var e jobEntity
 	if err := s.getEntity(ctx, key, &e); err != nil {
 		if isNotFound(err) {
@@ -379,7 +379,7 @@ func (s *Store) HeartbeatJob(ctx context.Context, jobID id.JobID, _ id.WorkerID)
 func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]*job.Job, error) {
 	cutoff := now().Add(-threshold)
 
-	ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: reap smembers: %w", err)
 	}
@@ -387,7 +387,7 @@ func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]*
 	var stale []*job.Job
 	for _, jID := range ids {
 		var e jobEntity
-		if getErr := s.getEntity(ctx, jobKey(jID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.job(jID), &e); getErr != nil {
 			continue
 		}
 		if job.State(e.State) != job.StateRunning {
@@ -411,14 +411,14 @@ func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]*
 
 // CountJobs returns the number of jobs matching the given options.
 func (s *Store) CountJobs(ctx context.Context, opts job.CountOpts) (int64, error) {
-	ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result()
 	if err != nil {
 		return 0, fmt.Errorf("dispatch/redis: count smembers: %w", err)
 	}
 
 	var count int64
 	for _, jID := range ids {
-		raw, getErr := s.kv.GetRaw(ctx, jobKey(jID))
+		raw, getErr := s.kv.GetRaw(ctx, s.keys.job(jID))
 		if getErr != nil {
 			continue
 		}
diff --git a/store/redis/keys.go b/store/redis/keys.go
index bcb37b3..c643765 100644
--- a/store/redis/keys.go
+++ b/store/redis/keys.go
@@ -2,109 +2,129 @@ package redis
 
 import "fmt"
 
-// Redis key naming conventions for dispatch data.
-// All keys are prefixed with "dispatch:" to avoid collisions.
+// keys builds every Redis key and channel name the store touches. All of
+// them sit under "dispatch:" so the store can share a Redis with other
+// users of the same database; a tenant prefix on top of that lets several
+// dispatch instances share one Redis without seeing each other's queues,
+// cron locks or leadership.
+type keys struct {
+	prefix string
+}
+
+// base is the namespace every dispatch key lives under, tenant or not.
+const base = "dispatch:"
 
-const keyPrefix = "dispatch:"
+func newKeys(prefix string) keys { return keys{prefix: prefix} }
+
+// full composes the key that hits Redis. The tenant prefix goes outside
+// the dispatch namespace (ws_acme:dispatch:job:1) so everything a tenant
+// owns shares one leading segment, the same layout the tenant's other
+// Redis keys already use; an empty prefix yields the historical key.
+func (k keys) full(suffix string) string {
+	return k.prefix + base + suffix
+}
 
 // ── Job keys ──
 
-// jobKey returns the key for a job entity: dispatch:job:{id}
-func jobKey(id string) string { return keyPrefix + "job:" + id }
+// job returns the key for a job entity.
+func (k keys) job(id string) string { return k.full("job:" + id) }
+
+// queue returns the Sorted Set key for a queue.
+func (k keys) queue(name string) string { return k.full("queue:" + name) }
 
-// queueKey returns the Sorted Set key for a queue: dispatch:queue:{name}
-func queueKey(name string) string { return keyPrefix + "queue:" + name }
+// jobIDs is the Set tracking all job IDs for enumeration.
+func (k keys) jobIDs() string { return k.full("job_ids") }
 
-// jobIDsKey is the Set tracking all job IDs for enumeration.
-const jobIDsKey = keyPrefix + "job_ids"
+// wakeChannel is the pub/sub channel that announces newly enqueued jobs.
+func (k keys) wakeChannel() string { return k.full("jobs:wake") }
 
 // ── Workflow keys ──
 
-// runKey returns the key for a workflow run entity: dispatch:run:{id}
-func runKey(id string) string { return keyPrefix + "run:" + id }
+// run returns the key for a workflow run entity.
+func (k keys) run(id string) string { return k.full("run:" + id) }
 
-// runIDsKey is the Set tracking all run IDs for enumeration.
-const runIDsKey = keyPrefix + "run_ids"
+// runIDs is the Set tracking all run IDs for enumeration.
+func (k keys) runIDs() string { return k.full("run_ids") }
 
-// checkpointKey returns the key for a checkpoint: dispatch:checkpoint:{runID}:{step}
-func checkpointKey(runID, step string) string {
-	return fmt.Sprintf("%scheckpoint:%s:%s", keyPrefix, runID, step)
+// checkpoint returns the key for a checkpoint.
+func (k keys) checkpoint(runID, step string) string {
+	return k.full(fmt.Sprintf("checkpoint:%s:%s", runID, step))
 }
 
-// checkpointIndexKey returns the Set key tracking checkpoints for a run.
-func checkpointIndexKey(runID string) string {
-	return keyPrefix + "checkpoint_idx:" + runID
+// checkpointIndex returns the Set key tracking checkpoints for a run.
+func (k keys) checkpointIndex(runID string) string {
+	return k.full("checkpoint_idx:" + runID)
 }
 
 // ── Cron keys ──
 
-// cronKey returns the key for a cron entry entity: dispatch:cron:{id}
-func cronKey(id string) string { return keyPrefix + "cron:" + id }
+// cron returns the key for a cron entry entity.
+func (k keys) cron(id string) string { return k.full("cron:" + id) }
 
-// cronIDsKey is the Set tracking all cron IDs for enumeration.
-const cronIDsKey = keyPrefix + "cron_ids"
+// cronIDs is the Set tracking all cron IDs for enumeration.
+func (k keys) cronIDs() string { return k.full("cron_ids") }
 
-// cronNamesKey maps cron names to IDs for duplicate detection.
-const cronNamesKey = keyPrefix + "cron_names"
+// cronNames maps cron names to IDs for duplicate detection.
+func (k keys) cronNames() string { return k.full("cron_names") }
 
 // ── DLQ keys ──
 
-// dlqKey returns the key for a DLQ entry entity: dispatch:dlq:{id}
-func dlqKey(id string) string { return keyPrefix + "dlq:" + id }
+// dlq returns the key for a DLQ entry entity.
+func (k keys) dlq(id string) string { return k.full("dlq:" + id) }
 
-// dlqIDsKey is the Set tracking all DLQ entry IDs for enumeration.
-const dlqIDsKey = keyPrefix + "dlq_ids"
+// dlqIDs is the Set tracking all DLQ entry IDs for enumeration.
+func (k keys) dlqIDs() string { return k.full("dlq_ids") }
 
 // ── Event keys ──
 
-// eventKey returns the key for an event entity: dispatch:event:{id}
-func eventKey(id string) string { return keyPrefix + "event:" + id }
+// event returns the key for an event entity.
+func (k keys) event(id string) string { return k.full("event:" + id) }
 
-// eventStreamKey returns the Stream key for an event name: dispatch:events:{name}
-func eventStreamKey(name string) string { return keyPrefix + "events:" + name }
+// eventStream returns the Stream key for an event name.
+func (k keys) eventStream(name string) string { return k.full("events:" + name) }
 
 // ── Cluster keys ──
 
-// workerKey returns the key for a worker entity: dispatch:worker:{id}
-func workerKey(id string) string { return keyPrefix + "worker:" + id }
+// worker returns the key for a worker entity.
+func (k keys) worker(id string) string { return k.full("worker:" + id) }
 
-// workerIDsKey is the Set tracking all worker IDs for enumeration.
-const workerIDsKey = keyPrefix + "worker_ids"
+// workerIDs is the Set tracking all worker IDs for enumeration.
+func (k keys) workerIDs() string { return k.full("worker_ids") }
 
-// leaderKey stores the current leader worker ID.
-const leaderKey = keyPrefix + "leader"
+// leader stores the current leader worker ID.
+func (k keys) leader() string { return k.full("leader") }
 
 // ── Artifact keys ──
 
-// artifactKey returns the key for an artifact entity: dispatch:artifact:{id}
-func artifactKey(id string) string { return keyPrefix + "artifact:" + id }
+// artifact returns the key for an artifact entity.
+func (k keys) artifact(id string) string { return k.full("artifact:" + id) }
 
-// artifactIDsKey is the Set tracking all artifact IDs for enumeration.
-const artifactIDsKey = keyPrefix + "artifact_ids"
+// artifactIDs is the Set tracking all artifact IDs for enumeration.
+func (k keys) artifactIDs() string { return k.full("artifact_ids") }
 
-// artifactKeyGuard maps live storage coordinates to an artifact ID. It is
+// artifactGuard maps live storage coordinates to an artifact ID. It is
 // claimed with SETNX so concurrent creates at the same coordinates resolve
 // to one winner, and released on soft-delete so a purged key is reusable.
-func artifactKeyGuard(backend, bucket, key string) string {
-	return fmt.Sprintf("%sartifact_key:%s:%s:%s", keyPrefix, backend, bucket, key)
+func (k keys) artifactGuard(backend, bucket, key string) string {
+	return k.full(fmt.Sprintf("artifact_key:%s:%s:%s", backend, bucket, key))
 }
 
-// artifactEphemeralKey is the Sorted Set of ephemeral artifact IDs scored
+// artifactEphemeral is the Sorted Set of ephemeral artifact IDs scored
 // by creation time. Durable artifacts are never members, which is this
 // backend's form of the SQL "lifecycle = 'ephemeral'" literal.
-const artifactEphemeralKey = keyPrefix + "artifact_ephemeral"
+func (k keys) artifactEphemeral() string { return k.full("artifact_ephemeral") }
 
-// artifactDeletedKey is the Sorted Set of soft-deleted artifact IDs scored
+// artifactDeleted is the Sorted Set of soft-deleted artifact IDs scored
 // by deletion time, driving the purge pass.
-const artifactDeletedKey = keyPrefix + "artifact_deleted"
+func (k keys) artifactDeleted() string { return k.full("artifact_deleted") }
 
-// artifactLinksKey is the Set of link members pointing at an artifact.
-func artifactLinksKey(artifactID string) string {
-	return keyPrefix + "artifact_links:" + artifactID
+// artifactLinks is the Set of link members pointing at an artifact.
+func (k keys) artifactLinks(artifactID string) string {
+	return k.full("artifact_links:" + artifactID)
 }
 
-// ownerLinksKey is the Hash of an owner's artifact links, keyed by
+// ownerLinks is the Hash of an owner's artifact links, keyed by
 // "name\x00attempt".
-func ownerLinksKey(kind, ownerID string) string {
-	return fmt.Sprintf("%sartifact_owner_links:%s:%s", keyPrefix, kind, ownerID)
+func (k keys) ownerLinks(kind, ownerID string) string {
+	return k.full(fmt.Sprintf("artifact_owner_links:%s:%s", kind, ownerID))
 }
diff --git a/store/redis/keys_internal_test.go b/store/redis/keys_internal_test.go
new file mode 100644
index 0000000..890240e
--- /dev/null
+++ b/store/redis/keys_internal_test.go
@@ -0,0 +1,23 @@
+package redis
+
+import "testing"
+
+func TestKeys_full(t *testing.T) {
+	tests := []struct {
+		name   string
+		prefix string
+		suffix string
+		want   string
+	}{
+		{name: "no prefix keeps the historical key", prefix: "", suffix: "job:1", want: "dispatch:job:1"},
+		{name: "tenant prefix wraps the namespace", prefix: "ws_acme:", suffix: "job:1", want: "ws_acme:dispatch:job:1"},
+		{name: "wake channel follows the same rule", prefix: "ws_acme:", suffix: "jobs:wake", want: "ws_acme:dispatch:jobs:wake"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := newKeys(tt.prefix).full(tt.suffix); got != tt.want {
+				t.Fatalf("full(%q) with prefix %q = %q, want %q", tt.suffix, tt.prefix, got, tt.want)
+			}
+		})
+	}
+}
diff --git a/store/redis/lease.go b/store/redis/lease.go
index 762237e..d0fee6b 100644
--- a/store/redis/lease.go
+++ b/store/redis/lease.go
@@ -183,7 +183,7 @@ func (s *Store) RenewLease(
 	epoch int,
 	leaseUntil time.Time,
 ) error {
-	key := jobKey(jobID.String())
+	key := s.keys.job(jobID.String())
 
 	var e jobEntity
 	if getErr := s.getEntity(ctx, key, &e); getErr != nil {
@@ -270,7 +270,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job
 
 	t := now()
 
-	ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: reclaim smembers: %w", err)
 	}
@@ -282,7 +282,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job
 		}
 
 		var e jobEntity
-		if getErr := s.getEntity(ctx, jobKey(jID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.job(jID), &e); getErr != nil {
 			continue // gone by the time we looked
 		}
 		if job.State(e.State) != job.StateRunning {
@@ -329,7 +329,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job
 		// even for a job that was enqueued straight into running (as the
 		// conformance suite's RunningJob helper does) and was therefore
 		// never popped in the first place.
-		zErr := s.rdb.ZAdd(ctx, queueKey(after.Queue),
+		zErr := s.rdb.ZAdd(ctx, s.keys.queue(after.Queue),
 			goredis.Z{Score: jobScore(after.Priority, after.RunAt), Member: jID}).Err()
 		if zErr != nil {
 			return nil, fmt.Errorf("dispatch/redis: reclaim requeue: %w", zErr)
@@ -361,7 +361,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job
 // knows, because it wrote it.
 func (s *Store) claimExpired(ctx context.Context, jID string, epoch int, blob []byte) (bool, error) {
 	res, err := reclaimScript.Run(ctx, s.rdb,
-		[]string{jobKey(jID)},
+		[]string{s.keys.job(jID)},
 		epoch,
 		blob,
 	).Int64()
@@ -404,7 +404,7 @@ func (s *Store) claimExpired(ctx context.Context, jID string, epoch int, blob []
 // which never need the index touched at all, but that is a property of
 // today's callers, not a license for this method to assume it.
 func (s *Store) UpdateLeasedJob(ctx context.Context, j *job.Job, workerID id.WorkerID, epoch int) error {
-	key := jobKey(j.ID.String())
+	key := s.keys.job(j.ID.String())
 
 	var cur jobEntity
 	if getErr := s.getEntity(ctx, key, &cur); getErr != nil {
@@ -444,7 +444,7 @@ func (s *Store) UpdateLeasedJob(ctx context.Context, j *job.Job, workerID id.Wor
 	// pointing at a running entity is inert until dequeue's own state
 	// check discards it.
 	jID := j.ID.String()
-	qk := queueKey(next.Queue)
+	qk := s.keys.queue(next.Queue)
 	runnable := job.State(next.State) == job.StatePending || job.State(next.State) == job.StateRetrying
 
 	if runnable {
diff --git a/store/redis/prefix_test.go b/store/redis/prefix_test.go
new file mode 100644
index 0000000..5906da2
--- /dev/null
+++ b/store/redis/prefix_test.go
@@ -0,0 +1,165 @@
+//go:build integration
+
+package redis_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/xraph/dispatch"
+	"github.com/xraph/dispatch/cluster"
+	"github.com/xraph/dispatch/cron"
+	"github.com/xraph/dispatch/id"
+	"github.com/xraph/dispatch/job"
+	redisstore "github.com/xraph/dispatch/store/redis"
+)
+
+// Two tenants sharing one Redis must never observe each other's dispatch
+// state. Each assertion below fails on an unprefixed store because every
+// key would collapse onto the same "dispatch:" namespace.
+
+func TestStore_KeyPrefix_jobsAreIsolated(t *testing.T) {
+	ctx := context.Background()
+	kvStore := setupTestKV(t)
+	a := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_a:"))
+	b := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_b:"))
+
+	j := &job.Job{
+		Entity:     dispatch.NewEntity(),
+		ID:         id.NewJobID(),
+		Name:       "tenant-a-only",
+		Queue:      "default",
+		Payload:    []byte(`{}`),
+		State:      job.StatePending,
+		MaxRetries: 3,
+		RunAt:      time.Now().UTC(),
+	}
+	if err := a.EnqueueJob(ctx, j); err != nil {
+		t.Fatalf("enqueue on a: %v", err)
+	}
+
+	if _, err := b.GetJob(ctx, j.ID); err == nil {
+		t.Fatalf("tenant b read tenant a's job %s by id", j.ID)
+	}
+
+	got, err := b.DequeueJobs(ctx, job.DequeueOpts{Queues: []string{"default"}, Limit: 10})
+	if err != nil {
+		t.Fatalf("dequeue on b: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("tenant b dequeued %d of tenant a's jobs", len(got))
+	}
+
+	n, err := b.CountJobs(ctx, job.CountOpts{State: job.StatePending})
+	if err != nil {
+		t.Fatalf("count on b: %v", err)
+	}
+	if n != 0 {
+		t.Fatalf("tenant b counted %d pending jobs, want 0", n)
+	}
+
+	got, err = a.DequeueJobs(ctx, job.DequeueOpts{Queues: []string{"default"}, Limit: 10})
+	if err != nil {
+		t.Fatalf("dequeue on a: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("tenant a dequeued %d jobs, want its own 1", len(got))
+	}
+}
+
+func TestStore_KeyPrefix_cronNamesAreIsolated(t *testing.T) {
+	ctx := context.Background()
+	kvStore := setupTestKV(t)
+	a := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_a:"))
+	b := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_b:"))
+
+	entry := func() *cron.Entry {
+		next := time.Now().Add(time.Hour).UTC()
+		return &cron.Entry{
+			Entity:    dispatch.NewEntity(),
+			ID:        id.NewCronID(),
+			Name:      "nightly",
+			Schedule:  "0 0 * * *",
+			JobName:   "rescore",
+			Payload:   []byte(`{}`),
+			Enabled:   true,
+			NextRunAt: &next,
+		}
+	}
+	if err := a.RegisterCron(ctx, entry()); err != nil {
+		t.Fatalf("register on a: %v", err)
+	}
+	// Same cron name in another tenant is a different cron, not a duplicate.
+	if err := b.RegisterCron(ctx, entry()); err != nil {
+		t.Fatalf("register on b collided with tenant a: %v", err)
+	}
+}
+
+func TestStore_KeyPrefix_leadershipIsIsolated(t *testing.T) {
+	ctx := context.Background()
+	kvStore := setupTestKV(t)
+	a := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_a:"))
+	b := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_b:"))
+
+	leaderA := registerWorker(t, a, "leader-a")
+	ok, err := a.AcquireLeadership(ctx, leaderA, time.Minute)
+	if err != nil {
+		t.Fatalf("acquire on a: %v", err)
+	}
+	if !ok {
+		t.Fatal("tenant a could not take leadership of an empty cluster")
+	}
+
+	leaderB := registerWorker(t, b, "leader-b")
+	ok, err = b.AcquireLeadership(ctx, leaderB, time.Minute)
+	if err != nil {
+		t.Fatalf("acquire on b: %v", err)
+	}
+	if !ok {
+		t.Fatal("tenant b was blocked by tenant a's leader")
+	}
+}
+
+// registerWorker enrols a worker on the store, since leadership can only be
+// taken by a registered worker.
+func registerWorker(t *testing.T, s *redisstore.Store, hostname string) id.WorkerID {
+	t.Helper()
+	w := &cluster.Worker{
+		ID:          id.NewWorkerID(),
+		Hostname:    hostname,
+		Queues:      []string{"default"},
+		Concurrency: 1,
+		State:       cluster.WorkerActive,
+		LastSeen:    time.Now().UTC(),
+		CreatedAt:   time.Now().UTC(),
+	}
+	if err := s.RegisterWorker(context.Background(), w); err != nil {
+		t.Fatalf("register worker %s: %v", hostname, err)
+	}
+	return w.ID
+}
+
+func TestStore_KeyPrefix_emptyPrefixKeepsLegacyKeys(t *testing.T) {
+	ctx := context.Background()
+	kvStore := setupTestKV(t)
+	legacy := redisstore.New(kvStore)
+	explicit := redisstore.New(kvStore, redisstore.WithKeyPrefix(""))
+
+	j := &job.Job{
+		Entity:     dispatch.NewEntity(),
+		ID:         id.NewJobID(),
+		Name:       "shared",
+		Queue:      "default",
+		Payload:    []byte(`{}`),
+		State:      job.StatePending,
+		MaxRetries: 3,
+		RunAt:      time.Now().UTC(),
+	}
+	if err := legacy.EnqueueJob(ctx, j); err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	if _, err := explicit.GetJob(ctx, j.ID); err != nil {
+		t.Fatalf("empty prefix must address the same keys as no prefix: %v", err)
+	}
+}
diff --git a/store/redis/store.go b/store/redis/store.go
index 5e21227..6b59f7f 100644
--- a/store/redis/store.go
+++ b/store/redis/store.go
@@ -43,11 +43,20 @@ func WithLogger(l log.Logger) Option {
 	return func(s *Store) { s.logger = l }
 }
 
+// WithKeyPrefix namespaces every key and channel this store touches so
+// several dispatch instances can share one Redis without seeing each
+// other's jobs, cron locks or leadership. Pass the tenant's own prefix
+// (for example "ws_acme:"); the empty string keeps the historical keys.
+func WithKeyPrefix(prefix string) Option {
+	return func(s *Store) { s.keys = newKeys(prefix) }
+}
+
 // Store implements the composite store.Store interface backed by Redis
 // via Grove KV.
 type Store struct {
 	kv     *kv.Store
 	rdb    goredis.UniversalClient
+	keys   keys
 	logger log.Logger
 }
 
@@ -57,6 +66,7 @@ func New(store *kv.Store, opts ...Option) *Store {
 	s := &Store{
 		kv:     store,
 		rdb:    redisdriver.UnwrapClient(store),
+		keys:   newKeys(""),
 		logger: log.NewNoopLogger(),
 	}
 	for _, o := range opts {
@@ -68,6 +78,10 @@ func New(store *kv.Store, opts ...Option) *Store {
 // KV returns the underlying KV store.
 func (s *Store) KV() *kv.Store { return s.kv }
 
+// KeyPrefix returns the tenant prefix every key is written under; empty
+// when the store uses the historical unprefixed keys.
+func (s *Store) KeyPrefix() string { return s.keys.prefix }
+
 // Migrate is a no-op for Redis (schemaless).
 func (s *Store) Migrate(_ context.Context) error { return nil }
 
diff --git a/store/redis/store_test.go b/store/redis/store_test.go
index 274809c..3952a4d 100644
--- a/store/redis/store_test.go
+++ b/store/redis/store_test.go
@@ -25,8 +25,10 @@ import (
 	"github.com/xraph/dispatch/workflow"
 )
 
-// setupTestStore creates a Redis container and returns a connected Redis Store.
-func setupTestStore(t *testing.T) *redisstore.Store {
+// setupTestKV starts a Redis container and returns a flushed KV store on
+// it. Tests that need more than one dispatch store on the same Redis (the
+// key-prefix isolation cases) build their stores from this directly.
+func setupTestKV(t *testing.T) *kv.Store {
 	t.Helper()
 
 	ctx := context.Background()
@@ -61,15 +63,19 @@ func setupTestStore(t *testing.T) *redisstore.Store {
 		_ = kvStore.Close()
 	})
 
-	store := redisstore.New(kvStore)
-
 	// FlushDB to start clean.
 	client := redisdriver.UnwrapClient(kvStore)
 	if flushErr := client.FlushDB(ctx).Err(); flushErr != nil {
 		t.Fatalf("flush: %v", flushErr)
 	}
 
-	return store
+	return kvStore
+}
+
+// setupTestStore creates a Redis container and returns a connected Redis Store.
+func setupTestStore(t *testing.T) *redisstore.Store {
+	t.Helper()
+	return redisstore.New(setupTestKV(t))
 }
 
 // ──────────────────────────────────────────────────
diff --git a/store/redis/wake.go b/store/redis/wake.go
index 0fe8789..70b8618 100644
--- a/store/redis/wake.go
+++ b/store/redis/wake.go
@@ -7,10 +7,6 @@ import (
 	dispatchstore "github.com/xraph/dispatch/store"
 )
 
-// wakeChannel is the pub/sub channel used to signal that new jobs were
-// enqueued. EnqueueJob publishes to it; StartWakeListener subscribes.
-const wakeChannel = "dispatch:jobs:wake"
-
 var _ dispatchstore.WakeNotifier = (*Store)(nil)
 
 // notifyWake signals listening instances that pending jobs exist.
@@ -18,7 +14,7 @@ var _ dispatchstore.WakeNotifier = (*Store)(nil)
 // failed publish only costs poll latency and is not worth failing the
 // enqueue over.
 func (s *Store) notifyWake(ctx context.Context) {
-	_ = s.rdb.Publish(ctx, wakeChannel, "").Err() //nolint:errcheck // best-effort: polling covers missed wakes
+	_ = s.rdb.Publish(ctx, s.keys.wakeChannel(), "").Err() //nolint:errcheck // best-effort: polling covers missed wakes
 }
 
 // StartWakeListener subscribes to the dispatch wake channel and invokes
@@ -30,7 +26,7 @@ func (s *Store) notifyWake(ctx context.Context) {
 func (s *Store) StartWakeListener(ctx context.Context, wake func()) (func(), error) {
 	ctx, cancel := context.WithCancel(ctx)
 
-	sub := s.rdb.Subscribe(ctx, wakeChannel)
+	sub := s.rdb.Subscribe(ctx, s.keys.wakeChannel())
 	// Confirm the subscription is established so callers know push is
 	// live before relying on it.
 	if _, err := sub.Receive(ctx); err != nil {
diff --git a/store/redis/workflow.go b/store/redis/workflow.go
index 407cf1c..e4a8396 100644
--- a/store/redis/workflow.go
+++ b/store/redis/workflow.go
@@ -79,7 +79,7 @@ type checkpointEntity struct {
 // CreateRun persists a new workflow run.
 func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error {
 	rID := run.ID.String()
-	key := runKey(rID)
+	key := s.keys.run(rID)
 
 	exists, err := s.entityExists(ctx, key)
 	if err != nil {
@@ -94,7 +94,7 @@ func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error {
 		return fmt.Errorf("dispatch/redis: create run set: %w", err)
 	}
 
-	if err := s.rdb.SAdd(ctx, runIDsKey, rID).Err(); err != nil {
+	if err := s.rdb.SAdd(ctx, s.keys.runIDs(), rID).Err(); err != nil {
 		return fmt.Errorf("dispatch/redis: create run index: %w", err)
 	}
 	return nil
@@ -103,7 +103,7 @@ func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error {
 // GetRun retrieves a workflow run by ID.
 func (s *Store) GetRun(ctx context.Context, runID id.RunID) (*workflow.Run, error) {
 	var e runEntity
-	if err := s.getEntity(ctx, runKey(runID.String()), &e); err != nil {
+	if err := s.getEntity(ctx, s.keys.run(runID.String()), &e); err != nil {
 		if isNotFound(err) {
 			return nil, dispatch.ErrRunNotFound
 		}
@@ -114,7 +114,7 @@ func (s *Store) GetRun(ctx context.Context, runID id.RunID) (*workflow.Run, erro
 
 // UpdateRun persists changes to an existing workflow run.
 func (s *Store) UpdateRun(ctx context.Context, run *workflow.Run) error {
-	key := runKey(run.ID.String())
+	key := s.keys.run(run.ID.String())
 	exists, err := s.entityExists(ctx, key)
 	if err != nil {
 		return fmt.Errorf("dispatch/redis: update run exists: %w", err)
@@ -130,7 +130,7 @@ func (s *Store) UpdateRun(ctx context.Context, run *workflow.Run) error {
 
 // ListRuns returns workflow runs matching the given options.
 func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workflow.Run, error) {
-	ids, err := s.rdb.SMembers(ctx, runIDsKey).Result()
+	ids, err := s.rdb.SMembers(ctx, s.keys.runIDs()).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list runs smembers: %w", err)
 	}
@@ -138,7 +138,7 @@ func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workfl
 	runs := make([]*workflow.Run, 0, len(ids))
 	for _, rID := range ids {
 		var e runEntity
-		if getErr := s.getEntity(ctx, runKey(rID), &e); getErr != nil {
+		if getErr := s.getEntity(ctx, s.keys.run(rID), &e); getErr != nil {
 			continue
 		}
 		if opts.State != "" && workflow.RunState(e.State) != opts.State {
@@ -157,7 +157,7 @@ func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workfl
 // SaveCheckpoint persists checkpoint data for a workflow step.
 func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName string, data []byte) error {
 	rID := runID.String()
-	key := checkpointKey(rID, stepName)
+	key := s.keys.checkpoint(rID, stepName)
 
 	e := &checkpointEntity{
 		ID:        id.NewCheckpointID().String(),
@@ -171,7 +171,7 @@ func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName str
 		return fmt.Errorf("dispatch/redis: save checkpoint: %w", err)
 	}
 
-	if err := s.rdb.SAdd(ctx, checkpointIndexKey(rID), stepName).Err(); err != nil {
+	if err := s.rdb.SAdd(ctx, s.keys.checkpointIndex(rID), stepName).Err(); err != nil {
 		return fmt.Errorf("dispatch/redis: save checkpoint index: %w", err)
 	}
 	return nil
@@ -179,7 +179,7 @@ func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName str
 
 // GetCheckpoint retrieves checkpoint data for a specific workflow step.
 func (s *Store) GetCheckpoint(ctx context.Context, runID id.RunID, stepName string) ([]byte, error) {
-	key := checkpointKey(runID.String(), stepName)
+	key := s.keys.checkpoint(runID.String(), stepName)
 	var e checkpointEntity
 	if err := s.getEntity(ctx, key, &e); err != nil {
 		if isNotFound(err) {
@@ -193,14 +193,14 @@ func (s *Store) GetCheckpoint(ctx context.Context, runID id.RunID, stepName stri
 // ListCheckpoints returns all checkpoints for a workflow run.
 func (s *Store) ListCheckpoints(ctx context.Context, runID id.RunID) ([]*workflow.Checkpoint, error) {
 	rID := runID.String()
-	steps, err := s.rdb.SMembers(ctx, checkpointIndexKey(rID)).Result()
+	steps, err := s.rdb.SMembers(ctx, s.keys.checkpointIndex(rID)).Result()
 	if err != nil {
 		return nil, fmt.Errorf("dispatch/redis: list checkpoints: %w", err)
 	}
 
 	checkpoints := make([]*workflow.Checkpoint, 0, len(steps))
 	for _, step := range steps {
-		key := checkpointKey(rID, step)
+		key := s.keys.checkpoint(rID, step)
 		var e checkpointEntity
 		if getErr := s.getEntity(ctx, key, &e); getErr != nil {
 			continue
@@ -244,7 +244,7 @@ func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afte
 
 	// Get the target checkpoint's time.
 	var target checkpointEntity
-	if err := s.getEntity(ctx, checkpointKey(rID, afterStep), &target); err != nil {
+	if err := s.getEntity(ctx, s.keys.checkpoint(rID, afterStep), &target); err != nil {
 		if isNotFound(err) {
 			return nil // step not found; nothing to delete
 		}
@@ -252,13 +252,13 @@ func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afte
 	}
 
 	// List all step names for this run.
-	steps, err := s.rdb.SMembers(ctx, checkpointIndexKey(rID)).Result()
+	steps, err := s.rdb.SMembers(ctx, s.keys.checkpointIndex(rID)).Result()
 	if err != nil {
 		return fmt.Errorf("dispatch/redis: list checkpoint steps: %w", err)
 	}
 
 	for _, step := range steps {
-		key := checkpointKey(rID, step)
+		key := s.keys.checkpoint(rID, step)
 		var e checkpointEntity
 		if getErr := s.getEntity(ctx, key, &e); getErr != nil {
 			continue
@@ -267,7 +267,7 @@ func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afte
 			if delErr := s.rdb.Del(ctx, key).Err(); delErr != nil {
 				return fmt.Errorf("delete checkpoint %s: %w", key, delErr)
 			}
-			if remErr := s.rdb.SRem(ctx, checkpointIndexKey(rID), step).Err(); remErr != nil {
+			if remErr := s.rdb.SRem(ctx, s.keys.checkpointIndex(rID), step).Err(); remErr != nil {
 				return fmt.Errorf("remove checkpoint index %s: %w", step, remErr)
 			}
 		}