From cd3075e75ba74a25cb7f87051fa778ac46e50c8e Mon Sep 17 00:00:00 2001
From: user <u@example.com>
Date: Thu, 3 Sep 2026 14:58:44 -0400
Subject: [PATCH] [HYPERSHELL-297] Filter gateways by cluster_id for
 managed-cluster pull model

A managed-cluster spoke deploys only the control-plane; it dials the hub
api-server over gRPC, watches for gateways whose cluster_id matches its own
identity, and provisions them locally. Because the event broker fans every
gateway out to every subscriber, scoping a spoke to its own cluster is a
security boundary, so the filter is applied server-side.

api-server:
- Add optional cluster_id to ListGatewaysRequest and WatchGatewaysRequest.
- ListGateways filters via a cluster_id = '...' search when set.
- WatchGateways skips events whose loaded gateway's cluster_id does not match,
  and (when a filter is set) skips deletes it cannot attribute to a cluster.

control-plane:
- Add Config.ClusterID from HYPERSHELL_CLUSTER_ID (empty = handle all gateways).
- Thread it through the gateway watch and its seed lists.
- De-duplicate the two listAllGateways helpers into one taking a clusterID.
- Health and sandbox-count reconcilers filter by cluster_id so a spoke never
  stamps or zeroes a foreign cluster's gateway; the namespace GC stays
  unfiltered on purpose (its live set only ever protects namespaces, which are
  globally unique, so a superset can never cause a wrong reap).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 .../pkg/api/grpc/hypershell/v1/gateways.pb.go | 48 +++++++++++++++---
 .../plugins/gateways/grpc_handler.go          | 39 +++++++++++++++
 .../proto/hypershell/v1/gateways.proto        | 13 ++++-
 .../cmd/hypershell-controller/main.go         | 13 +++--
 .../control-plane/internal/config/config.go   |  9 ++++
 .../internal/reconciler/backfill.go           | 14 ++++--
 .../internal/reconciler/backfill_test.go      | 25 ++++++++--
 .../internal/reconciler/health.go             | 46 ++++-------------
 .../internal/reconciler/health_test.go        | 49 ++++++++++++++++---
 .../internal/reconciler/namespace.go          |  8 ++-
 .../internal/reconciler/reconciler.go         | 14 ++++--
 .../internal/reconciler/reconciler_test.go    |  2 +-
 .../internal/reconciler/sandboxcount.go       | 13 ++++-
 .../internal/reconciler/sandboxcount_test.go  |  2 +-
 .../internal/watcher/seed_test.go             | 46 ++++++++++++++---
 .../control-plane/internal/watcher/watcher.go | 35 +++++++++----
 16 files changed, 286 insertions(+), 90 deletions(-)

diff --git a/components/api-server/pkg/api/grpc/hypershell/v1/gateways.pb.go b/components/api-server/pkg/api/grpc/hypershell/v1/gateways.pb.go
index ee13d468c..8878e8d7f 100644
--- a/components/api-server/pkg/api/grpc/hypershell/v1/gateways.pb.go
+++ b/components/api-server/pkg/api/grpc/hypershell/v1/gateways.pb.go
@@ -978,9 +978,14 @@ func (x *DeleteGatewayRequest) GetId() string {
 }
 
 type ListGatewaysRequest struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	Page          int32                  `protobuf:"varint,1,opt,name=page,proto3" json:"page,omitempty"`
-	Size          int32                  `protobuf:"varint,2,opt,name=size,proto3" json:"size,omitempty"`
+	state protoimpl.MessageState `protogen:"open.v1"`
+	Page  int32                  `protobuf:"varint,1,opt,name=page,proto3" json:"page,omitempty"`
+	Size  int32                  `protobuf:"varint,2,opt,name=size,proto3" json:"size,omitempty"`
+	// cluster_id, when set, restricts results to gateways assigned to that
+	// managed cluster. A control-plane agent sets it to its own cluster identity
+	// so it only ever lists its cluster's gateways (managed-cluster pull model).
+	// Unset preserves the prior behaviour of listing every gateway.
+	ClusterId     *string `protobuf:"bytes,3,opt,name=cluster_id,json=clusterId,proto3,oneof" json:"cluster_id,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1029,6 +1034,13 @@ func (x *ListGatewaysRequest) GetSize() int32 {
 	return 0
 }
 
+func (x *ListGatewaysRequest) GetClusterId() string {
+	if x != nil && x.ClusterId != nil {
+		return *x.ClusterId
+	}
+	return ""
+}
+
 type ListGatewaysResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Items         []*Gateway             `protobuf:"bytes,1,rep,name=items,proto3" json:"items,omitempty"`
@@ -1118,7 +1130,12 @@ func (*DeleteGatewayResponse) Descriptor() ([]byte, []int) {
 }
 
 type WatchGatewaysRequest struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// cluster_id, when set, restricts the stream to gateways assigned to that
+	// managed cluster (see ListGatewaysRequest.cluster_id). Because the event
+	// broker fans every gateway out to every subscriber, this filter is the
+	// security boundary that keeps a spoke's stream scoped to its own cluster.
+	ClusterId     *string `protobuf:"bytes,1,opt,name=cluster_id,json=clusterId,proto3,oneof" json:"cluster_id,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1153,6 +1170,13 @@ func (*WatchGatewaysRequest) Descriptor() ([]byte, []int) {
 	return file_hypershell_v1_gateways_proto_rawDescGZIP(), []int{15}
 }
 
+func (x *WatchGatewaysRequest) GetClusterId() string {
+	if x != nil && x.ClusterId != nil {
+		return *x.ClusterId
+	}
+	return ""
+}
+
 type WatchGatewaysResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Type          EventType              `protobuf:"varint,1,opt,name=type,proto3,enum=hypershell.v1.EventType" json:"type,omitempty"`
@@ -1346,15 +1370,21 @@ const file_hypershell_v1_gateways_proto_rawDesc = "" +
 	"\x1dSetActiveSandboxCountResponse\x120\n" +
 	"\x14active_sandbox_count\x18\x01 \x01(\x05R\x12activeSandboxCount\"&\n" +
 	"\x14DeleteGatewayRequest\x12\x0e\n" +
-	"\x02id\x18\x01 \x01(\tR\x02id\"=\n" +
+	"\x02id\x18\x01 \x01(\tR\x02id\"p\n" +
 	"\x13ListGatewaysRequest\x12\x12\n" +
 	"\x04page\x18\x01 \x01(\x05R\x04page\x12\x12\n" +
