From d1985e9a14cbc008a6791b9f3935e1547f82aed5 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 20:16:29 -0400 Subject: [PATCH 01/16] test(substrate): reproduce one tenant reading another's Substrate inventory Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index f28c0743c5..24199b2297 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -719,3 +719,77 @@ func TestSummaryDeniedPoolsDoNotRequireKubernetes(t *testing.T) { _, err = service.GetSubstrateSummary(t.Context(), "", "") require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeUnauthenticated)) } + +// Two kagent installs, each watching only its own namespace, share one ate-api. Each +// must see only its own atespace, and refuse a request naming the other. +func TestSubstrateTenantIsolation(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, atev1alpha1.AddToScheme(scheme)) + ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) + ateClient := &fakeATEClient{ + pageSize: 1, + actors: []*ateapipb.Actor{ + substrateActor("session-b", "tenant-b", ateapipb.ActorState_ACTOR_STATE_RUNNING, "tenant-b", "pod-b"), + substrateActor("session-a", "tenant-a", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, "", ""), + }, + templates: []*ateapipb.ActorTemplate{ + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-a", Name: "agent-a"}}, + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-b", Name: "agent-b"}}, + }, + workers: []*ateapipb.Worker{ + {WorkerNamespace: "tenant-a", WorkerPod: "pod-a"}, + {WorkerNamespace: "tenant-b", WorkerPod: "pod-b"}, + }, + } + tenantA := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + t.Run("lists only its own actors", func(t *testing.T) { + var names []string + request := &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 1}} + for { + page, err := tenantA.ListSubstrateActors(ctx, request) + require.NoError(t, err) + for _, actor := range page.Actors { + names = append(names, actor.GetMetadata().GetAtespace()+"/"+actor.GetMetadata().GetName()) + } + if page.NextPageToken == "" { + break + } + request.Page.PageToken = page.NextPageToken + } + assert.Equal(t, []string{"tenant-a/session-a"}, names) + }) + + t.Run("summarises only its own atespace", func(t *testing.T) { + summary, err := tenantA.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + require.Empty(t, summary.ATEAPIError) + require.Len(t, summary.ActorTemplates, 1) + assert.Equal(t, "agent-a", summary.ActorTemplates[0].GetMetadata().GetName()) + assert.Equal(t, int64(1), summary.ActorCount) + assert.Equal(t, int64(0), summary.RunningActorCount) + assert.Equal(t, int64(1), summary.WorkerCount) + }) + + t.Run("refuses another tenant's atespace or namespace", func(t *testing.T) { + _, err := tenantA.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + _, err = tenantA.GetSubstrateSummary(ctx, "", "tenant-b") + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + _, err = tenantA.GetSubstrateSummary(ctx, "tenant-b", "") + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + _, err = tenantA.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "tenant-b"}) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + }) + + t.Run("a cluster-wide install still sees every tenant", func(t *testing.T) { + clusterWide := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), nil, &pkgAuth.NoopAuthorizer{}, ateClient) + summary, err := clusterWide.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + assert.Equal(t, int64(2), summary.ActorCount) + assert.Len(t, summary.ActorTemplates, 2) + page, err := clusterWide.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) + require.NoError(t, err) + require.Len(t, page.Actors, 1) + }) +} From 363376935aeebe3143f02e45dec8ff9bdc0c6735 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 20:21:09 -0400 Subject: [PATCH 02/16] fix(substrate): scope actors and templates to the watched namespaces A kagent that watches only its own namespace listed every tenant's actors and ActorTemplates from a shared Substrate, and served a foreign atespace or namespace when a caller named one. Resolve one scope from the watched namespaces for every Substrate read. An empty request means the watched namespaces, and a namespace or atespace outside them is refused with PermissionDenied. A cluster-wide install still sees everything. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 16 +-- .../internal/service/system/service_test.go | 5 +- go/core/internal/service/system/substrate.go | 131 +++++++++++------- 3 files changed, 86 insertions(+), 66 deletions(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index 38fb882341..d556b11a35 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -148,16 +148,6 @@ func namespacesFromNames(names []string) []Namespace { return result } -func (s *Service) substrateNamespaces(requested string) []string { - if requested != "" { - return []string{requested} - } - if len(s.observedNamespaces) > 0 { - return slices.Clone(s.observedNamespaces) - } - return []string{""} -} - func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev1alpha1.WorkerPool, error) { var options []client.ListOption if namespace != "" { @@ -173,14 +163,14 @@ func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev } // substrateActorTemplates drains upstream pagination for the configuration-sized list. -func (s *Service) substrateActorTemplates(ctx context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) { - templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, atespace) +func (s *Service) substrateActorTemplates(ctx context.Context, scope substrateScope) ([]*ateapipb.ActorTemplate, error) { + templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, scope.atespace()) if err != nil { return nil, err } templates := make([]*ateapipb.ActorTemplate, 0, len(templatesFromAPI)) for _, template := range templatesFromAPI { - if template == nil { + if template == nil || !scope.contains(template.GetMetadata().GetAtespace()) { continue } // Only expose inventory fields: containers can contain resolved credentials. diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 24199b2297..31bc12594a 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -534,7 +534,7 @@ func TestSubstrateScopesAreIndependent(t *testing.T) { }, workers: []*ateapipb.Worker{{WorkerNamespace: "workers", WorkerPod: "pod"}}, } - service := system.NewService(kubeClient, []string{"workers"}, &pkgAuth.NoopAuthorizer{}, ateClient) + service := system.NewService(kubeClient, []string{"workers", "team-a", "shared"}, &pkgAuth.NoopAuthorizer{}, ateClient) for _, tc := range []struct { atespace string count int64 @@ -543,7 +543,6 @@ func TestSubstrateScopesAreIndependent(t *testing.T) { {atespace: "", count: 2, running: 1}, {atespace: "team-a", count: 1, running: 1}, {atespace: "shared", count: 1, running: 0}, - {atespace: "missing", count: 0, running: 0}, } { t.Run(tc.atespace, func(t *testing.T) { summary, err := service.GetSubstrateSummary(ctx, "", tc.atespace) @@ -659,7 +658,7 @@ func TestSummaryRequiresCapabilityAndScopesPools(t *testing.T) { {name: "all pools", pools: all, wantPools: 3}, {name: "no pools", pools: none}, {name: "configured intersection", pools: all, observed: []string{"team"}, wantPools: 2}, - {name: "request cannot escape configuration", pools: all, observed: []string{"team"}, requested: "other"}, + {name: "request cannot escape configuration", pools: all, observed: []string{"team"}, requested: "other", code: serviceerrors.CodePermissionDenied}, {name: "request cannot escape policy", pools: constrained, requested: "other"}, {name: "pool lookup failure", failScope: true, code: serviceerrors.CodeUnavailable}, {name: "invalid pool scope", pools: apiauthorization.AuthorizationScope{Kind: apiauthorization.ScopeAnyOf}, code: serviceerrors.CodeInternal}, diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index fe1c9684bd..6a9ffc8fd1 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -91,10 +91,14 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a if err != nil { return SubstrateSummary{}, err } - // Pool configuration and ATE inventory are separate authorization domains. - // Keep the legacy ATE namespace behavior until that inventory has its own - // object contract; it cannot widen the Kubernetes pool read boundary. - namespaces := s.substrateNamespaces(requestedNamespace) + workerScope, err := s.resolveSubstrateScope(requestedNamespace) + if err != nil { + return SubstrateSummary{}, err + } + actorScope, err := s.resolveSubstrateScope(atespace) + if err != nil { + return SubstrateSummary{}, err + } result := SubstrateSummary{ WorkerPools: []atev1alpha1.WorkerPool{}, @@ -103,7 +107,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a ComputedAt: time.Now().UTC(), } - poolNamespaces := s.workerPoolNamespaces(requestedNamespace) + poolNamespaces := workerScope.listNamespaces() if poolMatcher.Empty() { poolNamespaces = nil } @@ -122,18 +126,16 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return strings.Compare(left.Namespace+"/"+left.Name, right.Namespace+"/"+right.Name) }) - allowAll, allowed := substrateScopeFilter(namespaces) - // Three independent reads: none gates the others, so one failure leaves the rest // counted rather than zeroing the whole summary. - if templates, err := s.substrateActorTemplates(ctx, atespace); err != nil { + if templates, err := s.substrateActorTemplates(ctx, actorScope); err != nil { result.recordATEError(ctx, err) } else { result.ActorTemplates = templates } statusCounts := map[ateapipb.ActorState]int64{} - if err := s.walkActors(ctx, atespace, func(actor *ateapipb.Actor) { + if err := s.walkActors(ctx, actorScope, func(actor *ateapipb.Actor) { if actor == nil { return } @@ -147,7 +149,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a } if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { - if worker == nil || !allowedWorkerNamespace(worker.GetWorkerNamespace(), allowAll, allowed) { + if !workerScope.containsWorker(worker) { return } result.WorkerCount++ @@ -168,19 +170,24 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return result, nil } -// ListSubstrateActors returns one upstream page without changing its order or token. +// ListSubstrateActors filters one upstream page by atespace and preserves its order and token. func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubstrateActorsRequest) (SubstrateActorPage, error) { - if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: substrateResource}); err != nil { + scope, err := s.authorizeSubstrateScope(ctx, input.GetAtespace()) + if err != nil { return SubstrateActorPage{}, err } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, input.GetAtespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + actors, next, err := s.ateClient.ListActorsPage(ctx, scope.atespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) return result, nil } - result.Actors = actors + for _, actor := range actors { + if actor != nil && scope.contains(actor.GetMetadata().GetAtespace()) { + result.Actors = append(result.Actors, actor) + } + } result.NextPageToken = next return result, nil } @@ -188,7 +195,7 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs // ListSubstrateWorkers filters one upstream page by namespace and preserves its token. // A page with no workers in scope may still have a next page. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { - namespaces, err := s.substrateScope(ctx, input.GetNamespace()) + scope, err := s.authorizeSubstrateScope(ctx, input.GetNamespace()) if err != nil { return SubstrateWorkerPage{}, err } @@ -199,9 +206,8 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) return result, nil } - allowAll, allowed := substrateScopeFilter(namespaces) for _, worker := range workers { - if worker != nil && allowedWorkerNamespace(worker.GetWorkerNamespace(), allowAll, allowed) { + if scope.containsWorker(worker) { result.Workers = append(result.Workers, worker) } } @@ -209,12 +215,16 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } -// walkActors visits every actor in the requested atespace, one page at a time. -func (s *Service) walkActors(ctx context.Context, atespace string, visit func(*ateapipb.Actor)) error { +// walkActors visits every actor in scope, one page at a time. +func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit func(*ateapipb.Actor)) error { read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { - return s.ateClient.ListActorsPage(ctx, atespace, pageSize, pageToken) + return s.ateClient.ListActorsPage(ctx, scope.atespace(), pageSize, pageToken) } - return walkSubstrate(ctx, read, visit) + return walkSubstrate(ctx, read, func(actor *ateapipb.Actor) { + if actor != nil && scope.contains(actor.GetMetadata().GetAtespace()) { + visit(actor) + } + }) } // walkWorkers calls visit for every worker ate-api holds, one page at a time. @@ -248,36 +258,64 @@ func walkSubstrate[Row any]( return fmt.Errorf("ate-api did not finish paging after %d pages", maxATEPagesPerWalk) } -// substrateScope authorizes the caller and resolves the namespaces a read covers. -// ATE-only actor reads authorize independently of Kubernetes scope. -func (s *Service) substrateScope(ctx context.Context, requestedNamespace string) ([]string, error) { +// substrateScope is the set of namespaces a Substrate read covers. An atespace is named +// after the namespace of the agent it serves, so the same set bounds actors and templates. +type substrateScope struct { + all bool + namespaces []string +} + +// authorizeSubstrateScope authorizes the caller and resolves the scope of a read. +func (s *Service) authorizeSubstrateScope(ctx context.Context, requested string) (substrateScope, error) { if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: substrateResource}); err != nil { - return nil, err + return substrateScope{}, err } - return s.substrateNamespaces(requestedNamespace), nil + return s.resolveSubstrateScope(requested) } -// substrateScopeFilter turns resolved namespaces into the pair every row filter here -// takes: whether every namespace is in scope, and the set that is when it is not. -func substrateScopeFilter(namespaces []string) (bool, map[string]struct{}) { - allowAll := len(namespaces) == 1 && namespaces[0] == "" - allowed := make(map[string]struct{}, len(namespaces)) - for _, namespace := range namespaces { - if namespace != "" { - allowed[namespace] = struct{}{} - } +// resolveSubstrateScope narrows the watched namespaces to the requested one. A +// controller that watches only some namespaces refuses any other. +func (s *Service) resolveSubstrateScope(requested string) (substrateScope, error) { + switch { + case requested == "" && len(s.observedNamespaces) == 0: + return substrateScope{all: true}, nil + case requested == "": + return substrateScope{namespaces: slices.Clone(s.observedNamespaces)}, nil + case len(s.observedNamespaces) == 0 || slices.Contains(s.observedNamespaces, requested): + return substrateScope{namespaces: []string{requested}}, nil + default: + return substrateScope{}, serviceerrors.NewPermissionDenied("Not authorized", fmt.Errorf("namespace %q is not watched by this controller", requested)) + } +} + +func (scope substrateScope) contains(namespace string) bool { + return scope.all || slices.Contains(scope.namespaces, namespace) +} + +// containsWorker also admits a worker that reports no namespace. +func (scope substrateScope) containsWorker(worker *ateapipb.Worker) bool { + if worker == nil { + return false } - return allowAll, allowed + namespace := strings.TrimSpace(worker.GetWorkerNamespace()) + return namespace == "" || scope.contains(namespace) } -// allowedWorkerNamespace filters workers by their Kubernetes namespace. -func allowedWorkerNamespace(namespace string, allowAll bool, allowed map[string]struct{}) bool { - namespace = strings.TrimSpace(namespace) - if allowAll || namespace == "" { - return true +// atespace is the atespace to ask ate-api for: the only one in scope, or every +// atespace ("") for the caller to filter. +func (scope substrateScope) atespace() string { + if len(scope.namespaces) == 1 { + return scope.namespaces[0] } - _, ok := allowed[namespace] - return ok + return "" +} + +// listNamespaces is the namespaces to list Kubernetes objects in, "" meaning all. +func (scope substrateScope) listNamespaces() []string { + if scope.all { + return []string{""} + } + return scope.namespaces } func substratePageSize(requested int32) int32 { @@ -286,10 +324,3 @@ func substratePageSize(requested int32) int32 { } return requested } - -func (s *Service) workerPoolNamespaces(requested string) []string { - if requested != "" && len(s.observedNamespaces) != 0 && !slices.Contains(s.observedNamespaces, requested) { - return nil - } - return s.substrateNamespaces(requested) -} From edb3a0dd83592be9bcdd0ddf6766ff8cdda23fdc Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 21:00:34 -0400 Subject: [PATCH 03/16] test(substrate): cover multi-namespace scope, unlabelled workers and refusal wording A controller watching tenant-a and tenant-c must never page tenant-b actors to the caller, must ask ate-api only for its own atespaces when it builds the summary, and must hide a worker that reports no namespace. The refusal message must say the namespace is not watched, since the gRPC layer sends only the message. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 112 +++++++++++++++--- 1 file changed, 97 insertions(+), 15 deletions(-) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 31bc12594a..a158a02a9c 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "strconv" "strings" "testing" @@ -54,10 +55,14 @@ type fakeATEClient struct { // was followed rather than that the rows came back. actorReads int workerReads int + // The atespaces ate-api was asked for, since asking for "" reads every tenant's rows. + actorAtespaces []string + templateAtespaces []string } func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) { client.templateReads++ + client.templateAtespaces = append(client.templateAtespaces, atespace) templates := []*ateapipb.ActorTemplate{} for _, template := range client.templates { if atespace == "" || template.GetMetadata().GetAtespace() == atespace { @@ -69,6 +74,7 @@ func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace stri func (client *fakeATEClient) ListActorsPage(_ context.Context, atespace string, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { client.actorReads++ + client.actorAtespaces = append(client.actorAtespaces, atespace) if err := client.readError(client.actorReads); err != nil { return nil, "", err } @@ -719,8 +725,8 @@ func TestSummaryDeniedPoolsDoNotRequireKubernetes(t *testing.T) { require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeUnauthenticated)) } -// Two kagent installs, each watching only its own namespace, share one ate-api. Each -// must see only its own atespace, and refuse a request naming the other. +// Kagent installs that each watch only some namespaces share one ate-api. Each must see +// only the atespaces it watches, and refuse a request naming another. func TestSubstrateTenantIsolation(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, atev1alpha1.AddToScheme(scheme)) @@ -730,49 +736,116 @@ func TestSubstrateTenantIsolation(t *testing.T) { actors: []*ateapipb.Actor{ substrateActor("session-b", "tenant-b", ateapipb.ActorState_ACTOR_STATE_RUNNING, "tenant-b", "pod-b"), substrateActor("session-a", "tenant-a", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, "", ""), + substrateActor("session-c", "tenant-c", ateapipb.ActorState_ACTOR_STATE_RUNNING, "tenant-c", "pod-c"), + substrateActor("session-b2", "tenant-b", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, "", ""), }, templates: []*ateapipb.ActorTemplate{ {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-a", Name: "agent-a"}}, {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-b", Name: "agent-b"}}, + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-c", Name: "agent-c"}}, }, workers: []*ateapipb.Worker{ {WorkerNamespace: "tenant-a", WorkerPod: "pod-a"}, {WorkerNamespace: "tenant-b", WorkerPod: "pod-b"}, + {WorkerNamespace: "tenant-c", WorkerPod: "pod-c"}, + {WorkerPod: "pod-unlabelled"}, }, } - tenantA := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + newService := func(observedNamespaces []string) *system.Service { + return system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), observedNamespaces, &pkgAuth.NoopAuthorizer{}, ateClient) + } + tenantA := newService([]string{"tenant-a"}) + tenantsAC := newService([]string{"tenant-a", "tenant-c"}) + clusterWide := newService(nil) - t.Run("lists only its own actors", func(t *testing.T) { + listAllActors := func(t *testing.T, service *system.Service, atespace string) []string { var names []string - request := &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 1}} + request := &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: 1}} for { - page, err := tenantA.ListSubstrateActors(ctx, request) + page, err := service.ListSubstrateActors(ctx, request) require.NoError(t, err) for _, actor := range page.Actors { names = append(names, actor.GetMetadata().GetAtespace()+"/"+actor.GetMetadata().GetName()) } if page.NextPageToken == "" { - break + return names } request.Page.PageToken = page.NextPageToken } - assert.Equal(t, []string{"tenant-a/session-a"}, names) + } + workerPods := func(t *testing.T, service *system.Service) []string { + var pods []string + request := &apiv1.ListSubstrateWorkersRequest{Page: &apiv1.PageRequest{}} + for { + page, err := service.ListSubstrateWorkers(ctx, request) + require.NoError(t, err) + for _, worker := range page.Workers { + pods = append(pods, worker.GetWorkerPod()) + } + if page.NextPageToken == "" { + return pods + } + request.Page.PageToken = page.NextPageToken + } + } + templateNames := func(summary system.SubstrateSummary) []string { + var names []string + for _, template := range summary.ActorTemplates { + names = append(names, template.GetMetadata().GetName()) + } + return names + } + + t.Run("lists only its own actors", func(t *testing.T) { + assert.Equal(t, []string{"tenant-a/session-a"}, listAllActors(t, tenantA, "")) }) t.Run("summarises only its own atespace", func(t *testing.T) { summary, err := tenantA.GetSubstrateSummary(ctx, "", "") require.NoError(t, err) require.Empty(t, summary.ATEAPIError) - require.Len(t, summary.ActorTemplates, 1) - assert.Equal(t, "agent-a", summary.ActorTemplates[0].GetMetadata().GetName()) + assert.Equal(t, []string{"agent-a"}, templateNames(summary)) assert.Equal(t, int64(1), summary.ActorCount) assert.Equal(t, int64(0), summary.RunningActorCount) assert.Equal(t, int64(1), summary.WorkerCount) }) + t.Run("hides a worker that reports no namespace", func(t *testing.T) { + assert.Equal(t, []string{"pod-a"}, workerPods(t, tenantA)) + assert.Equal(t, []string{"pod-a", "pod-c"}, workerPods(t, tenantsAC)) + }) + + t.Run("a controller watching two namespaces lists only those atespaces", func(t *testing.T) { + assert.Equal(t, []string{"tenant-a/session-a", "tenant-c/session-c"}, listAllActors(t, tenantsAC, "")) + + first, err := tenantsAC.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 1}}) + require.NoError(t, err) + assert.Empty(t, first.Actors, "the first upstream page holds only a tenant-b actor") + assert.NotEmpty(t, first.NextPageToken, "a page filtered to nothing still leads to the rest") + }) + + t.Run("a controller watching two namespaces asks ate-api for only those atespaces", func(t *testing.T) { + ateClient.actorAtespaces, ateClient.templateAtespaces = nil, nil + summary, err := tenantsAC.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + require.Empty(t, summary.ATEAPIError) + assert.Equal(t, int64(2), summary.ActorCount) + assert.Equal(t, int64(1), summary.RunningActorCount) + assert.ElementsMatch(t, []system.SubstrateActorStatusCount{ + {State: ateapipb.ActorState_ACTOR_STATE_RUNNING, Count: 1}, + {State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, Count: 1}, + }, summary.ActorStatusCounts) + assert.Equal(t, []string{"agent-a", "agent-c"}, templateNames(summary)) + assert.Equal(t, int64(2), summary.WorkerCount) + + assert.Equal(t, []string{"tenant-a", "tenant-c"}, slices.Compact(slices.Sorted(slices.Values(ateClient.actorAtespaces)))) + assert.ElementsMatch(t, []string{"tenant-a", "tenant-c"}, ateClient.templateAtespaces) + }) + t.Run("refuses another tenant's atespace or namespace", func(t *testing.T) { _, err := tenantA.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + assert.Contains(t, serviceerrors.MessageOf(err), "not watched by this controller", "the caller sees only the message, not the cause") _, err = tenantA.GetSubstrateSummary(ctx, "", "tenant-b") assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) _, err = tenantA.GetSubstrateSummary(ctx, "tenant-b", "") @@ -782,13 +855,22 @@ func TestSubstrateTenantIsolation(t *testing.T) { }) t.Run("a cluster-wide install still sees every tenant", func(t *testing.T) { - clusterWide := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), nil, &pkgAuth.NoopAuthorizer{}, ateClient) summary, err := clusterWide.GetSubstrateSummary(ctx, "", "") require.NoError(t, err) - assert.Equal(t, int64(2), summary.ActorCount) - assert.Len(t, summary.ActorTemplates, 2) - page, err := clusterWide.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) + assert.Equal(t, int64(4), summary.ActorCount) + assert.Len(t, summary.ActorTemplates, 3) + assert.Equal(t, int64(4), summary.WorkerCount) + assert.Equal(t, []string{"pod-a", "pod-b", "pod-c", "pod-unlabelled"}, workerPods(t, clusterWide)) + assert.Equal(t, []string{"tenant-b/session-b", "tenant-b/session-b2"}, listAllActors(t, clusterWide, "tenant-b")) + }) + + t.Run("a cluster-wide install counts nothing in an atespace with no rows", func(t *testing.T) { + summary, err := clusterWide.GetSubstrateSummary(ctx, "", "missing") require.NoError(t, err) - require.Len(t, page.Actors, 1) + assert.Empty(t, summary.ATEAPIError) + assert.Zero(t, summary.ActorCount) + assert.Zero(t, summary.RunningActorCount) + assert.Empty(t, summary.ActorStatusCounts) + assert.Empty(t, summary.ActorTemplates) }) } From ddc943bdf4ce4a5405de456c775d7a38538e3f45 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 21:02:39 -0400 Subject: [PATCH 04/16] fix(substrate): ask ate-api per watched atespace and hide unlabelled workers The summary for a controller watching several namespaces asked ate-api for every atespace and filtered the rows itself, so ate-api served every tenant's actors and templates to it. It now walks actors and drains templates once per atespace in scope. ListSubstrateActors still reads one upstream page across all atespaces and filters it, because one upstream token cannot span several requests. A worker that reports no namespace now belongs only to a cluster-wide scope, and the refusal message names the namespace that is not watched, since the gRPC layer sends only the message. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 30 ++++++----- .../internal/service/system/service_test.go | 7 +-- go/core/internal/service/system/substrate.go | 52 ++++++------------- 3 files changed, 34 insertions(+), 55 deletions(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index d556b11a35..89ae8439c2 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -164,21 +164,23 @@ func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev // substrateActorTemplates drains upstream pagination for the configuration-sized list. func (s *Service) substrateActorTemplates(ctx context.Context, scope substrateScope) ([]*ateapipb.ActorTemplate, error) { - templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, scope.atespace()) - if err != nil { - return nil, err - } - templates := make([]*ateapipb.ActorTemplate, 0, len(templatesFromAPI)) - for _, template := range templatesFromAPI { - if template == nil || !scope.contains(template.GetMetadata().GetAtespace()) { - continue + templates := []*ateapipb.ActorTemplate{} + for _, atespace := range scope.listNamespaces() { + templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, atespace) + if err != nil { + return nil, err + } + for _, template := range templatesFromAPI { + if template == nil { + continue + } + // Only expose inventory fields: containers can contain resolved credentials. + templates = append(templates, &ateapipb.ActorTemplate{ + Metadata: template.GetMetadata(), + Status: template.GetStatus(), + WorkerSelector: template.GetWorkerSelector(), + }) } - // Only expose inventory fields: containers can contain resolved credentials. - templates = append(templates, &ateapipb.ActorTemplate{ - Metadata: template.GetMetadata(), - Status: template.GetStatus(), - WorkerSelector: template.GetWorkerSelector(), - }) } slices.SortStableFunc(templates, func(left, right *ateapipb.ActorTemplate) int { diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index a158a02a9c..fbded3fced 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -53,9 +53,8 @@ type fakeATEClient struct { pageSize int // How many page reads each list has taken, so a test can assert that the token // was followed rather than that the rows came back. - actorReads int - workerReads int - // The atespaces ate-api was asked for, since asking for "" reads every tenant's rows. + actorReads int + workerReads int actorAtespaces []string templateAtespaces []string } @@ -725,8 +724,6 @@ func TestSummaryDeniedPoolsDoNotRequireKubernetes(t *testing.T) { require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeUnauthenticated)) } -// Kagent installs that each watch only some namespaces share one ate-api. Each must see -// only the atespaces it watches, and refuse a request naming another. func TestSubstrateTenantIsolation(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, atev1alpha1.AddToScheme(scheme)) diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 6a9ffc8fd1..dc020ac3d6 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -149,7 +149,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a } if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { - if !workerScope.containsWorker(worker) { + if worker == nil || !workerScope.contains(worker.GetWorkerNamespace()) { return } result.WorkerCount++ @@ -176,8 +176,12 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs if err != nil { return SubstrateActorPage{}, err } + atespace := "" + if len(scope.namespaces) == 1 { + atespace = scope.namespaces[0] + } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, scope.atespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + actors, next, err := s.ateClient.ListActorsPage(ctx, atespace, substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) @@ -207,7 +211,7 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } for _, worker := range workers { - if scope.containsWorker(worker) { + if worker != nil && scope.contains(worker.GetWorkerNamespace()) { result.Workers = append(result.Workers, worker) } } @@ -215,16 +219,16 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } -// walkActors visits every actor in scope, one page at a time. func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit func(*ateapipb.Actor)) error { - read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { - return s.ateClient.ListActorsPage(ctx, scope.atespace(), pageSize, pageToken) - } - return walkSubstrate(ctx, read, func(actor *ateapipb.Actor) { - if actor != nil && scope.contains(actor.GetMetadata().GetAtespace()) { - visit(actor) + for _, atespace := range scope.listNamespaces() { + read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { + return s.ateClient.ListActorsPage(ctx, atespace, pageSize, pageToken) } - }) + if err := walkSubstrate(ctx, read, visit); err != nil { + return err + } + } + return nil } // walkWorkers calls visit for every worker ate-api holds, one page at a time. @@ -258,14 +262,11 @@ func walkSubstrate[Row any]( return fmt.Errorf("ate-api did not finish paging after %d pages", maxATEPagesPerWalk) } -// substrateScope is the set of namespaces a Substrate read covers. An atespace is named -// after the namespace of the agent it serves, so the same set bounds actors and templates. type substrateScope struct { all bool namespaces []string } -// authorizeSubstrateScope authorizes the caller and resolves the scope of a read. func (s *Service) authorizeSubstrateScope(ctx context.Context, requested string) (substrateScope, error) { if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: substrateResource}); err != nil { return substrateScope{}, err @@ -273,8 +274,6 @@ func (s *Service) authorizeSubstrateScope(ctx context.Context, requested string) return s.resolveSubstrateScope(requested) } -// resolveSubstrateScope narrows the watched namespaces to the requested one. A -// controller that watches only some namespaces refuses any other. func (s *Service) resolveSubstrateScope(requested string) (substrateScope, error) { switch { case requested == "" && len(s.observedNamespaces) == 0: @@ -284,7 +283,7 @@ func (s *Service) resolveSubstrateScope(requested string) (substrateScope, error case len(s.observedNamespaces) == 0 || slices.Contains(s.observedNamespaces, requested): return substrateScope{namespaces: []string{requested}}, nil default: - return substrateScope{}, serviceerrors.NewPermissionDenied("Not authorized", fmt.Errorf("namespace %q is not watched by this controller", requested)) + return substrateScope{}, serviceerrors.NewPermissionDenied(fmt.Sprintf("Namespace %q is not watched by this controller", requested), nil) } } @@ -292,25 +291,6 @@ func (scope substrateScope) contains(namespace string) bool { return scope.all || slices.Contains(scope.namespaces, namespace) } -// containsWorker also admits a worker that reports no namespace. -func (scope substrateScope) containsWorker(worker *ateapipb.Worker) bool { - if worker == nil { - return false - } - namespace := strings.TrimSpace(worker.GetWorkerNamespace()) - return namespace == "" || scope.contains(namespace) -} - -// atespace is the atespace to ask ate-api for: the only one in scope, or every -// atespace ("") for the caller to filter. -func (scope substrateScope) atespace() string { - if len(scope.namespaces) == 1 { - return scope.namespaces[0] - } - return "" -} - -// listNamespaces is the namespaces to list Kubernetes objects in, "" meaning all. func (scope substrateScope) listNamespaces() []string { if scope.all { return []string{""} From 14eee7c3e234f3340a449bdf1808f9255f3f0248 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Mon, 19 Oct 2026 10:51:23 -0400 Subject: [PATCH 05/16] test(substrate): cover worker pools and a namespace watched twice The isolation test built its fake cluster with no WorkerPools, so it never checked that the summary lists only pools in the watched namespaces. It also never listed a namespace twice in watchNamespaces, which counted tenant-a's actors, templates and pools twice. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 28 ++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index fbded3fced..84e0dfce4e 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -748,11 +748,17 @@ func TestSubstrateTenantIsolation(t *testing.T) { {WorkerPod: "pod-unlabelled"}, }, } + kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + &atev1alpha1.WorkerPool{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant-a", Name: "pool-a"}}, + &atev1alpha1.WorkerPool{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant-b", Name: "pool-b"}}, + &atev1alpha1.WorkerPool{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant-c", Name: "pool-c"}}, + ).Build() newService := func(observedNamespaces []string) *system.Service { - return system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), observedNamespaces, &pkgAuth.NoopAuthorizer{}, ateClient) + return system.NewService(kubeClient, observedNamespaces, &pkgAuth.NoopAuthorizer{}, ateClient) } tenantA := newService([]string{"tenant-a"}) tenantsAC := newService([]string{"tenant-a", "tenant-c"}) + tenantsACDuplicated := newService([]string{"tenant-a", "tenant-c", "tenant-a"}) clusterWide := newService(nil) listAllActors := func(t *testing.T, service *system.Service, atespace string) []string { @@ -785,6 +791,13 @@ func TestSubstrateTenantIsolation(t *testing.T) { request.Page.PageToken = page.NextPageToken } } + poolNames := func(summary system.SubstrateSummary) []string { + var names []string + for _, pool := range summary.WorkerPools { + names = append(names, pool.Namespace+"/"+pool.Name) + } + return names + } templateNames := func(summary system.SubstrateSummary) []string { var names []string for _, template := range summary.ActorTemplates { @@ -802,6 +815,7 @@ func TestSubstrateTenantIsolation(t *testing.T) { require.NoError(t, err) require.Empty(t, summary.ATEAPIError) assert.Equal(t, []string{"agent-a"}, templateNames(summary)) + assert.Equal(t, []string{"tenant-a/pool-a"}, poolNames(summary)) assert.Equal(t, int64(1), summary.ActorCount) assert.Equal(t, int64(0), summary.RunningActorCount) assert.Equal(t, int64(1), summary.WorkerCount) @@ -833,12 +847,23 @@ func TestSubstrateTenantIsolation(t *testing.T) { {State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, Count: 1}, }, summary.ActorStatusCounts) assert.Equal(t, []string{"agent-a", "agent-c"}, templateNames(summary)) + assert.Equal(t, []string{"tenant-a/pool-a", "tenant-c/pool-c"}, poolNames(summary)) assert.Equal(t, int64(2), summary.WorkerCount) assert.Equal(t, []string{"tenant-a", "tenant-c"}, slices.Compact(slices.Sorted(slices.Values(ateClient.actorAtespaces)))) assert.ElementsMatch(t, []string{"tenant-a", "tenant-c"}, ateClient.templateAtespaces) }) + t.Run("a namespace listed twice in the config is counted once", func(t *testing.T) { + summary, err := tenantsACDuplicated.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + require.Empty(t, summary.ATEAPIError) + assert.Equal(t, int64(2), summary.ActorCount) + assert.Equal(t, []string{"agent-a", "agent-c"}, templateNames(summary)) + assert.Equal(t, []string{"tenant-a/pool-a", "tenant-c/pool-c"}, poolNames(summary)) + assert.Equal(t, int64(2), summary.WorkerCount) + }) + t.Run("refuses another tenant's atespace or namespace", func(t *testing.T) { _, err := tenantA.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) @@ -856,6 +881,7 @@ func TestSubstrateTenantIsolation(t *testing.T) { require.NoError(t, err) assert.Equal(t, int64(4), summary.ActorCount) assert.Len(t, summary.ActorTemplates, 3) + assert.Equal(t, []string{"tenant-a/pool-a", "tenant-b/pool-b", "tenant-c/pool-c"}, poolNames(summary)) assert.Equal(t, int64(4), summary.WorkerCount) assert.Equal(t, []string{"pod-a", "pod-b", "pod-c", "pod-unlabelled"}, workerPods(t, clusterWide)) assert.Equal(t, []string{"tenant-b/session-b", "tenant-b/session-b2"}, listAllActors(t, clusterWide, "tenant-b")) From 470714230da6c32dea1d1aafd4e5a936853371c9 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Mon, 19 Oct 2026 10:51:23 -0400 Subject: [PATCH 06/16] fix(substrate): count a namespace watched twice once watchNamespaces allows duplicates, and every Substrate read asks ate-api and Kubernetes once per watched namespace, so a duplicate entry doubled the summary's actors, templates and worker pools. Drop duplicates once when the service is built, keeping the configured order. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index 89ae8439c2..6a3f8a60a3 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -49,9 +49,15 @@ func NewService( authorizer auth.CollectionAuthorizer, ateClient ATEClient, ) *Service { + var observed []string + for _, namespace := range observedNamespaces { + if !slices.Contains(observed, namespace) { + observed = append(observed, namespace) + } + } return &Service{ kubeClient: kubeClient, - observedNamespaces: slices.Clone(observedNamespaces), + observedNamespaces: observed, authorizer: authorizer, ateClient: ateClient, } From 21afb4c73180a26c14900fd0c748c78e5d0eb466 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Mon, 19 Oct 2026 10:51:23 -0400 Subject: [PATCH 07/16] docs(api): say an empty atespace means the watched namespaces GetSubstrateSummaryRequest.atespace and ListSubstrateActorsRequest.atespace still said an empty value lists all atespaces. It now means the watched namespaces unless the controller is cluster-wide, and an unwatched atespace returns PermissionDenied. Signed-off-by: Danielle Mustillo --- go/api/gen/kagent/api/v1/system.pb.go | 6 ++++-- proto/kagent/api/v1/system.proto | 6 ++++-- ui/src/generated/kagent/api/v1/system_pb.ts | 6 ++++-- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/go/api/gen/kagent/api/v1/system.pb.go b/go/api/gen/kagent/api/v1/system.pb.go index 1dd3795de6..7fd5f28bd8 100644 --- a/go/api/gen/kagent/api/v1/system.pb.go +++ b/go/api/gen/kagent/api/v1/system.pb.go @@ -396,7 +396,8 @@ type GetSubstrateSummaryRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // Kubernetes namespace for workers and pools. Empty uses watched namespaces. Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` - // ATE atespace for actors and templates. Empty lists all atespaces. + // ATE atespace for actors and templates. Empty uses the watched namespaces, or every + // atespace on a cluster-wide controller. An unwatched atespace returns PermissionDenied. Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -617,7 +618,8 @@ func (x *GetSubstrateSummaryResponse) GetActorTemplates() []*ateapipb.ActorTempl // One upstream page of actors, in Substrate order. type ListSubstrateActorsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` - // ATE atespace of the actor itself. Empty lists all atespaces. + // ATE atespace of the actor itself. Empty uses the watched namespaces, or every + // atespace on a cluster-wide controller. An unwatched atespace returns PermissionDenied. Atespace string `protobuf:"bytes,6,opt,name=atespace,proto3" json:"atespace,omitempty"` // `limit` is capped at 100 in common.proto and zero means the server's default; // above the cap is refused rather than clamped. diff --git a/proto/kagent/api/v1/system.proto b/proto/kagent/api/v1/system.proto index b8e7cbecdf..863ffff3bf 100644 --- a/proto/kagent/api/v1/system.proto +++ b/proto/kagent/api/v1/system.proto @@ -62,7 +62,8 @@ message GetSubstrateSummaryRequest { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" }]; - // ATE atespace for actors and templates. Empty lists all atespaces. + // ATE atespace for actors and templates. Empty uses the watched namespaces, or every + // atespace on a cluster-wide controller. An unwatched atespace returns PermissionDenied. string atespace = 2 [(buf.validate.field).string = { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" @@ -96,7 +97,8 @@ message GetSubstrateSummaryResponse { // One upstream page of actors, in Substrate order. message ListSubstrateActorsRequest { - // ATE atespace of the actor itself. Empty lists all atespaces. + // ATE atespace of the actor itself. Empty uses the watched namespaces, or every + // atespace on a cluster-wide controller. An unwatched atespace returns PermissionDenied. string atespace = 6 [(buf.validate.field).string = { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" diff --git a/ui/src/generated/kagent/api/v1/system_pb.ts b/ui/src/generated/kagent/api/v1/system_pb.ts index 504a6cdba8..24b8b86279 100644 --- a/ui/src/generated/kagent/api/v1/system_pb.ts +++ b/ui/src/generated/kagent/api/v1/system_pb.ts @@ -184,7 +184,8 @@ export type GetSubstrateSummaryRequest = Message<"kagent.api.v1.GetSubstrateSumm namespace: string; /** - * ATE atespace for actors and templates. Empty lists all atespaces. + * ATE atespace for actors and templates. Empty uses the watched namespaces, or every + * atespace on a cluster-wide controller. An unwatched atespace returns PermissionDenied. * * @generated from field: string atespace = 2; */ @@ -297,7 +298,8 @@ export const GetSubstrateSummaryResponseSchema: GenMessage & { /** - * ATE atespace of the actor itself. Empty lists all atespaces. + * ATE atespace of the actor itself. Empty uses the watched namespaces, or every + * atespace on a cluster-wide controller. An unwatched atespace returns PermissionDenied. * * @generated from field: string atespace = 6; */ From 11974a7c0fa0d77c491c6a51615fe83163f895d9 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Mon, 19 Oct 2026 12:21:03 -0400 Subject: [PATCH 08/16] refactor(substrate): dedupe watched namespaces with slices.Compact Sorting before Compact removes duplicates that are not adjacent. ListNamespaces already sorts its output, so the order change is not visible to callers. Verified with go test ./core/internal/service/system, including the case that watches [tenant-a, tenant-c, tenant-a]. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index 6a3f8a60a3..6bd2446420 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -49,15 +49,9 @@ func NewService( authorizer auth.CollectionAuthorizer, ateClient ATEClient, ) *Service { - var observed []string - for _, namespace := range observedNamespaces { - if !slices.Contains(observed, namespace) { - observed = append(observed, namespace) - } - } return &Service{ kubeClient: kubeClient, - observedNamespaces: observed, + observedNamespaces: slices.Compact(slices.Sorted(slices.Values(observedNamespaces))), authorizer: authorizer, ateClient: ateClient, } From 343f89d3b432a19386659749d9608b60025ffe37 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:44:04 -0400 Subject: [PATCH 09/16] test(substrate): actor page tokens must name only watched atespaces A controller watching tenant-a and tenant-c must page through its own actors in watched-atespace order, scope every ate-api read to one watched atespace, never hand out an empty page with a token, and never return a token that names a tenant-b actor. A token it did not issue is InvalidArgument, without echoing the token and without an ate-api call. The cluster-wide pass-through test now runs cluster-wide, and the assertion that a filtered-out first page still leads on is dropped: that page is the leak. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 9 +- .../service/system/substrate_paging_test.go | 122 ++++++++++++++++++ 2 files changed, 124 insertions(+), 7 deletions(-) create mode 100644 go/core/internal/service/system/substrate_paging_test.go diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 84e0dfce4e..972ad2a991 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -602,14 +602,14 @@ func (c *emptyPageATEClient) ListWorkersPage(_ context.Context, size int32, toke func TestSubstrateEmptyPagesPreserveUpstreamTokens(t *testing.T) { ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) - service := system.NewService(nil, []string{"team"}, &pkgAuth.NoopAuthorizer{}, &emptyPageATEClient{t: t}) + service := system.NewService(nil, nil, &pkgAuth.NoopAuthorizer{}, &emptyPageATEClient{t: t}) page := &apiv1.PageRequest{Limit: 7, PageToken: "upstream:opaque/cursor="} actors, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "team", Page: page}) require.NoError(t, err) assert.Empty(t, actors.Actors) assert.Empty(t, actors.ATEAPIError) assert.Equal(t, "upstream:next/cursor=", actors.NextPageToken) - workers, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Page: page}) + workers, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "team", Page: page}) require.NoError(t, err) assert.Empty(t, workers.Workers) assert.Empty(t, workers.ATEAPIError) @@ -828,11 +828,6 @@ func TestSubstrateTenantIsolation(t *testing.T) { t.Run("a controller watching two namespaces lists only those atespaces", func(t *testing.T) { assert.Equal(t, []string{"tenant-a/session-a", "tenant-c/session-c"}, listAllActors(t, tenantsAC, "")) - - first, err := tenantsAC.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 1}}) - require.NoError(t, err) - assert.Empty(t, first.Actors, "the first upstream page holds only a tenant-b actor") - assert.NotEmpty(t, first.NextPageToken, "a page filtered to nothing still leads to the rest") }) t.Run("a controller watching two namespaces asks ate-api for only those atespaces", func(t *testing.T) { diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go new file mode 100644 index 0000000000..1fe305b197 --- /dev/null +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -0,0 +1,122 @@ +package system_test + +import ( + "context" + "encoding/base64" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.ate.dev/substrate/pkg/proto/ateapipb" + + apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" + authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" + "github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors" + "github.com/kagent-dev/kagent/go/core/internal/service/system" + pkgAuth "github.com/kagent-dev/kagent/go/core/pkg/auth" +) + +func pagingContext(t *testing.T) context.Context { + return pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) +} + +// pagingATEClient mixes three tenants' rows in ate-api order, two rows a page. Every +// tenant-b name holds "tenant-b", so a test can look for it in any text. +func pagingATEClient() *fakeATEClient { + client := &fakeATEClient{pageSize: 2} + for _, actor := range []string{"tenant-a/a-1", "tenant-b/b-1", "tenant-c/c-1", "tenant-a/a-2", "tenant-b/b-2", "tenant-c/c-2", "tenant-a/a-3"} { + atespace, name, _ := strings.Cut(actor, "/") + client.actors = append(client.actors, substrateActor(name, atespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) + } + return client +} + +// pagingService watches tenant-a and tenant-c, listed out of order and with a repeat. +func pagingService(ateClient system.ATEClient) *system.Service { + return system.NewService(nil, []string{"tenant-c", "tenant-a", "tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) +} + +// assertOwnToken fails when a page token names a tenant-b row, or when an empty page +// carries a token: that would tell the caller a hidden row exists. +func assertOwnToken(t *testing.T, rows int, pageToken string) { + t.Helper() + raw, _ := base64.RawURLEncoding.DecodeString(pageToken) + assert.NotContains(t, pageToken+string(raw), "tenant-b") + if rows == 0 { + assert.Empty(t, pageToken, "an empty page must not lead on") + } +} + +func walkActorNames(t *testing.T, service *system.Service, atespace string, limit int32) []string { + t.Helper() + names := []string{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + assertOwnToken(t, len(page.Actors), page.NextPageToken) + names = append(names, actorNames(page.Actors)...) + if pageToken = page.NextPageToken; pageToken == "" { + return names + } + } + t.Fatal("the walk did not end") + return nil +} + +func TestListSubstrateActorsPagesWatchedAtespaces(t *testing.T) { + for atespace, want := range map[string][]string{ + "": {"a-1", "a-2", "a-3", "c-1", "c-2"}, + "tenant-c": {"c-1", "c-2"}, + } { + for _, limit := range []int32{1, 2, 3, 4, 100} { + t.Run(fmt.Sprintf("%q/%d", atespace, limit), func(t *testing.T) { + ateClient := pagingATEClient() + assert.Equal(t, want, walkActorNames(t, pagingService(ateClient), atespace, limit)) + assert.Subset(t, []string{"tenant-a", "tenant-c"}, ateClient.actorAtespaces, "every ate-api read is scoped to a watched atespace") + }) + } + } +} + +func actorNames(actors []*ateapipb.Actor) []string { + names := []string{} + for _, actor := range actors { + names = append(names, actor.GetMetadata().GetName()) + } + return names +} + +func actorPageToken(raw string) string { + return base64.RawURLEncoding.EncodeToString([]byte(raw)) +} + +// A token this controller did not issue never reaches ate-api and never restarts the +// walk at page one. Its error repeats neither the token nor a tenant-b name. +func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { + for name, tc := range map[string]struct { + atespace string + pageToken string + }{ + "not base64": {pageToken: "upstream:2"}, + "no separator": {pageToken: actorPageToken("tenant-a")}, + "empty atespace": {pageToken: actorPageToken("/upstream:2")}, + "unwatched atespace": {pageToken: actorPageToken("tenant-b/upstream:2")}, + "outside the request": {atespace: "tenant-c", pageToken: actorPageToken("tenant-a/upstream:2")}, + "ate-api actor token": {pageToken: actorPageToken(`{"v":1,"kind":"actor","last":["tenant-b","b-1"]}`)}, + "over 1 KiB, else good": {pageToken: actorPageToken("tenant-a/" + strings.Repeat("x", 800))}, + } { + t.Run(name, func(t *testing.T) { + ateClient := pagingATEClient() + _, err := pagingService(ateClient).ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: tc.atespace, Page: &apiv1.PageRequest{Limit: 2, PageToken: tc.pageToken}}) + require.Error(t, err) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + assert.NotContains(t, err.Error(), "tenant-b") + assert.NotContains(t, err.Error(), tc.pageToken) + assert.Zero(t, ateClient.actorReads) + }) + } +} From 58d2e87bdf32b06cd4f5f222255fb8a8170adff4 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:44:28 -0400 Subject: [PATCH 10/16] fix(substrate): page actors through watched atespaces only With two or more watched namespaces, ListSubstrateActors asked ate-api for every atespace and passed its page token through. ate-api tokens are readable base64 JSON holding the last row's atespace and name, so a token could name another tenant's actor, and a page filtered to nothing still carried a token. A controller that watches some namespaces now reads each watched atespace with its own scoped list, in sorted order, and returns its own token: the atespace to read and ate-api's token inside it. A token it did not issue, or one longer than 1 KiB, is InvalidArgument and never reaches ate-api. NewService sorts and dedupes the watched list, since tokens resume in its order. A cluster-wide controller still passes ate-api's page and token through. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 3 +- go/core/internal/service/system/substrate.go | 66 ++++++++++++++++++-- 2 files changed, 63 insertions(+), 6 deletions(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index 6bd2446420..53384bc6aa 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -32,7 +32,8 @@ type ATEClient interface { } type Service struct { - kubeClient client.Client + kubeClient client.Client + // Sorted and unique: actor page tokens resume in this order. Empty means every namespace. observedNamespaces []string authorizer auth.CollectionAuthorizer ateClient ATEClient diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index dc020ac3d6..7b0d2d7d9c 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -2,6 +2,7 @@ package system import ( "context" + "encoding/base64" "fmt" "maps" "slices" @@ -170,18 +171,28 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return result, nil } -// ListSubstrateActors filters one upstream page by atespace and preserves its order and token. +// ListSubstrateActors returns one page of actors. A cluster-wide controller passes +// ate-api's page and token through. A controller that watches some namespaces reads +// each watched atespace with its own scoped list, so no page token names an actor of +// a namespace it does not watch. func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubstrateActorsRequest) (SubstrateActorPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetAtespace()) if err != nil { return SubstrateActorPage{}, err } - atespace := "" - if len(scope.namespaces) == 1 { - atespace = scope.namespaces[0] + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + var actors []*ateapipb.Actor + var next string + if len(s.observedNamespaces) == 0 { + actors, next, err = s.ateClient.ListActorsPage(ctx, input.GetAtespace(), limit, token) + } else { + start, inner, tokenErr := parseActorPageToken(scope.namespaces, token) + if tokenErr != nil { + return SubstrateActorPage{}, tokenErr + } + actors, next, err = s.listScopedActors(ctx, scope.namespaces[start:], inner, limit) } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, atespace, substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) @@ -196,6 +207,51 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs return result, nil } +// The longest actor page token accepted. This controller's own tokens are far shorter. +const maxActorPageTokenBytes = 1024 + +// actorPageToken is the actor page token of a controller that watches some +// namespaces: the atespace to read and ate-api's token inside it. An atespace holds no "/". +func actorPageToken(atespace, inner string) string { + return base64.RawURLEncoding.EncodeToString([]byte(atespace + "/" + inner)) +} + +// parseActorPageToken returns where in atespaces a page starts, and ate-api's token there. +func parseActorPageToken(atespaces []string, token string) (int, string, error) { + if token == "" { + return 0, "", nil + } + decoded, err := base64.RawURLEncoding.DecodeString(token) + atespace, inner, ok := strings.Cut(string(decoded), "/") + start := slices.Index(atespaces, atespace) + if len(token) > maxActorPageTokenBytes || err != nil || !ok || start < 0 { + // The token is the caller's text and can name anything, so the error omits it. + return 0, "", serviceerrors.NewInvalidArgument("Invalid page token", nil) + } + return start, inner, nil +} + +// listScopedActors fills one page from the atespaces in order, each with its own scoped +// list, starting at ate-api's inner token in the first. +func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inner string, limit int32) ([]*ateapipb.Actor, string, error) { + var actors []*ateapipb.Actor + for i, atespace := range atespaces { + rows, next, err := s.ateClient.ListActorsPage(ctx, atespace, limit-int32(len(actors)), inner) + if err != nil { + return nil, "", err + } + actors = append(actors, rows...) + switch { + case next != "": + return actors, actorPageToken(atespace, next), nil + case int32(len(actors)) >= limit && i+1 < len(atespaces): + return actors, actorPageToken(atespaces[i+1], ""), nil + } + inner = "" + } + return actors, "", nil +} + // ListSubstrateWorkers filters one upstream page by namespace and preserves its token. // A page with no workers in scope may still have a next page. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { From 3db786198b278001b917cbd9ad20d31f27907702 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:44:59 -0400 Subject: [PATCH 11/16] test(substrate): worker page tokens and errors must name only watched workers A controller watching tenant-a and tenant-c must page its own workers once each, in worker name order whatever order ate-api sends, never hand out an empty page with a token, and never send its own token to ate-api. An ate-api worker list error must reach the caller as its gRPC code only, since the text can name another tenant's worker. The tenant isolation fixtures now name their workers, as ate-api does. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 8 +-- .../service/system/substrate_paging_test.go | 52 +++++++++++++++++++ 2 files changed, 56 insertions(+), 4 deletions(-) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 972ad2a991..9dfd67a5dd 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -742,10 +742,10 @@ func TestSubstrateTenantIsolation(t *testing.T) { {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-c", Name: "agent-c"}}, }, workers: []*ateapipb.Worker{ - {WorkerNamespace: "tenant-a", WorkerPod: "pod-a"}, - {WorkerNamespace: "tenant-b", WorkerPod: "pod-b"}, - {WorkerNamespace: "tenant-c", WorkerPod: "pod-c"}, - {WorkerPod: "pod-unlabelled"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-a"}, WorkerNamespace: "tenant-a", WorkerPod: "pod-a"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-b"}, WorkerNamespace: "tenant-b", WorkerPod: "pod-b"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-c"}, WorkerNamespace: "tenant-c", WorkerPod: "pod-c"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-unlabelled"}, WorkerPod: "pod-unlabelled"}, }, } kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects( diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go index 1fe305b197..bc51cb18c0 100644 --- a/go/core/internal/service/system/substrate_paging_test.go +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -10,6 +10,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.ate.dev/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" @@ -30,6 +32,10 @@ func pagingATEClient() *fakeATEClient { atespace, name, _ := strings.Cut(actor, "/") client.actors = append(client.actors, substrateActor(name, atespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) } + for _, worker := range []string{"tenant-a/u4", "tenant-b/u2", "tenant-a/u1", "tenant-c/u3", "tenant-a/u6", "tenant-b/u5"} { + namespace, uid, _ := strings.Cut(worker, "/") + client.workers = append(client.workers, &ateapipb.Worker{Metadata: &ateapipb.ResourceMetadata{Name: uid + "-" + namespace}, WorkerNamespace: namespace}) + } return client } @@ -120,3 +126,49 @@ func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { }) } } + +func walkWorkerNames(t *testing.T, service *system.Service, namespace string, limit int32) []string { + t.Helper() + names := []string{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: namespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + assertOwnToken(t, len(page.Workers), page.NextPageToken) + for _, worker := range page.Workers { + names = append(names, worker.GetMetadata().GetName()) + } + if pageToken = page.NextPageToken; pageToken == "" { + return names + } + } + t.Fatal("the walk did not end") + return nil +} + +// ate-api cannot list one namespace's workers, so its token can name any worker. The +// fake refuses a token it did not issue, so a kagent token sent to it fails the walk. +func TestListSubstrateWorkersPagesWatchedNamespaces(t *testing.T) { + for namespace, want := range map[string][]string{ + "": {"u1-tenant-a", "u3-tenant-c", "u4-tenant-a", "u6-tenant-a"}, + "tenant-a": {"u1-tenant-a", "u4-tenant-a", "u6-tenant-a"}, + } { + for _, limit := range []int32{1, 2, 3, 100} { + t.Run(fmt.Sprintf("%q/%d", namespace, limit), func(t *testing.T) { + assert.Equal(t, want, walkWorkerNames(t, pagingService(pagingATEClient()), namespace, limit)) + }) + } + } +} + +func TestListSubstrateWorkersHidesATEAPIErrorText(t *testing.T) { + ateClient := pagingATEClient() + ateClient.err = status.Error(codes.Internal, "worker u2-tenant-b is down") + + page, err := pagingService(ateClient).ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{}) + require.NoError(t, err) + assert.Equal(t, "ate-api could not list workers: Internal", page.ATEAPIError) + assert.Empty(t, page.Workers) + assert.Empty(t, page.NextPageToken) +} From b6407a360cde9faa232a9de5d358acf0c551443e Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:45:28 -0400 Subject: [PATCH 12/16] fix(substrate): page workers by name and hide ate-api worker errors A controller that watches some namespaces passed ate-api's worker page token through. The token holds the last worker's name, so it could name another tenant's worker, and a page filtered to nothing still carried a token. Such a controller now drains ate-api's workers, keeps the ones in scope, and pages them in worker name order. Its token is the last worker's name: a cursor only, never sent to ate-api. ate-api worker list errors reach the caller as the gRPC code only, for the list and the summary, since the text can name another tenant's worker; the full error is logged. A cluster-wide controller still passes ate-api's page, token and error through. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/substrate.go | 50 ++++++++++++++++++-- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 7b0d2d7d9c..5e08e555c4 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -11,6 +11,7 @@ import ( atev1alpha1 "go.ate.dev/substrate/pkg/api/v1alpha1" "go.ate.dev/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/status" apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" "github.com/kagent-dev/kagent/go/core/internal/service/kubeauth" @@ -252,15 +253,21 @@ func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inne return actors, "", nil } -// ListSubstrateWorkers filters one upstream page by namespace and preserves its token. -// A page with no workers in scope may still have a next page. +// ListSubstrateWorkers returns one page of workers. A cluster-wide controller filters +// ate-api's page and passes its token through. A controller that watches some +// namespaces pages its own workers with its own token, since ate-api's can name any +// worker. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetNamespace()) if err != nil { return SubstrateWorkerPage{}, err } + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + if len(s.observedNamespaces) > 0 { + return s.listWatchedWorkers(ctx, scope, limit, token), nil + } result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} - workers, next, err := s.ateClient.ListWorkersPage(ctx, substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + workers, next, err := s.ateClient.ListWorkersPage(ctx, limit, token) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) @@ -275,6 +282,31 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } +// listWatchedWorkers fills one page with the workers in scope, in worker name order. +// The token is the name of the last worker on the previous page: a cursor only, never +// sent to ate-api. +func (s *Service) listWatchedWorkers(ctx context.Context, scope substrateScope, limit int32, after string) SubstrateWorkerPage { + result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} + var workers []*ateapipb.Worker + if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { + if scope.contains(worker.GetWorkerNamespace()) && worker.GetMetadata().GetName() > after { + workers = append(workers, worker) + } + }); err != nil { + result.ATEAPIError = err.Error() + return result + } + slices.SortFunc(workers, func(left, right *ateapipb.Worker) int { + return strings.Compare(left.GetMetadata().GetName(), right.GetMetadata().GetName()) + }) + if int32(len(workers)) > limit { + workers = workers[:limit] + result.NextPageToken = workers[limit-1].GetMetadata().GetName() + } + result.Workers = workers + return result +} + func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit func(*ateapipb.Actor)) error { for _, atespace := range scope.listNamespaces() { read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { @@ -288,8 +320,18 @@ func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit fu } // walkWorkers calls visit for every worker ate-api holds, one page at a time. +// +// NOTE: ate-api cannot list one namespace's workers, so a controller that watches some +// drains every tenant's workers per request. Use an ate-api worker namespace filter +// when one lands. func (s *Service) walkWorkers(ctx context.Context, visit func(*ateapipb.Worker)) error { - return walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) + err := walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) + if err != nil && len(s.observedNamespaces) > 0 { + // ate-api's error text can name another tenant's worker, so the caller gets the code. + logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) + return fmt.Errorf("ate-api could not list workers: %s", status.Code(err)) + } + return err } // walkSubstrate calls visit for every row ate-api holds, holding one page at a time From b29e09776c7b866aa20cdba9b85740b1ec11e197 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:46:20 -0400 Subject: [PATCH 13/16] docs(substrate): describe paging for controllers that watch some namespaces The proto comments, the generated Go and TypeScript, the UI page type and the Playwright deferral notes said an empty atespace lists all atespaces and that every page passes Substrate's token through. A controller that watches some namespaces lists only its watched atespaces, issues its own tokens, and never returns an empty page with a token. The atespace filter placeholder now reads "All watched atespaces". Signed-off-by: Danielle Mustillo --- go/api/gen/kagent/api/v1/system.pb.go | 13 +++++++++---- proto/kagent/api/v1/system.proto | 13 +++++++++---- ui/playwright/DEFERRED.md | 12 +++++++----- ui/src/api/domain/substrate.ts | 6 +++++- ui/src/generated/kagent/api/v1/system_pb.ts | 13 +++++++++---- ui/src/pages/SubstratePage.tsx | 2 +- 6 files changed, 40 insertions(+), 19 deletions(-) diff --git a/go/api/gen/kagent/api/v1/system.pb.go b/go/api/gen/kagent/api/v1/system.pb.go index 7fd5f28bd8..0a138d6ab0 100644 --- a/go/api/gen/kagent/api/v1/system.pb.go +++ b/go/api/gen/kagent/api/v1/system.pb.go @@ -615,7 +615,8 @@ func (x *GetSubstrateSummaryResponse) GetActorTemplates() []*ateapipb.ActorTempl return nil } -// One upstream page of actors, in Substrate order. +// One page of actors. A controller that watches some namespaces reads its watched +// atespaces one at a time, in name order, and Substrate order inside each. type ListSubstrateActorsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // ATE atespace of the actor itself. Empty uses the watched namespaces, or every @@ -676,7 +677,9 @@ type ListSubstrateActorsResponse struct { state protoimpl.MessageState `protogen:"open.v1"` AteApiError string `protobuf:"bytes,2,opt,name=ate_api_error,json=ateApiError,proto3" json:"ate_api_error,omitempty"` Actors []*ateapipb.Actor `protobuf:"bytes,3,rep,name=actors,proto3" json:"actors,omitempty"` - // Upstream continuation token. A page can be empty even when another page exists. + // Continuation token. A controller that watches all namespaces passes Substrate's + // token through. A controller that watches some issues its own, which names only + // watched data. Only a last page can be empty. Page *PageResponse `protobuf:"bytes,4,opt,name=page,proto3" json:"page,omitempty"` ComputedAt *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=computed_at,json=computedAt,proto3" json:"computed_at,omitempty"` unknownFields protoimpl.UnknownFields @@ -741,8 +744,10 @@ func (x *ListSubstrateActorsResponse) GetComputedAt() *timestamppb.Timestamp { return nil } -// One upstream page of workers, in Substrate order. -// Namespace filtering applies to this page only; it may be empty and still have a next token. +// One page of workers, in worker name order. A controller that watches all namespaces +// filters one Substrate page, which may be empty and still have a next token. A +// controller that watches some reads every Substrate worker and returns only those of +// its watched namespaces; only its last page can be empty. type ListSubstrateWorkersRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` diff --git a/proto/kagent/api/v1/system.proto b/proto/kagent/api/v1/system.proto index 863ffff3bf..2acddbca03 100644 --- a/proto/kagent/api/v1/system.proto +++ b/proto/kagent/api/v1/system.proto @@ -95,7 +95,8 @@ message GetSubstrateSummaryResponse { repeated ateapi.ActorTemplate actor_templates = 11; } -// One upstream page of actors, in Substrate order. +// One page of actors. A controller that watches some namespaces reads its watched +// atespaces one at a time, in name order, and Substrate order inside each. message ListSubstrateActorsRequest { // ATE atespace of the actor itself. Empty uses the watched namespaces, or every // atespace on a cluster-wide controller. An unwatched atespace returns PermissionDenied. @@ -111,13 +112,17 @@ message ListSubstrateActorsRequest { message ListSubstrateActorsResponse { string ate_api_error = 2; repeated ateapi.Actor actors = 3; - // Upstream continuation token. A page can be empty even when another page exists. + // Continuation token. A controller that watches all namespaces passes Substrate's + // token through. A controller that watches some issues its own, which names only + // watched data. Only a last page can be empty. PageResponse page = 4; google.protobuf.Timestamp computed_at = 5; } -// One upstream page of workers, in Substrate order. -// Namespace filtering applies to this page only; it may be empty and still have a next token. +// One page of workers, in worker name order. A controller that watches all namespaces +// filters one Substrate page, which may be empty and still have a next token. A +// controller that watches some reads every Substrate worker and returns only those of +// its watched namespaces; only its last page can be empty. message ListSubstrateWorkersRequest { string namespace = 1 [(buf.validate.field).string = { max_len: 63 diff --git a/ui/playwright/DEFERRED.md b/ui/playwright/DEFERRED.md index be664c4f12..530fd90a6f 100644 --- a/ui/playwright/DEFERRED.md +++ b/ui/playwright/DEFERRED.md @@ -219,11 +219,13 @@ when it holds one page of them, and only the proto says which. | `ListSubstrateActors` | `ListSubstrateActorsRequest { atespace, page }` | a page | `string filter` and a sort field enum — **and ate-api has to grow them first** | | `ListSubstrateWorkers` | `ListSubstrateWorkersRequest { namespace, page }` | a page | the same two, on the same condition | -**Substrate actor and worker lists use upstream pagination.** Each list request makes -one ate-api call and preserves its order and continuation token. Actors use upstream -atespace filtering. Worker namespace filtering applies only to the returned page, so an -empty worker page may still have a next token. The UI labels counts as "on this page" -and offers no actor/worker text search or column sorting. +**Substrate actor and worker lists use upstream pagination.** A controller that watches +every namespace makes one ate-api call per list request and passes its order and +continuation token through. Worker namespace filtering then applies only to the returned +page, so an empty worker page may still have a next token. A controller that watches some +namespaces reads actors with one scoped ate-api list per watched atespace and returns its +own token. It reads every ate-api worker page and returns a worker-name cursor. The UI +labels counts as "on this page" and offers no actor/worker text search or column sorting. Two capabilities remain deferred until Substrate supports them: diff --git a/ui/src/api/domain/substrate.ts b/ui/src/api/domain/substrate.ts index eff4e0034a..8e2a8c8a21 100644 --- a/ui/src/api/domain/substrate.ts +++ b/ui/src/api/domain/substrate.ts @@ -93,7 +93,11 @@ export interface Timed { computedAt?: string; } -/** One upstream page; an empty page may still have a continuation token. */ +/** + * One page. Only a last page is empty, except a workers page from a controller that + * watches all namespaces: it filters one upstream page, so it can be empty and still + * have a continuation token. + */ interface SubstratePage extends Timed { /** Upstream read failure. No rows or continuation token accompany it. */ ateApiError?: string; diff --git a/ui/src/generated/kagent/api/v1/system_pb.ts b/ui/src/generated/kagent/api/v1/system_pb.ts index 24b8b86279..7f7ead9ac3 100644 --- a/ui/src/generated/kagent/api/v1/system_pb.ts +++ b/ui/src/generated/kagent/api/v1/system_pb.ts @@ -292,7 +292,8 @@ export const GetSubstrateSummaryResponseSchema: GenMessage Date: Mon, 19 Oct 2026 00:18:32 -0400 Subject: [PATCH 14/16] docs(substrate): trim paging comments and correct empty-page wording Drop comments that restate the code. The proto now says that an actor page can be empty when another page exists, and gives the worker order for each mode. Signed-off-by: Danielle Mustillo --- go/api/gen/kagent/api/v1/system.pb.go | 8 +++--- go/core/internal/service/system/service.go | 3 +-- go/core/internal/service/system/substrate.go | 27 +++---------------- .../service/system/substrate_paging_test.go | 9 ------- proto/kagent/api/v1/system.proto | 8 +++--- ui/src/api/domain/substrate.ts | 5 ---- ui/src/generated/kagent/api/v1/system_pb.ts | 8 +++--- 7 files changed, 17 insertions(+), 51 deletions(-) diff --git a/go/api/gen/kagent/api/v1/system.pb.go b/go/api/gen/kagent/api/v1/system.pb.go index 0a138d6ab0..50123b05ab 100644 --- a/go/api/gen/kagent/api/v1/system.pb.go +++ b/go/api/gen/kagent/api/v1/system.pb.go @@ -679,7 +679,7 @@ type ListSubstrateActorsResponse struct { Actors []*ateapipb.Actor `protobuf:"bytes,3,rep,name=actors,proto3" json:"actors,omitempty"` // Continuation token. A controller that watches all namespaces passes Substrate's // token through. A controller that watches some issues its own, which names only - // watched data. Only a last page can be empty. + // watched data. A page can be empty even when another page exists. Page *PageResponse `protobuf:"bytes,4,opt,name=page,proto3" json:"page,omitempty"` ComputedAt *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=computed_at,json=computedAt,proto3" json:"computed_at,omitempty"` unknownFields protoimpl.UnknownFields @@ -744,10 +744,10 @@ func (x *ListSubstrateActorsResponse) GetComputedAt() *timestamppb.Timestamp { return nil } -// One page of workers, in worker name order. A controller that watches all namespaces -// filters one Substrate page, which may be empty and still have a next token. A +// One page of workers. A controller that watches all namespaces filters one Substrate +// page, in Substrate order, which may be empty and still have a next token. A // controller that watches some reads every Substrate worker and returns only those of -// its watched namespaces; only its last page can be empty. +// its watched namespaces, in worker name order; only its last page can be empty. type ListSubstrateWorkersRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index 53384bc6aa..6bd2446420 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -32,8 +32,7 @@ type ATEClient interface { } type Service struct { - kubeClient client.Client - // Sorted and unique: actor page tokens resume in this order. Empty means every namespace. + kubeClient client.Client observedNamespaces []string authorizer auth.CollectionAuthorizer ateClient ATEClient diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 5e08e555c4..62af51fd13 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -172,10 +172,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return result, nil } -// ListSubstrateActors returns one page of actors. A cluster-wide controller passes -// ate-api's page and token through. A controller that watches some namespaces reads -// each watched atespace with its own scoped list, so no page token names an actor of -// a namespace it does not watch. +// ListSubstrateActors returns one page of actors. func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubstrateActorsRequest) (SubstrateActorPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetAtespace()) if err != nil { @@ -208,16 +205,12 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs return result, nil } -// The longest actor page token accepted. This controller's own tokens are far shorter. const maxActorPageTokenBytes = 1024 -// actorPageToken is the actor page token of a controller that watches some -// namespaces: the atespace to read and ate-api's token inside it. An atespace holds no "/". func actorPageToken(atespace, inner string) string { return base64.RawURLEncoding.EncodeToString([]byte(atespace + "/" + inner)) } -// parseActorPageToken returns where in atespaces a page starts, and ate-api's token there. func parseActorPageToken(atespaces []string, token string) (int, string, error) { if token == "" { return 0, "", nil @@ -226,14 +219,11 @@ func parseActorPageToken(atespaces []string, token string) (int, string, error) atespace, inner, ok := strings.Cut(string(decoded), "/") start := slices.Index(atespaces, atespace) if len(token) > maxActorPageTokenBytes || err != nil || !ok || start < 0 { - // The token is the caller's text and can name anything, so the error omits it. return 0, "", serviceerrors.NewInvalidArgument("Invalid page token", nil) } return start, inner, nil } -// listScopedActors fills one page from the atespaces in order, each with its own scoped -// list, starting at ate-api's inner token in the first. func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inner string, limit int32) ([]*ateapipb.Actor, string, error) { var actors []*ateapipb.Actor for i, atespace := range atespaces { @@ -253,10 +243,7 @@ func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inne return actors, "", nil } -// ListSubstrateWorkers returns one page of workers. A cluster-wide controller filters -// ate-api's page and passes its token through. A controller that watches some -// namespaces pages its own workers with its own token, since ate-api's can name any -// worker. +// ListSubstrateWorkers returns one page of workers. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetNamespace()) if err != nil { @@ -282,9 +269,6 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } -// listWatchedWorkers fills one page with the workers in scope, in worker name order. -// The token is the name of the last worker on the previous page: a cursor only, never -// sent to ate-api. func (s *Service) listWatchedWorkers(ctx context.Context, scope substrateScope, limit int32, after string) SubstrateWorkerPage { result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} var workers []*ateapipb.Worker @@ -320,14 +304,11 @@ func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit fu } // walkWorkers calls visit for every worker ate-api holds, one page at a time. -// -// NOTE: ate-api cannot list one namespace's workers, so a controller that watches some -// drains every tenant's workers per request. Use an ate-api worker namespace filter -// when one lands. +// ate-api's ListWorkersRequest has no namespace filter, so a controller that watches +// some namespaces still reads every tenant's workers. func (s *Service) walkWorkers(ctx context.Context, visit func(*ateapipb.Worker)) error { err := walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) if err != nil && len(s.observedNamespaces) > 0 { - // ate-api's error text can name another tenant's worker, so the caller gets the code. logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) return fmt.Errorf("ate-api could not list workers: %s", status.Code(err)) } diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go index bc51cb18c0..fbde15e456 100644 --- a/go/core/internal/service/system/substrate_paging_test.go +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -24,8 +24,6 @@ func pagingContext(t *testing.T) context.Context { return pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) } -// pagingATEClient mixes three tenants' rows in ate-api order, two rows a page. Every -// tenant-b name holds "tenant-b", so a test can look for it in any text. func pagingATEClient() *fakeATEClient { client := &fakeATEClient{pageSize: 2} for _, actor := range []string{"tenant-a/a-1", "tenant-b/b-1", "tenant-c/c-1", "tenant-a/a-2", "tenant-b/b-2", "tenant-c/c-2", "tenant-a/a-3"} { @@ -39,13 +37,10 @@ func pagingATEClient() *fakeATEClient { return client } -// pagingService watches tenant-a and tenant-c, listed out of order and with a repeat. func pagingService(ateClient system.ATEClient) *system.Service { return system.NewService(nil, []string{"tenant-c", "tenant-a", "tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) } -// assertOwnToken fails when a page token names a tenant-b row, or when an empty page -// carries a token: that would tell the caller a hidden row exists. func assertOwnToken(t *testing.T, rows int, pageToken string) { t.Helper() raw, _ := base64.RawURLEncoding.DecodeString(pageToken) @@ -100,8 +95,6 @@ func actorPageToken(raw string) string { return base64.RawURLEncoding.EncodeToString([]byte(raw)) } -// A token this controller did not issue never reaches ate-api and never restarts the -// walk at page one. Its error repeats neither the token nor a tenant-b name. func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { for name, tc := range map[string]struct { atespace string @@ -147,8 +140,6 @@ func walkWorkerNames(t *testing.T, service *system.Service, namespace string, li return nil } -// ate-api cannot list one namespace's workers, so its token can name any worker. The -// fake refuses a token it did not issue, so a kagent token sent to it fails the walk. func TestListSubstrateWorkersPagesWatchedNamespaces(t *testing.T) { for namespace, want := range map[string][]string{ "": {"u1-tenant-a", "u3-tenant-c", "u4-tenant-a", "u6-tenant-a"}, diff --git a/proto/kagent/api/v1/system.proto b/proto/kagent/api/v1/system.proto index 2acddbca03..2baa8e1652 100644 --- a/proto/kagent/api/v1/system.proto +++ b/proto/kagent/api/v1/system.proto @@ -114,15 +114,15 @@ message ListSubstrateActorsResponse { repeated ateapi.Actor actors = 3; // Continuation token. A controller that watches all namespaces passes Substrate's // token through. A controller that watches some issues its own, which names only - // watched data. Only a last page can be empty. + // watched data. A page can be empty even when another page exists. PageResponse page = 4; google.protobuf.Timestamp computed_at = 5; } -// One page of workers, in worker name order. A controller that watches all namespaces -// filters one Substrate page, which may be empty and still have a next token. A +// One page of workers. A controller that watches all namespaces filters one Substrate +// page, in Substrate order, which may be empty and still have a next token. A // controller that watches some reads every Substrate worker and returns only those of -// its watched namespaces; only its last page can be empty. +// its watched namespaces, in worker name order; only its last page can be empty. message ListSubstrateWorkersRequest { string namespace = 1 [(buf.validate.field).string = { max_len: 63 diff --git a/ui/src/api/domain/substrate.ts b/ui/src/api/domain/substrate.ts index 8e2a8c8a21..6755d02d1b 100644 --- a/ui/src/api/domain/substrate.ts +++ b/ui/src/api/domain/substrate.ts @@ -93,11 +93,6 @@ export interface Timed { computedAt?: string; } -/** - * One page. Only a last page is empty, except a workers page from a controller that - * watches all namespaces: it filters one upstream page, so it can be empty and still - * have a continuation token. - */ interface SubstratePage extends Timed { /** Upstream read failure. No rows or continuation token accompany it. */ ateApiError?: string; diff --git a/ui/src/generated/kagent/api/v1/system_pb.ts b/ui/src/generated/kagent/api/v1/system_pb.ts index 7f7ead9ac3..9d5bbbba3a 100644 --- a/ui/src/generated/kagent/api/v1/system_pb.ts +++ b/ui/src/generated/kagent/api/v1/system_pb.ts @@ -339,7 +339,7 @@ export type ListSubstrateActorsResponse = Message<"kagent.api.v1.ListSubstrateAc /** * Continuation token. A controller that watches all namespaces passes Substrate's * token through. A controller that watches some issues its own, which names only - * watched data. Only a last page can be empty. + * watched data. A page can be empty even when another page exists. * * @generated from field: kagent.api.v1.PageResponse page = 4; */ @@ -359,10 +359,10 @@ export const ListSubstrateActorsResponseSchema: GenMessage Date: Mon, 19 Oct 2026 00:42:46 -0400 Subject: [PATCH 15/16] test(substrate): expose row keys in fake page tokens and probe forged tokens The fake ate-api now hands out base64 JSON tokens naming the page's last row, as ate-api does, so assertOwnToken can see a token that would name another tenant's row. New tests: scoped worker page tokens must be refused unless kagent issued them, a scoped summary hides ate-api worker error text, forged tokens never reach or return an unwatched namespace, and the longest names fit the page token. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 56 +++++--- .../service/system/substrate_paging_test.go | 133 ++++++++++++++++-- 2 files changed, 163 insertions(+), 26 deletions(-) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 9dfd67a5dd..8e52dd107b 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -2,11 +2,10 @@ package system_test import ( "context" + "encoding/base64" + "encoding/json" "errors" - "fmt" "slices" - "strconv" - "strings" "testing" "github.com/stretchr/testify/assert" @@ -77,7 +76,7 @@ func (client *fakeATEClient) ListActorsPage(_ context.Context, atespace string, if err := client.readError(client.actorReads); err != nil { return nil, "", err } - return fakePage(actorsInAtespace(client.actors, atespace), client.pageSize, pageSize, pageToken) + return fakePage(actorsInAtespace(client.actors, atespace), actorKey, client.pageSize, pageSize, pageToken) } func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, pageToken string) ([]*ateapipb.Worker, string, error) { @@ -85,42 +84,61 @@ func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, if err := client.readError(client.workerReads); err != nil { return nil, "", err } - return fakePage(client.workers, client.pageSize, pageSize, pageToken) + return fakePage(client.workers, workerKey, client.pageSize, pageSize, pageToken) } func (client *fakeATEClient) readError(int) error { return client.err } -// fakePage slices rows the way ate-api pages them: an opaque token, empty on the last -// page, and never more rows than its own ceiling however many were asked for. -func fakePage[T any](rows []T, ceiling int, requested int32, pageToken string) ([]T, string, error) { +type fakeCursor struct { + Last string `json:"last"` + Offset int `json:"offset"` +} + +// fakePage slices rows the way ate-api pages them: never more rows than its own +// ceiling however many were asked for, and a token that is empty on the last page and +// otherwise base64 JSON naming the page's last row, as ate-api's does. Carrying the +// row key lets a test see a token that would tell one tenant about another's rows. +func fakePage[T any](rows []T, key func(T) string, ceiling int, requested int32, pageToken string) ([]T, string, error) { pageSize := ceiling if requested > 0 && (ceiling <= 0 || int(requested) < ceiling) { pageSize = int(requested) } start := 0 if pageToken != "" { - parsed, err := strconv.Atoi(strings.TrimPrefix(pageToken, "upstream:")) - if err != nil { - return nil, "", fmt.Errorf("invalid page token %q", pageToken) + var cursor fakeCursor + raw, err := base64.StdEncoding.DecodeString(pageToken) + if err != nil || json.Unmarshal(raw, &cursor) != nil { + return nil, "", errors.New("invalid page token") } - start = parsed - } - if start > len(rows) { - start = len(rows) + start = max(cursor.Offset, 0) } + start = min(start, len(rows)) if pageSize <= 0 { return rows[start:], "", nil } end := min(start+pageSize, len(rows)) next := "" if end < len(rows) { - next = "upstream:" + strconv.Itoa(end) + next = fakePageToken(key(rows[end-1]), end) } return rows[start:end], next, nil } +func fakePageToken(last string, offset int) string { + raw, _ := json.Marshal(fakeCursor{Last: last, Offset: offset}) + return base64.StdEncoding.EncodeToString(raw) +} + +func actorKey(actor *ateapipb.Actor) string { + return actor.GetMetadata().GetAtespace() + "/" + actor.GetMetadata().GetName() +} + +func workerKey(worker *ateapipb.Worker) string { + return worker.GetMetadata().GetName() +} + func TestCurrentUser(t *testing.T) { service := system.NewService(nil, nil, nil, nil) claims := map[string]any{"sub": "user-1", "groups": []any{"admins"}} @@ -285,7 +303,7 @@ func TestListSubstrateWorkers(t *testing.T) { require.NoError(t, err) require.Len(t, page.Workers, 1) assert.Equal(t, "worker-0", page.Workers[0].WorkerPod) - assert.Equal(t, "upstream:2", page.NextPageToken) + assert.Equal(t, fakePageToken("", 2), page.NextPageToken) assert.Equal(t, 1, ateClient.workerReads) next, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "team", Page: &apiv1.PageRequest{Limit: 2, PageToken: page.NextPageToken}}) require.NoError(t, err) @@ -462,9 +480,13 @@ func TestGetSubstrateSummaryReadsAreIndependent(t *testing.T) { // counted while the pods they sit on cannot. type failingWorkersATEClient struct { fakeATEClient + workerErr error } func (client *failingWorkersATEClient) ListWorkersPage(context.Context, int32, string) ([]*ateapipb.Worker, string, error) { + if client.workerErr != nil { + return nil, "", client.workerErr + } return nil, "", errors.New("workers unavailable") } diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go index fbde15e456..6aa9f54ba3 100644 --- a/go/core/internal/service/system/substrate_paging_test.go +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -4,14 +4,18 @@ import ( "context" "encoding/base64" "fmt" + "math/rand/v2" "strings" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + atev1alpha1 "go.ate.dev/substrate/pkg/api/v1alpha1" "go.ate.dev/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" @@ -41,10 +45,24 @@ func pagingService(ateClient system.ATEClient) *system.Service { return system.NewService(nil, []string{"tenant-c", "tenant-a", "tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) } +// tokenText is a page token with every base64 layer under it decoded, so a row key +// nested in an ate-api cursor inside a kagent cursor is still visible. +func tokenText(token string) string { + text := token + for _, encoding := range []*base64.Encoding{base64.RawURLEncoding, base64.StdEncoding} { + if raw, err := encoding.DecodeString(token); err == nil && len(raw) > 0 { + text += "\n" + tokenText(string(raw)) + } + } + if _, inner, ok := strings.Cut(token, "/"); ok { + text += "\n" + tokenText(inner) + } + return text +} + func assertOwnToken(t *testing.T, rows int, pageToken string) { t.Helper() - raw, _ := base64.RawURLEncoding.DecodeString(pageToken) - assert.NotContains(t, pageToken+string(raw), "tenant-b") + assert.NotContains(t, tokenText(pageToken), "tenant-b") if rows == 0 { assert.Empty(t, pageToken, "an empty page must not lead on") } @@ -91,7 +109,7 @@ func actorNames(actors []*ateapipb.Actor) []string { return names } -func actorPageToken(raw string) string { +func scopedPageToken(raw string) string { return base64.RawURLEncoding.EncodeToString([]byte(raw)) } @@ -101,12 +119,12 @@ func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { pageToken string }{ "not base64": {pageToken: "upstream:2"}, - "no separator": {pageToken: actorPageToken("tenant-a")}, - "empty atespace": {pageToken: actorPageToken("/upstream:2")}, - "unwatched atespace": {pageToken: actorPageToken("tenant-b/upstream:2")}, - "outside the request": {atespace: "tenant-c", pageToken: actorPageToken("tenant-a/upstream:2")}, - "ate-api actor token": {pageToken: actorPageToken(`{"v":1,"kind":"actor","last":["tenant-b","b-1"]}`)}, - "over 1 KiB, else good": {pageToken: actorPageToken("tenant-a/" + strings.Repeat("x", 800))}, + "no separator": {pageToken: scopedPageToken("tenant-a")}, + "empty atespace": {pageToken: scopedPageToken("/upstream:2")}, + "unwatched atespace": {pageToken: scopedPageToken("tenant-b/upstream:2")}, + "outside the request": {atespace: "tenant-c", pageToken: scopedPageToken("tenant-a/upstream:2")}, + "ate-api actor token": {pageToken: scopedPageToken(`{"v":1,"kind":"actor","last":["tenant-b","b-1"]}`)}, + "over 1 KiB, else good": {pageToken: scopedPageToken("tenant-a/" + strings.Repeat("x", 800))}, } { t.Run(name, func(t *testing.T) { ateClient := pagingATEClient() @@ -120,6 +138,31 @@ func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { } } +func TestListSubstrateWorkersRefusesForeignTokens(t *testing.T) { + for name, tc := range map[string]struct { + namespace string + pageToken string + }{ + "not base64": {pageToken: "upstream:2"}, + "ate-api worker token": {pageToken: fakePageToken("u2-tenant-b", 2)}, + "bare worker name": {pageToken: scopedPageToken("u1-tenant-a")}, + "no worker name": {pageToken: scopedPageToken("tenant-a/")}, + "unwatched namespace": {pageToken: scopedPageToken("tenant-b/u2-tenant-b")}, + "outside the request": {namespace: "tenant-c", pageToken: scopedPageToken("tenant-a/u1-tenant-a")}, + "over 1 KiB, else good": {pageToken: scopedPageToken("tenant-a/" + strings.Repeat("x", 800))}, + } { + t.Run(name, func(t *testing.T) { + ateClient := pagingATEClient() + _, err := pagingService(ateClient).ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: tc.namespace, Page: &apiv1.PageRequest{Limit: 2, PageToken: tc.pageToken}}) + require.Error(t, err) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + assert.NotContains(t, err.Error(), "tenant-b") + assert.NotContains(t, err.Error(), tc.pageToken) + assert.Zero(t, ateClient.workerReads) + }) + } +} + func walkWorkerNames(t *testing.T, service *system.Service, namespace string, limit int32) []string { t.Helper() names := []string{} @@ -163,3 +206,75 @@ func TestListSubstrateWorkersHidesATEAPIErrorText(t *testing.T) { assert.Empty(t, page.Workers) assert.Empty(t, page.NextPageToken) } + +func TestGetSubstrateSummaryHidesATEAPIWorkerErrorText(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, atev1alpha1.AddToScheme(scheme)) + ateClient := &failingWorkersATEClient{fakeATEClient: *pagingATEClient()} + ateClient.workerErr = status.Error(codes.Internal, "worker u2-tenant-b is down") + service := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + summary, err := service.GetSubstrateSummary(pagingContext(t), "", "") + require.NoError(t, err) + assert.Equal(t, "ate-api could not list workers: Internal", summary.ATEAPIError) + assert.Equal(t, int64(3), summary.ActorCount, "the actor walk is independent of the worker walk") +} + +func TestSubstrateForgedTokensStayInWatchedNamespaces(t *testing.T) { + forged := []string{ + scopedPageToken("tenant-b/"), + scopedPageToken("tenant-a/" + fakePageToken("tenant-b/b-1", 1)), + scopedPageToken("tenant-a/" + fakePageToken("", -5)), + scopedPageToken("tenant-a/" + fakePageToken("", 1<<40)), + scopedPageToken("/" + fakePageToken("", 0)), + fakePageToken("tenant-a/a-1", 1), + "!!!", + } + random := rand.New(rand.NewPCG(1, 1)) + for range 300 { + raw := make([]byte, random.IntN(80)) + for i := range raw { + raw[i] = "tenant-abc/xyz=+"[random.IntN(16)] + } + forged = append(forged, scopedPageToken(string(raw)), string(raw)) + } + + ateClient := pagingATEClient() + service := pagingService(ateClient) + watched := []string{"tenant-a", "tenant-c"} + for _, scope := range []string{"", "tenant-a", "tenant-c"} { + for _, pageToken := range forged { + actors, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: scope, Page: &apiv1.PageRequest{Limit: 3, PageToken: pageToken}}) + if err != nil { + require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + } + for _, actor := range actors.Actors { + require.Contains(t, watched, actor.GetMetadata().GetAtespace()) + } + require.NotContains(t, tokenText(actors.NextPageToken), "tenant-b") + + workers, err := service.ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: scope, Page: &apiv1.PageRequest{Limit: 3, PageToken: pageToken}}) + if err != nil { + require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + } + for _, worker := range workers.Workers { + require.Contains(t, watched, worker.GetWorkerNamespace()) + } + require.NotContains(t, tokenText(workers.NextPageToken), "tenant-b") + } + } + assert.Subset(t, watched, ateClient.actorAtespaces, "every ate-api actor read is scoped to a watched atespace") +} + +func TestSubstrateLongestNamesFitThePageToken(t *testing.T) { + namespace := strings.Repeat("n", 63) + ateClient := &fakeATEClient{pageSize: 1} + for _, suffix := range []string{"0", "1"} { + ateClient.actors = append(ateClient.actors, substrateActor(strings.Repeat("a", 252)+suffix, namespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) + ateClient.workers = append(ateClient.workers, &ateapipb.Worker{Metadata: &ateapipb.ResourceMetadata{Name: strings.Repeat("w", 252) + suffix}, WorkerNamespace: namespace}) + } + service := system.NewService(nil, []string{namespace}, &pkgAuth.NoopAuthorizer{}, ateClient) + + assert.Len(t, walkActorNames(t, service, "", 1), 2) + assert.Len(t, walkWorkerNames(t, service, "", 1), 2) +} From a5d74efd7d59450df59f2221098e25ad310b4249 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Mon, 19 Oct 2026 00:43:31 -0400 Subject: [PATCH 16/16] fix(substrate): validate worker page tokens A controller that watches some namespaces used any worker page_token raw as the name to resume after, so an ate-api token or junk silently returned a wrong or short page. Worker tokens now share the actor tokens' shape (base64url of a watched namespace, a slash, then the cursor) and the same 1 KiB cap, and anything else is InvalidArgument. The length check now runs before decoding. Cluster-wide installs still pass ate-api's tokens through. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/substrate.go | 45 ++++++++++++++------ 1 file changed, 31 insertions(+), 14 deletions(-) diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 62af51fd13..16c7d56982 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -184,7 +184,7 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs if len(s.observedNamespaces) == 0 { actors, next, err = s.ateClient.ListActorsPage(ctx, input.GetAtespace(), limit, token) } else { - start, inner, tokenErr := parseActorPageToken(scope.namespaces, token) + start, inner, tokenErr := parseScopedPageToken(scope.namespaces, token) if tokenErr != nil { return SubstrateActorPage{}, tokenErr } @@ -205,23 +205,32 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs return result, nil } -const maxActorPageTokenBytes = 1024 +const maxPageTokenBytes = 1024 -func actorPageToken(atespace, inner string) string { - return base64.RawURLEncoding.EncodeToString([]byte(atespace + "/" + inner)) +// scopedPageToken is the cursor a controller that watches some namespaces hands out: +// the watched namespace a page ended in, then where to resume. +func scopedPageToken(namespace, cursor string) string { + return base64.RawURLEncoding.EncodeToString([]byte(namespace + "/" + cursor)) } -func parseActorPageToken(atespaces []string, token string) (int, string, error) { +func parseScopedPageToken(namespaces []string, token string) (int, string, error) { if token == "" { return 0, "", nil } + if len(token) > maxPageTokenBytes { + return 0, "", invalidPageToken() + } decoded, err := base64.RawURLEncoding.DecodeString(token) - atespace, inner, ok := strings.Cut(string(decoded), "/") - start := slices.Index(atespaces, atespace) - if len(token) > maxActorPageTokenBytes || err != nil || !ok || start < 0 { - return 0, "", serviceerrors.NewInvalidArgument("Invalid page token", nil) + namespace, cursor, ok := strings.Cut(string(decoded), "/") + start := slices.Index(namespaces, namespace) + if err != nil || !ok || start < 0 { + return 0, "", invalidPageToken() } - return start, inner, nil + return start, cursor, nil +} + +func invalidPageToken() error { + return serviceerrors.NewInvalidArgument("Invalid page token", nil) } func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inner string, limit int32) ([]*ateapipb.Actor, string, error) { @@ -234,9 +243,9 @@ func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inne actors = append(actors, rows...) switch { case next != "": - return actors, actorPageToken(atespace, next), nil + return actors, scopedPageToken(atespace, next), nil case int32(len(actors)) >= limit && i+1 < len(atespaces): - return actors, actorPageToken(atespaces[i+1], ""), nil + return actors, scopedPageToken(atespaces[i+1], ""), nil } inner = "" } @@ -251,7 +260,14 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub } limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() if len(s.observedNamespaces) > 0 { - return s.listWatchedWorkers(ctx, scope, limit, token), nil + _, after, err := parseScopedPageToken(scope.namespaces, token) + if err == nil && token != "" && after == "" { + err = invalidPageToken() + } + if err != nil { + return SubstrateWorkerPage{}, err + } + return s.listWatchedWorkers(ctx, scope, limit, after), nil } result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} workers, next, err := s.ateClient.ListWorkersPage(ctx, limit, token) @@ -285,7 +301,8 @@ func (s *Service) listWatchedWorkers(ctx context.Context, scope substrateScope, }) if int32(len(workers)) > limit { workers = workers[:limit] - result.NextPageToken = workers[limit-1].GetMetadata().GetName() + last := workers[limit-1] + result.NextPageToken = scopedPageToken(last.GetWorkerNamespace(), last.GetMetadata().GetName()) } result.Workers = workers return result