-	"\x04size\x18\x02 \x01(\x05R\x04size\"y\n" +
+	"\x04size\x18\x02 \x01(\x05R\x04size\x12\"\n" +
+	"\n" +
+	"cluster_id\x18\x03 \x01(\tH\x00R\tclusterId\x88\x01\x01B\r\n" +
+	"\v_cluster_id\"y\n" +
 	"\x14ListGatewaysResponse\x12,\n" +
 	"\x05items\x18\x01 \x03(\v2\x16.hypershell.v1.GatewayR\x05items\x123\n" +
 	"\bmetadata\x18\x02 \x01(\v2\x17.hypershell.v1.ListMetaR\bmetadata\"\x17\n" +
-	"\x15DeleteGatewayResponse\"\x16\n" +
-	"\x14WatchGatewaysRequest\"\x98\x01\n" +
+	"\x15DeleteGatewayResponse\"I\n" +
+	"\x14WatchGatewaysRequest\x12\"\n" +
+	"\n" +
+	"cluster_id\x18\x01 \x01(\tH\x00R\tclusterId\x88\x01\x01B\r\n" +
+	"\v_cluster_id\"\x98\x01\n" +
 	"\x15WatchGatewaysResponse\x12,\n" +
 	"\x04type\x18\x01 \x01(\x0e2\x18.hypershell.v1.EventTypeR\x04type\x120\n" +
 	"\agateway\x18\x02 \x01(\v2\x16.hypershell.v1.GatewayR\agateway\x12\x1f\n" +
@@ -1447,6 +1477,8 @@ func file_hypershell_v1_gateways_proto_init() {
 	file_hypershell_v1_gateways_proto_msgTypes[0].OneofWrappers = []any{}
 	file_hypershell_v1_gateways_proto_msgTypes[1].OneofWrappers = []any{}
 	file_hypershell_v1_gateways_proto_msgTypes[5].OneofWrappers = []any{}
+	file_hypershell_v1_gateways_proto_msgTypes[12].OneofWrappers = []any{}
+	file_hypershell_v1_gateways_proto_msgTypes[15].OneofWrappers = []any{}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
diff --git a/components/api-server/plugins/gateways/grpc_handler.go b/components/api-server/plugins/gateways/grpc_handler.go
index 5cd048472..977555cb0 100644
--- a/components/api-server/plugins/gateways/grpc_handler.go
+++ b/components/api-server/plugins/gateways/grpc_handler.go
@@ -3,6 +3,7 @@ package gateways
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 
 	"github.com/golang/glog"
 	"google.golang.org/grpc"
@@ -234,6 +235,22 @@ func (h *gatewayGRPCHandler) ListGateways(ctx context.Context, req *pb.ListGatew
 		Size: int64(size),
 	}
 
+	// A managed-cluster control-plane sets cluster_id to its own identity so it
+	// only ever lists the gateways it is responsible for provisioning. Filtering
+	// server-side keeps foreign gateways off the wire entirely (see WatchGateways,
+	// which applies the same cooperative scoping to the event stream).
+	//
+	// Validate before interpolating into the search DSL: cluster_id is
+	// request-supplied, so an unvalidated value (e.g. one containing a quote)
+	// could break the filter parse or broaden it. This applies the same field
+	// contract the create/update paths enforce on cluster_id.
+	if clusterID := req.GetClusterId(); clusterID != "" {
+		if err := grpcutil.ValidateStringField("cluster_id", clusterID, false); err != nil {
+			return nil, err
+		}
+		listArgs.Search = fmt.Sprintf("cluster_id = '%s'", clusterID)
+	}
+
 	var gateways []Gateway
 	paging, svcErr := h.generic.List(ctx, "id", listArgs, &gateways)
 	if svcErr != nil {
@@ -257,6 +274,15 @@ func (h *gatewayGRPCHandler) WatchGateways(req *pb.WatchGatewaysRequest, stream
 		return status.Error(codes.Unavailable, "event broker not available")
 	}
 
+	// clusterFilter, when set, scopes this stream to a single managed cluster.
+	// The broker fans EVERY gateway out to EVERY subscriber, so without this a
+	// spoke would receive (and could act on) other clusters' gateways. This is
+	// cooperative scoping, not an enforced trust boundary: the server does not yet
+	// authenticate that the caller owns the claimed cluster_id (no per-caller
+	// RBAC), so any control-plane could pass any cluster_id. Enforcement is pending
+	// the managed-cluster caller-identity binding (remote gRPC TLS+OIDC dial).
+	clusterFilter := req.GetClusterId()
+
 	ctx := stream.Context()
 	sub, err := broker.Subscribe(ctx)
 	if err != nil {
@@ -296,7 +322,17 @@ func (h *gatewayGRPCHandler) WatchGateways(req *pb.WatchGatewaysRequest, stream
 				gateway, svcErr := h.service.GetUnscoped(ctx, evt.SourceID)
 				if svcErr != nil {
 					glog.Warningf("WatchGateways: failed to load soft-deleted gateway %s: %v", evt.SourceID, svcErr)
+					// When a cluster filter is set we cannot attribute an
+					// unloadable delete to a cluster, so we must not leak it to a
+					// scoped subscriber. Skip it; the spoke's namespace GC still
+					// reaps the orphaned namespace.
+					if clusterFilter != "" {
+						continue
+					}
 				} else {
+					if clusterFilter != "" && gateway.ClusterId != clusterFilter {
+						continue
+					}
 					watchEvent.Gateway = gatewayToProto(gateway)
 				}
 			} else {
@@ -305,6 +341,9 @@ func (h *gatewayGRPCHandler) WatchGateways(req *pb.WatchGatewaysRequest, stream
 					glog.Warningf("WatchGateways: failed to load gateway %s: %v", evt.SourceID, svcErr)
 					continue
 				}
+				if clusterFilter != "" && gateway.ClusterId != clusterFilter {
+					continue
+				}
 				watchEvent.Gateway = gatewayToProto(gateway)
 			}
 
diff --git a/components/api-server/proto/hypershell/v1/gateways.proto b/components/api-server/proto/hypershell/v1/gateways.proto
index b515414f0..aa0d56403 100644
--- a/components/api-server/proto/hypershell/v1/gateways.proto
+++ b/components/api-server/proto/hypershell/v1/gateways.proto
@@ -127,6 +127,11 @@ message DeleteGatewayRequest {
 message ListGatewaysRequest {
   int32 page = 1;
   int32 size = 2;
+  // cluster_id, when set, restricts results to gateways assigned to that
+  // managed cluster. A control-plane agent sets it to its own cluster identity
+  // so it only ever lists its cluster's gateways (managed-cluster pull model).
+  // Unset preserves the prior behaviour of listing every gateway.
+  optional string cluster_id = 3;
 }
 
 message ListGatewaysResponse {
@@ -136,7 +141,13 @@ message ListGatewaysResponse {
 
 message DeleteGatewayResponse {}
 
-message WatchGatewaysRequest {}
+message WatchGatewaysRequest {
+  // cluster_id, when set, restricts the stream to gateways assigned to that
+  // managed cluster (see ListGatewaysRequest.cluster_id). Because the event
+  // broker fans every gateway out to every subscriber, this filter is the
+  // security boundary that keeps a spoke's stream scoped to its own cluster.
+  optional string cluster_id = 1;
+}
 
 message WatchGatewaysResponse {
   EventType type = 1;
diff --git a/components/control-plane/cmd/hypershell-controller/main.go b/components/control-plane/cmd/hypershell-controller/main.go
index ecb4e69e9..b385e8da4 100644
--- a/components/control-plane/cmd/hypershell-controller/main.go
+++ b/components/control-plane/cmd/hypershell-controller/main.go
@@ -52,6 +52,11 @@ func main() {
 
 	log.Printf("INFO hypershell-controller starting")
 	log.Printf("INFO grpc=%s api=%s namespace=%s database_provider=%s", cfg.GRPCServerAddr, cfg.APIServerURL, cfg.Namespace, cfg.DatabaseProvider)
+	if cfg.ClusterID != "" {
+		log.Printf("INFO managed-cluster mode: scoping gateway watch/seed/health to cluster_id=%s", cfg.ClusterID)
+	} else {
+		log.Printf("INFO single-cluster mode: handling all gateways (no cluster_id filter)")
+	}
 
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer cancel()
@@ -281,7 +286,7 @@ func main() {
 		return watcher.WatchGatewayReleases(ctx, conn, releaseReconciler)
 	})
 	supervise("Gateway watch", func(ctx context.Context) error {
-		return watcher.WatchGateways(ctx, conn, gatewayReconciler)
+		return watcher.WatchGateways(ctx, conn, gatewayReconciler, cfg.ClusterID)
 	})
 	supervise("GatewayNetwork watch", func(ctx context.Context) error {
 		return watcher.WatchGatewayNetworks(ctx, conn, networkReconciler)
@@ -298,7 +303,7 @@ func main() {
 	// status synchronized with observed workload health (Running <-> Degraded).
 	// It requires an in-cluster Kubernetes client to observe Deployments.
 	if clientset != nil {
-		healthReconciler := reconciler.NewGatewayHealthReconciler(clientset, dynamicClient, conn, exposurePort, keycloakConfig)
+		healthReconciler := reconciler.NewGatewayHealthReconciler(clientset, dynamicClient, conn, exposurePort, keycloakConfig, cfg.ClusterID)
 		supervise("gateway health reconciler", healthReconciler.Run)
 		log.Printf("INFO gateway health reconciler launched")
 	} else {
@@ -310,7 +315,7 @@ func main() {
 	// its cache), instead of a repeated full-namespace pod LIST. It requires an
 	// in-cluster Kubernetes client to watch pods.
 	if clientset != nil {
-		sandboxCountReconciler := reconciler.NewSandboxCountReconciler(clientset, conn, 0)
+		sandboxCountReconciler := reconciler.NewSandboxCountReconciler(clientset, conn, 0, cfg.ClusterID)
 		supervise("sandbox count reconciler", sandboxCountReconciler.Run)
 		log.Printf("INFO sandbox count reconciler launched")
 	} else {
@@ -330,7 +335,7 @@ func main() {
 		// synchronously so the first sweep sees the freshly-labeled namespaces; it
 		// is best-effort and never blocks startup on failure.
 		backfillCtx, cancelBackfill := context.WithTimeout(ctx, instanceLabelBackfillTimeout)
-		reconciler.RunInstanceLabelBackfill(backfillCtx, clientset, conn, cfg.Namespace)
+		reconciler.RunInstanceLabelBackfill(backfillCtx, clientset, conn, cfg.Namespace, cfg.ClusterID)
 		cancelBackfill()
 
 		gcReconciler := reconciler.NewNamespaceGCReconciler(
diff --git a/components/control-plane/internal/config/config.go b/components/control-plane/internal/config/config.go
index 7868ca769..8e161515e 100644
--- a/components/control-plane/internal/config/config.go
+++ b/components/control-plane/internal/config/config.go
@@ -27,6 +27,14 @@ type Config struct {
 	Namespace      string
 	LogLevel       string
 
+	// ClusterID is this control-plane's managed-cluster identity (a Gateway
+	// cluster_id / KSUID). When set, the control-plane restricts the gateways it
+	// watches, seeds, and health-checks to those whose cluster_id matches, so a
+	// managed-cluster spoke only ever provisions its own gateways (the pull
+	// model). Empty preserves the single-cluster behaviour of handling every
+	// gateway. Sourced from HYPERSHELL_CLUSTER_ID.
+	ClusterID string
+
 	// ServiceAccountProvisionerAddress is the in-cluster bind address for the
 	// internal service-account provisioner gRPC server. A NetworkPolicy restricts
 	// the port to the API server pod, so the channel is plaintext (no mTLS).
@@ -63,6 +71,7 @@ func Load() (*Config, error) {
 		APIServerURL:                     getEnv("HYPERSHELL_API_SERVER_URL", "http://localhost:8000"),
 		Namespace:                        getEnv("HYPERSHELL_NAMESPACE", "hypershell"),
 		LogLevel:                         strings.ToLower(getEnv("HYPERSHELL_LOG_LEVEL", "info")),
+		ClusterID:                        getEnv("HYPERSHELL_CLUSTER_ID", ""),
 		ServiceAccountProvisionerAddress: getEnv("HYPERSHELL_SERVICE_ACCOUNT_PROVISIONER_BIND_ADDRESS", ""),
 
 		NamespaceGCEnabled:     getEnvBool("GATEWAY_NAMESPACE_GC_ENABLED", true),
diff --git a/components/control-plane/internal/reconciler/backfill.go b/components/control-plane/internal/reconciler/backfill.go
index e82fb1f36..805530b15 100644
--- a/components/control-plane/internal/reconciler/backfill.go
+++ b/components/control-plane/internal/reconciler/backfill.go
@@ -41,11 +41,17 @@ import (
 // two generic management labels, they are indistinguishable from another
 // HyperShell's namespaces on a shared cluster, so claiming them is unsafe. Those
 // require manual cleanup.
-func BackfillInstanceLabels(ctx context.Context, client kubernetes.Interface, gwClient pb.GatewayServiceClient, instance string) (int, error) {
+func BackfillInstanceLabels(ctx context.Context, client kubernetes.Interface, gwClient pb.GatewayServiceClient, instance, clusterID string) (int, error) {
 	if instance == "" {
 		return 0, fmt.Errorf("refusing to backfill instance labels without a control-plane instance identity")
 	}
-	gateways, err := listAllGateways(ctx, gwClient)
+	// Scope the listing to this cluster's gateways (clusterID from
+	// HYPERSHELL_CLUSTER_ID; empty in single-cluster mode lists all). Unlike the
+	// namespace reaper -- which lists unfiltered because a superset only ever
+	// protects namespaces -- backfill WRITES this instance's label, so on a
+	// co-located cluster an unscoped list could stamp another instance's
+	// not-yet-labeled namespace. Filtering keeps a spoke to its own gateways.
+	gateways, err := listAllGateways(ctx, gwClient, clusterID)
 	if err != nil {
 		// A partial inventory would silently skip namespaces that need the label,
 		// leaving them to leak; fail so the caller can log and retry on the next
@@ -86,9 +92,9 @@ func BackfillInstanceLabels(ctx context.Context, client kubernetes.Interface, gw
 // but never aborts controller startup, because the periodic GC sweep still
 // functions for already-labeled namespaces and the backfill is retried on the
 // next restart.
-func RunInstanceLabelBackfill(ctx context.Context, client kubernetes.Interface, conn *grpc.ClientConn, instance string) {
+func RunInstanceLabelBackfill(ctx context.Context, client kubernetes.Interface, conn *grpc.ClientConn, instance, clusterID string) {
 	gwClient := pb.NewGatewayServiceClient(conn)
-	labeled, err := BackfillInstanceLabels(ctx, client, gwClient, instance)
+	labeled, err := BackfillInstanceLabels(ctx, client, gwClient, instance, clusterID)
 	if err != nil {
 		log.Printf("WARN instance-label backfill completed with errors (labeled=%d): %v", labeled, err)
 		return
diff --git a/components/control-plane/internal/reconciler/backfill_test.go b/components/control-plane/internal/reconciler/backfill_test.go
index 64affcafd..f5924d318 100644
--- a/components/control-plane/internal/reconciler/backfill_test.go
+++ b/components/control-plane/internal/reconciler/backfill_test.go
@@ -72,7 +72,7 @@ func TestBackfillInstanceLabels(t *testing.T) {
 			gatewayWithNamespace("id-empty", ""),
 		})
 
-		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell")
+		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", "")
 		if err != nil {
 			t.Fatalf("BackfillInstanceLabels() error = %v", err)
 		}
@@ -104,7 +104,7 @@ func TestBackfillInstanceLabels(t *testing.T) {
 				return nil, fmt.Errorf("boom")
 			},
 		}
-		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell"); err == nil {
+		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", ""); err == nil {
 			t.Fatalf("BackfillInstanceLabels() error = nil, want list failure")
 		}
 		got, err := client.CoreV1().Namespaces().Get(ctx, "openshell-a", metav1.GetOptions{})
@@ -133,7 +133,7 @@ func TestBackfillInstanceLabels(t *testing.T) {
 			gatewayWithNamespace("id-b", "openshell-b"),
 		})
 
-		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell")
+		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", "")
 		if err == nil {
 			t.Fatalf("BackfillInstanceLabels() error = nil, want a collected per-namespace error")
 		}
@@ -152,8 +152,25 @@ func TestBackfillInstanceLabels(t *testing.T) {
 	t.Run("refuses an empty instance identity", func(t *testing.T) {
 		client := fake.NewSimpleClientset()
 		gwClient := singlePageGatewayClient(nil)
-		if _, err := BackfillInstanceLabels(ctx, client, gwClient, ""); err == nil {
+		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "", ""); err == nil {
 			t.Fatalf("BackfillInstanceLabels() error = nil, want empty instance error")
 		}
 	})
+
+	t.Run("scopes the gateway listing to this cluster", func(t *testing.T) {
+		client := fake.NewSimpleClientset()
+		var gotClusterID *string
+		gwClient := &fakeGatewayClient{
+			listFn: func(ctx context.Context, in *pb.ListGatewaysRequest, opts ...grpc.CallOption) (*pb.ListGatewaysResponse, error) {
+				gotClusterID = in.ClusterId
+				return &pb.ListGatewaysResponse{Metadata: &pb.ListMeta{Total: 0}}, nil
+			},
+		}
+		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", "mc1"); err != nil {
+			t.Fatalf("BackfillInstanceLabels() error = %v", err)
+		}
+		if gotClusterID == nil || *gotClusterID != "mc1" {
+			t.Fatalf("cluster_id = %v, want \"mc1\" (backfill must scope to its own cluster)", gotClusterID)
+		}
+	})
 }
diff --git a/components/control-plane/internal/reconciler/health.go b/components/control-plane/internal/reconciler/health.go
index 7de61b5cd..e68a38336 100644
--- a/components/control-plane/internal/reconciler/health.go
+++ b/components/control-plane/internal/reconciler/health.go
@@ -29,10 +29,6 @@ const defaultHealthInterval = 30 * time.Second
 // specs/platform/openshell-gateway-routing.spec.md § Gateway Exposure Configuration.
 const defaultRouteReadyTimeout = 10 * time.Minute
 
-// defaultListGatewaysPageSize is the number of gateways requested per page
-// when retrieving the full gateway fleet for health observation.
-const defaultListGatewaysPageSize = 100
-
 // routeVerifyInterval is the minimum time between residual route/console
 // absence re-checks for a settled (torn-down, addressless) gateway.
 //
@@ -49,9 +45,13 @@ const routeVerifyInterval = 5 * time.Minute
 // moved to Degraded, and a Degraded gateway whose workload and exposure recover
 // is moved back to Running. See openshell-gateway-health.spec.md.
 type GatewayHealthReconciler struct {
-	clientset           *kubernetes.Clientset
-	dynamicClient       dynamic.Interface
-	grpcConn            *grpc.ClientConn
+	clientset     *kubernetes.Clientset
+	dynamicClient dynamic.Interface
+	grpcConn      *grpc.ClientConn
+	// clusterID scopes the health sweep to this managed cluster's gateways. When
+	// non-empty the fleet list is filtered server-side so a spoke never stamps
+	// (Degraded/Running) a gateway owned by another cluster. Empty sweeps all.
+	clusterID           string
 	interval            time.Duration
 	exposure            exposure.Port
 	routeReadyTimeout   time.Duration
@@ -99,7 +99,7 @@ type GatewayHealthReconciler struct {
 	routeVerifiedAt    map[string]time.Time
 }
 
-func NewGatewayHealthReconciler(clientset *kubernetes.Clientset, dynamicClient dynamic.Interface, grpcConn *grpc.ClientConn, exposurePort exposure.Port, keycloakConfig *gateway.KeycloakConfig) *GatewayHealthReconciler {
+func NewGatewayHealthReconciler(clientset *kubernetes.Clientset, dynamicClient dynamic.Interface, grpcConn *grpc.ClientConn, exposurePort exposure.Port, keycloakConfig *gateway.KeycloakConfig, clusterID string) *GatewayHealthReconciler {
 	// Build one long-lived Keycloak client for residual-absence checks so its
 	// token cache survives across ticks (see consoleClientChecker).
 	var consoleClientChecker gateway.ConsoleClientChecker
@@ -120,6 +120,7 @@ func NewGatewayHealthReconciler(clientset *kubernetes.Clientset, dynamicClient d
 		clientset:            clientset,
 		dynamicClient:        dynamicClient,
 		grpcConn:             grpcConn,
+		clusterID:            clusterID,
 		interval:             defaultHealthInterval,
 		exposure:             exposurePort,
 		routeReadyTimeout:    routeReadyTimeout(),
@@ -173,7 +174,7 @@ func (h *GatewayHealthReconciler) reconcileOnce(ctx context.Context) {
 	// Page through the whole fleet: the list endpoint is server-side paginated
 	// (default page size 20), so an unpaged request would only ever refresh the
 	// health of the first page of gateways.
-	gateways, err := h.listAllGateways(ctx, client)
+	gateways, err := listAllGateways(ctx, client, h.clusterID)
 	if err != nil {
 		tickErr = err
 		log.Printf("WARN gateway health: list gateways: %v", err)
@@ -185,33 +186,6 @@ func (h *GatewayHealthReconciler) reconcileOnce(ctx context.Context) {
 	}
 }
 
-// listAllGateways retrieves all gateways from the API server across all pages.
-func (h *GatewayHealthReconciler) listAllGateways(ctx context.Context, client pb.GatewayServiceClient) ([]*pb.Gateway, error) {
-	var all []*pb.Gateway
-	page := int32(1)
-
-	for {
-		resp, err := client.ListGateways(ctx, &pb.ListGatewaysRequest{
-			Page: page,
-			Size: defaultListGatewaysPageSize,
-		})
-		if err != nil {
-			return nil, err
-		}
-
-		items := resp.GetItems()
-		all = append(all, items...)
-
-		meta := resp.GetMetadata()
-		if len(items) == 0 || (meta != nil && int64(len(all)) >= int64(meta.GetTotal())) || len(items) < int(defaultListGatewaysPageSize) {
-			break
-		}
-		page++
-	}
-
-	return all, nil
-}
-
 func (h *GatewayHealthReconciler) reconcileGatewayHealth(ctx context.Context, client pb.GatewayServiceClient, gw *pb.Gateway) {
 	gatewayID := gw.GetMetadata().GetId()
 	if gatewayID == "" {
diff --git a/components/control-plane/internal/reconciler/health_test.go b/components/control-plane/internal/reconciler/health_test.go
index a0a9da902..aa697e60a 100644
--- a/components/control-plane/internal/reconciler/health_test.go
+++ b/components/control-plane/internal/reconciler/health_test.go
@@ -526,8 +526,9 @@ func (f *fakeGatewayClient) UpdateGateway(ctx context.Context, in *pb.UpdateGate
 }
 
 func TestListAllGateways_Pagination(t *testing.T) {
-	// 250 gateways distributed across 3 pages (100, 100, 50).
-	total := 250
+	// 1200 gateways distributed across 3 pages (500, 500, 200) at the shared
+	// helper's gatewayListPageSize.
+	total := 1200
 	allGWs := make([]*pb.Gateway, total)
 	for i := 0; i < total; i++ {
 		allGWs[i] = &pb.Gateway{
@@ -558,8 +559,7 @@ func TestListAllGateways_Pagination(t *testing.T) {
 		},
 	}
 
-	h := &GatewayHealthReconciler{}
-	got, err := h.listAllGateways(context.Background(), client)
+	got, err := listAllGateways(context.Background(), client, "")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -605,8 +605,7 @@ func TestListAllGateways_SinglePage(t *testing.T) {
 		},
 	}
 
-	h := &GatewayHealthReconciler{}
-	got, err := h.listAllGateways(context.Background(), client)
+	got, err := listAllGateways(context.Background(), client, "")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -618,6 +617,41 @@ func TestListAllGateways_SinglePage(t *testing.T) {
 	}
 }
 
+func TestListAllGateways_ClusterIDFilter(t *testing.T) {
+	// A non-empty clusterID must be sent as the request's optional cluster_id so
+	// the api-server scopes the listing server-side (the pull-model boundary); an
+	// empty clusterID must send nil so the single-cluster default lists all.
+	tests := []struct {
+		name      string
+		clusterID string
+		wantSet   bool
+	}{
+		{name: "scoped", clusterID: "2abc", wantSet: true},
+		{name: "unscoped", clusterID: "", wantSet: false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var gotClusterID *string
+			client := &fakeGatewayClient{
+				listFn: func(ctx context.Context, in *pb.ListGatewaysRequest, opts ...grpc.CallOption) (*pb.ListGatewaysResponse, error) {
+					gotClusterID = in.ClusterId
+					return &pb.ListGatewaysResponse{Metadata: &pb.ListMeta{Total: 0}}, nil
+				},
+			}
+			if _, err := listAllGateways(context.Background(), client, tc.clusterID); err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if tc.wantSet {
+				if gotClusterID == nil || *gotClusterID != tc.clusterID {
+					t.Fatalf("cluster_id = %v, want %q", gotClusterID, tc.clusterID)
+				}
+			} else if gotClusterID != nil {
+				t.Fatalf("cluster_id = %q, want unset", *gotClusterID)
+			}
+		})
+	}
+}
+
 func TestListAllGateways_Empty(t *testing.T) {
 	var callCount int
 	client := &fakeGatewayClient{
@@ -630,8 +664,7 @@ func TestListAllGateways_Empty(t *testing.T) {
 		},
 	}
 
-	h := &GatewayHealthReconciler{}
-	got, err := h.listAllGateways(context.Background(), client)
+	got, err := listAllGateways(context.Background(), client, "")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
diff --git a/components/control-plane/internal/reconciler/namespace.go b/components/control-plane/internal/reconciler/namespace.go
index 9db22e81b..3138710ec 100644
--- a/components/control-plane/internal/reconciler/namespace.go
+++ b/components/control-plane/internal/reconciler/namespace.go
@@ -180,7 +180,13 @@ func (r *NamespaceGCReconciler) grpcLiveNamespaces(ctx context.Context) (map[str
 	listCtx, cancel := context.WithTimeout(ctx, gatewayListTimeout)
 	defer cancel()
 	client := pb.NewGatewayServiceClient(r.grpcConn)
-	gateways, err := listAllGateways(listCtx, client)
+	// Deliberately unfiltered by cluster_id: the live set only ever PROTECTS
+	// namespaces from reaping, and gateway namespaces are globally unique, so a
+	// superset that includes other clusters' gateways can never cause a wrong
+	// delete -- whereas filtering to this cluster could reap a co-located hub's
+	// namespace during the same-cluster test. Reaping decisions are further gated
+	// by a local label selector, so foreign entries in the live set are inert.
+	gateways, err := listAllGateways(listCtx, client, "")
 	if err != nil {
 		return nil, fmt.Errorf("list gateways: %w", err)
 	}
diff --git a/components/control-plane/internal/reconciler/reconciler.go b/components/control-plane/internal/reconciler/reconciler.go
index c7bc5af8a..f66bcdcc1 100644
--- a/components/control-plane/internal/reconciler/reconciler.go
+++ b/components/control-plane/internal/reconciler/reconciler.go
@@ -1888,12 +1888,20 @@ const gatewayListPageSize = 500
 // gateway. The list endpoint is server-side paginated (default page size 20),
 // so callers that must reason about the whole fleet (the namespace reaper and
 // the health reconciler) cannot rely on a single unpaged request.
-func listAllGateways(ctx context.Context, client pb.GatewayServiceClient) ([]*pb.Gateway, error) {
+//
+// clusterID, when non-empty, scopes the listing server-side to gateways with
+// that cluster_id. The health reconciler passes its managed-cluster identity so
+// it never stamps a foreign cluster's gateway; the namespace reaper passes ""
+// on purpose (an unfiltered live set is a safe superset that only ever protects
+// namespaces, and gateway namespaces are globally unique, so it never reaps
+// another cluster's namespace).
+func listAllGateways(ctx context.Context, client pb.GatewayServiceClient, clusterID string) ([]*pb.Gateway, error) {
 	var all []*pb.Gateway
 	for page := int32(1); ; page++ {
 		resp, err := client.ListGateways(ctx, &pb.ListGatewaysRequest{
-			Page: page,
-			Size: gatewayListPageSize,
+			Page:      page,
+			Size:      gatewayListPageSize,
+			ClusterId: watcher.OptionalClusterID(clusterID),
 		})
 		if err != nil {
 			return nil, err
diff --git a/components/control-plane/internal/reconciler/reconciler_test.go b/components/control-plane/internal/reconciler/reconciler_test.go
index 7d741b19f..ee8164b9e 100644
--- a/components/control-plane/internal/reconciler/reconciler_test.go
+++ b/components/control-plane/internal/reconciler/reconciler_test.go
@@ -718,7 +718,7 @@ func TestWatchGateways_KeycloakRetryPreservesGatedPayload(t *testing.T) {
 	ctx, cancel := context.WithCancel(t.Context())
 	watchErr := make(chan error, 1)
 	go func() {
-		watchErr <- watcher.WatchGateways(ctx, grpcConn, r)
+		watchErr <- watcher.WatchGateways(ctx, grpcConn, r, "")
 	}()
 
 	deadline := time.NewTimer(8 * time.Second)
diff --git a/components/control-plane/internal/reconciler/sandboxcount.go b/components/control-plane/internal/reconciler/sandboxcount.go
index dfb116e62..b920e2d52 100644
--- a/components/control-plane/internal/reconciler/sandboxcount.go
+++ b/components/control-plane/internal/reconciler/sandboxcount.go
@@ -53,6 +53,14 @@ type SandboxCountReconciler struct {
 	grpcConn       *grpc.ClientConn
 	resyncInterval time.Duration
 
+	// clusterID scopes the self-heal's gateway enumeration to this managed
+	// cluster. The self-heal writes an absolute count (including zero) for every
+	// gateway it enumerates, reconciling against pods it observes LOCALLY; an
+	// unfiltered enumeration would make a spoke SET a foreign cluster's gateway
+	// count to zero because it observes none of that cluster's pods. Empty
+	// enumerates all gateways (single-cluster default).
+	clusterID string
+
 	// adjust, set, and namespaces are the seams to the API server, overridable in
 	// tests so the reconciliation logic can be exercised without a live gRPC
 	// server. adjust applies a relative delta; set writes an absolute count; and
@@ -88,7 +96,7 @@ type SandboxCountReconciler struct {
 
 // NewSandboxCountReconciler builds a SandboxCountReconciler, applying the default
 // resync interval for any non-positive value.
-func NewSandboxCountReconciler(client kubernetes.Interface, grpcConn *grpc.ClientConn, resyncInterval time.Duration) *SandboxCountReconciler {
+func NewSandboxCountReconciler(client kubernetes.Interface, grpcConn *grpc.ClientConn, resyncInterval time.Duration, clusterID string) *SandboxCountReconciler {
 	if resyncInterval <= 0 {
 		resyncInterval = defaultSandboxCountResyncInterval
 	}
@@ -96,6 +104,7 @@ func NewSandboxCountReconciler(client kubernetes.Interface, grpcConn *grpc.Clien
 		client:         client,
 		grpcConn:       grpcConn,
 		resyncInterval: resyncInterval,
+		clusterID:      clusterID,
 		baseCtx:        context.Background(),
 		nsLocks:        make(map[string]*sync.Mutex),
 	}
@@ -353,7 +362,7 @@ func (r *SandboxCountReconciler) grpcSet(ctx context.Context, namespace string,
 // must converge back to zero and which therefore have no pods in the cache.
 func (r *SandboxCountReconciler) grpcGatewayNamespaces(ctx context.Context) ([]string, error) {
 	client := pb.NewGatewayServiceClient(r.grpcConn)
-	gateways, err := listAllGateways(ctx, client)
+	gateways, err := listAllGateways(ctx, client, r.clusterID)
 	if err != nil {
 		return nil, err
 	}
diff --git a/components/control-plane/internal/reconciler/sandboxcount_test.go b/components/control-plane/internal/reconciler/sandboxcount_test.go
index 978d300b5..c986f1f82 100644
--- a/components/control-plane/internal/reconciler/sandboxcount_test.go
+++ b/components/control-plane/internal/reconciler/sandboxcount_test.go
@@ -65,7 +65,7 @@ func (rec *seamRecorder) adjust(ns string, delta int) int {
 // no gRPC connection, following the namespace_test.go convention of exercising
 // the pure sub-methods directly.
 func newTestSandboxCount(rec *seamRecorder) *SandboxCountReconciler {
-	r := NewSandboxCountReconciler(fake.NewSimpleClientset(), nil, time.Minute)
+	r := NewSandboxCountReconciler(fake.NewSimpleClientset(), nil, time.Minute, "")
 	r.adjust = func(ctx context.Context, ns string, delta int) error {
 		rec.mu.Lock()
 		defer rec.mu.Unlock()
diff --git a/components/control-plane/internal/watcher/seed_test.go b/components/control-plane/internal/watcher/seed_test.go
index 255eb8da1..2408192cf 100644
--- a/components/control-plane/internal/watcher/seed_test.go
+++ b/components/control-plane/internal/watcher/seed_test.go
@@ -26,6 +26,10 @@ type fakeGatewayLister struct {
 	getExtra map[string]*pb.Gateway
 	getErr   error
 	getCalls int
+	// gotClusterID records the cluster_id sent on the most recent ListGateways
+	// call, so tests can assert the seed threads the control-plane's identity
+	// through to the server-side filter.
+	gotClusterID *string
 }
 
 // GetGateway confirms whether an id still exists. It returns any gateway in the
@@ -50,6 +54,7 @@ func (f *fakeGatewayLister) GetGateway(_ context.Context, in *pb.GetGatewayReque
 
 func (f *fakeGatewayLister) ListGateways(_ context.Context, in *pb.ListGatewaysRequest, _ ...grpc.CallOption) (*pb.ListGatewaysResponse, error) {
 	f.calls++
+	f.gotClusterID = in.ClusterId
 	if f.err != nil {
 		return nil, f.err
 	}
@@ -142,7 +147,7 @@ func TestSeedGateways_ForcesActivePhasesOnly(t *testing.T) {
 	}}
 
 	sink := newRecordingSink(nil)
-	err := seedGateways(context.Background(), lister, sink)
+	err := seedGateways(context.Background(), lister, sink, "")
 	if err != nil {
 		t.Fatalf("seedGateways: %v", err)
 	}
@@ -175,6 +180,31 @@ func TestSeedGateways_ForcesActivePhasesOnly(t *testing.T) {
 	}
 }
 
+// seedGateways must thread the control-plane's cluster identity into the
+// ListGateways request so the server-side filter scopes the seed to this
+// managed cluster (the pull-model boundary); an empty identity must leave the
+// request's cluster_id unset so the single-cluster default seeds every gateway.
+func TestSeedGateways_ThreadsClusterID(t *testing.T) {
+	t.Run("scoped", func(t *testing.T) {
+		lister := &fakeGatewayLister{items: []*pb.Gateway{gw("a", "Running")}}
+		if err := seedGateways(context.Background(), lister, newRecordingSink(nil), "2abc"); err != nil {
+			t.Fatalf("seedGateways: %v", err)
+		}
+		if lister.gotClusterID == nil || *lister.gotClusterID != "2abc" {
+			t.Fatalf("ListGateways cluster_id = %v, want %q", lister.gotClusterID, "2abc")
+		}
+	})
+	t.Run("unscoped", func(t *testing.T) {
+		lister := &fakeGatewayLister{items: []*pb.Gateway{gw("a", "Running")}}
+		if err := seedGateways(context.Background(), lister, newRecordingSink(nil), ""); err != nil {
+			t.Fatalf("seedGateways: %v", err)
+		}
+		if lister.gotClusterID != nil {
+			t.Fatalf("ListGateways cluster_id = %q, want unset", *lister.gotClusterID)
+		}
+	})
+}
+
 // A gateway inventory larger than one page must be seeded in full.
 func TestSeedGateways_Paginates(t *testing.T) {
 	var items []*pb.Gateway
@@ -184,7 +214,7 @@ func TestSeedGateways_Paginates(t *testing.T) {
 	lister := &fakeGatewayLister{items: items, pageSize: gatewaySeedPageSize}
 
 	sink := newRecordingSink(nil)
-	if err := seedGateways(context.Background(), lister, sink); err != nil {
+	if err := seedGateways(context.Background(), lister, sink, ""); err != nil {
 		t.Fatalf("seedGateways: %v", err)
 	}
 	if count := len(sink.enqueued); count != len(items) {
@@ -234,7 +264,7 @@ func TestSeedGateways_RepeatsUntilInventoryStable(t *testing.T) {
 	}}
 
 	sink := newRecordingSink(nil)
-	if err := seedGateways(context.Background(), lister, sink); err != nil {
+	if err := seedGateways(context.Background(), lister, sink, ""); err != nil {
 		t.Fatalf("seedGateways: %v", err)
 	}
 
@@ -262,7 +292,7 @@ func TestSeedGateways_ErrorsWhenInventoryNeverStabilizes(t *testing.T) {
 	}}
 
 	sink := newRecordingSink(nil)
-	if err := seedGateways(context.Background(), lister, sink); err == nil {
+	if err := seedGateways(context.Background(), lister, sink, ""); err == nil {
 		t.Fatal("seedGateways must error when the inventory never stabilizes, so watchLoop retries")
 	}
 }
@@ -277,7 +307,7 @@ func TestSeedGateways_ListErrorPropagates(t *testing.T) {
 	sink := newRecordingSink(map[string]Event[*pb.Gateway]{
 		"gw-1": {Type: EventUpdated, ResourceID: "gw-1", Resource: gw("gw-1", "Provisioning")},
 	})
-	err := seedGateways(context.Background(), lister, sink)
+	err := seedGateways(context.Background(), lister, sink, "")
 	if err == nil {
 		t.Fatal("want an error when ListGateways fails")
 	}
@@ -303,7 +333,7 @@ func TestSeedGateways_PrunesAbsentTrackedGateways(t *testing.T) {
 		"deleting":      {Type: EventDeleted, ResourceID: "deleting", Resource: gw("deleting", "Running")},
 	})
 
-	if err := seedGateways(context.Background(), lister, sink); err != nil {
+	if err := seedGateways(context.Background(), lister, sink, ""); err != nil {
 		t.Fatalf("seedGateways: %v", err)
 	}
 
@@ -340,7 +370,7 @@ func TestSeedGateways_KeepsListOmittedButLiveGateway(t *testing.T) {
 		"shifted": {Type: EventUpdated, ResourceID: "shifted", Resource: gw("shifted", "Provisioning")},
 	})
 
-	if err := seedGateways(context.Background(), lister, sink); err != nil {
+	if err := seedGateways(context.Background(), lister, sink, ""); err != nil {
 		t.Fatalf("seedGateways: %v", err)
 	}
 
@@ -377,7 +407,7 @@ func TestSeedGateways_KeepsAbsentWhenConfirmFails(t *testing.T) {
 		"maybe":   {Type: EventUpdated, ResourceID: "maybe", Resource: gw("maybe", "Provisioning")},
 	})
 
-	if err := seedGateways(context.Background(), lister, sink); err != nil {
+	if err := seedGateways(context.Background(), lister, sink, ""); err != nil {
 		t.Fatalf("seedGateways: %v", err)
 	}
 
diff --git a/components/control-plane/internal/watcher/watcher.go b/components/control-plane/internal/watcher/watcher.go
index 79562510f..e07297f1a 100644
--- a/components/control-plane/internal/watcher/watcher.go
+++ b/components/control-plane/internal/watcher/watcher.go
@@ -365,7 +365,24 @@ func WatchGatewayReleases(ctx context.Context, conn *grpc.ClientConn, handler Ha
 	})
 }
 
-func WatchGateways(ctx context.Context, conn *grpc.ClientConn, handler Handler[*pb.Gateway]) error {
+// OptionalClusterID maps a control-plane cluster identity to the proto optional
+// cluster_id field: an empty identity becomes nil (no server-side filter, the
+// single-cluster default), a non-empty one is sent so the api-server scopes the
+// list/watch to that cluster. Exported and shared: the reconciler's list helper
+// uses it too, so the mapping stays single-sourced.
+func OptionalClusterID(clusterID string) *string {
+	if clusterID == "" {
+		return nil
+	}
+	return &clusterID
+}
+
+// WatchGateways streams gateway events and drives them through a per-resource
+// reconcile queue. When clusterID is non-empty the watch and its seed lists are
+// scoped server-side to gateways with that cluster_id, so a managed-cluster
+// spoke only ever reconciles its own gateways (the pull model); empty watches
+// every gateway.
+func WatchGateways(ctx context.Context, conn *grpc.ClientConn, handler Handler[*pb.Gateway], clusterID string) error {
 	client := pb.NewGatewayServiceClient(conn)
 	// Gateway reconciliation is driven through a per-resource reconcile queue rather
 	// than invoked inline: the watch stream does not replay state on reconnect, so a
@@ -385,7 +402,7 @@ func WatchGateways(ctx context.Context, conn *grpc.ClientConn, handler Handler[*
 		runCtx, runCancel := context.WithCancel(ctx)
 		defer runCancel()
 
-		stream, err := client.WatchGateways(runCtx, &pb.WatchGatewaysRequest{})
+		stream, err := client.WatchGateways(runCtx, &pb.WatchGatewaysRequest{ClusterId: OptionalClusterID(clusterID)})
 		if err != nil {
 			return fmt.Errorf("starting gateway watch: %w", err)
 		}
@@ -452,7 +469,7 @@ func WatchGateways(ctx context.Context, conn *grpc.ClientConn, handler Handler[*
 		// the seed to the receiver: a Recv error cancels runCtx, aborting the
 		// seed's in-flight RPCs so mutations during the dead window are not
 		// masked by an unchanged stable ID set.
-		if err := seedGateways(runCtx, client, rq); err != nil {
+		if err := seedGateways(runCtx, client, rq, clusterID); err != nil {
 			// Distinguish two causes so a genuine seed failure is never masked by
 			// the cancellation we would cause ourselves. If runCtx is already
 			// canceled, the receiver ended first (its Recv error/EOF canceled
@@ -577,8 +594,8 @@ var _ gatewaySeedSink = (*reconcileQueue[*pb.Gateway])(nil)
 // NamespaceGCReconciler, which rechecks liveness before it deletes -- safer than
 // synthesizing a delete here. Absence is only trusted after a successful list:
 // any page error aborts before pruning.
-func seedGateways(ctx context.Context, client pb.GatewayServiceClient, sink gatewaySeedSink) error {
-	inventory, err := listGatewaysStable(ctx, client)
+func seedGateways(ctx context.Context, client pb.GatewayServiceClient, sink gatewaySeedSink, clusterID string) error {
+	inventory, err := listGatewaysStable(ctx, client, clusterID)
 	if err != nil {
 		return err
 	}
@@ -671,11 +688,11 @@ func enqueueSeed(sink gatewaySeedSink, gw *pb.Gateway) (forced bool) {
 // stream that never reconnects would never reseed to correct it. The error aborts
 // this connect attempt so watchLoop backs off and retries the whole seed on a
 // fresh stream.
-func listGatewaysStable(ctx context.Context, client pb.GatewayServiceClient) (map[string]*pb.Gateway, error) {
+func listGatewaysStable(ctx context.Context, client pb.GatewayServiceClient, clusterID string) (map[string]*pb.Gateway, error) {
 	const maxSeedListPasses = 5
 	var prevIDs map[string]struct{}
 	for pass := 1; pass <= maxSeedListPasses; pass++ {
-		current, err := listGatewaysOnce(ctx, client)
+		current, err := listGatewaysOnce(ctx, client, clusterID)
 		if err != nil {
 			return nil, err
 		}
@@ -694,10 +711,10 @@ func listGatewaysStable(ctx context.Context, client pb.GatewayServiceClient) (ma
 // listGatewaysOnce performs a single paginated pass over the gateway inventory,
 // returning it keyed by ID (which also dedupes an item a concurrent create caused
 // to appear on two pages).
-func listGatewaysOnce(ctx context.Context, client pb.GatewayServiceClient) (map[string]*pb.Gateway, error) {
+func listGatewaysOnce(ctx context.Context, client pb.GatewayServiceClient, clusterID string) (map[string]*pb.Gateway, error) {
 	inventory := make(map[string]*pb.Gateway)
 	for page := int32(1); ; page++ {
-		resp, err := client.ListGateways(ctx, &pb.ListGatewaysRequest{Page: page, Size: gatewaySeedPageSize})
+		resp, err := client.ListGateways(ctx, &pb.ListGatewaysRequest{Page: page, Size: gatewaySeedPageSize, ClusterId: OptionalClusterID(clusterID)})
 		if err != nil {
 			return nil, fmt.Errorf("listing gateways to seed reconcile queue: %w", err)
 		}