feat (cli): add a wait command and watch flag

cmd/kubectl-ate/README.md

@@ -83,9 +83,15 @@ kubectl ate get actors -a <atespace>
 # List actors across all atespaces
 kubectl ate get actors -A
 
+# List actors, then print actors again when their state changes
+kubectl ate get actors -A --watch
+
 # Get a specific actor by name and output as raw YAML
 kubectl ate get actor <actor-name> --atespace <atespace> -o yaml
 
+# Get an actor, then watch it for lifecycle changes
+kubectl ate get actor <actor-name> --atespace <atespace> -w
+
 # List all physical workers and see which actors are assigned to them
 kubectl ate get workers
 ```
@@ -162,6 +168,9 @@ kubectl ate suspend actor my-actor -a <atespace>
 
 # Delete an actor.
 kubectl ate delete actor my-actor -a <atespace>
+
+# Wait up to 60 seconds for an actor to finish resuming.
+kubectl ate wait actor my-actor -a <atespace> --for=status=running --timeout=60s
 ```
 
 ### Logs

cmd/kubectl-ate/internal/cmd/get_actors.go

@@ -15,6 +15,7 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/agent-substrate/substrate/cmd/kubectl-ate/internal/printer"
@@ -26,6 +27,7 @@ import (
 var (
 	getActorsAtespaceFlag string
 	getActorsAllAtespaces bool
+	getActorsWatch        bool
 )
 
 var getActorsCmd = &cobra.Command{
@@ -53,15 +55,18 @@ var getActorsCmd = &cobra.Command{
 				return fmt.Errorf("--atespace is required when getting actors")
 			}
 
-			actors := make([]*ateapipb.Actor, 0, len(args))
-			for _, actorName := range args {
-				resp, err := apiClient.GetActor(ctx, &ateapipb.GetActorRequest{Actor: &ateapipb.ObjectRef{Atespace: getActorsAtespaceFlag, Name: actorName}})
-				if err != nil {
-					return fmt.Errorf("failed to get actor %q: %w", actorName, err)
+			fetch := func(ctx context.Context) ([]*ateapipb.Actor, error) {
+				actors := make([]*ateapipb.Actor, 0, len(args))
+				for _, actorName := range args {
+					resp, err := apiClient.GetActor(ctx, &ateapipb.GetActorRequest{Actor: &ateapipb.ObjectRef{Atespace: getActorsAtespaceFlag, Name: actorName}})
+					if err != nil {
+						return nil, fmt.Errorf("failed to get actor %q: %w", actorName, err)
+					}
+					actors = append(actors, resp)
 				}
-				actors = append(actors, resp)
+				return actors, nil
 			}
-			return printer.PrintActors(actors, outputFmt)
+			return printOrWatch(ctx, cmd.OutOrStdout(), getActorsWatch, outputFmt, fetch, printer.PrintActorsTo, actorWatchKey)
 		}
 
 		// Listing requires exactly one of --atespace (one atespace) or -A (all
@@ -74,32 +79,30 @@ var getActorsCmd = &cobra.Command{
 		}
 
 		// 3. Handle List All Actors
-		var allActors []*ateapipb.Actor
-		pageToken := ""
-
-		for {
-			resp, err := apiClient.ListActors(ctx, &ateapipb.ListActorsRequest{
-				PageSize:  1000,
-				PageToken: pageToken,
-				Atespace:  getActorsAtespaceFlag,
+		fetch := func(ctx context.Context) ([]*ateapipb.Actor, error) {
+			return fetchAllPages(ctx, func(ctx context.Context, pageToken string) ([]*ateapipb.Actor, string, error) {
+				resp, err := apiClient.ListActors(ctx, &ateapipb.ListActorsRequest{
+					PageSize:  1000,
+					PageToken: pageToken,
+					Atespace:  getActorsAtespaceFlag,
+				})
+				if err != nil {
+					return nil, "", fmt.Errorf("failed to list actors: %w", err)
+				}
+				return resp.GetActors(), resp.GetNextPageToken(), nil
 			})
-			if err != nil {
-				return fmt.Errorf("failed to list actors: %w", err)
-			}
-			allActors = append(allActors, resp.GetActors()...)
-
-			pageToken = resp.GetNextPageToken()
-			if pageToken == "" {
-				break
-			}
 		}
-
-		return printer.PrintActors(allActors, outputFmt)
+		return printOrWatch(ctx, cmd.OutOrStdout(), getActorsWatch, outputFmt, fetch, printer.PrintActorsTo, actorWatchKey)
 	},
 }
 
+func actorWatchKey(actor *ateapipb.Actor) string {
+	return actor.GetMetadata().GetAtespace() + "/" + actor.GetMetadata().GetName()
+}
+
 func init() {
 	getActorsCmd.Flags().StringVarP(&getActorsAtespaceFlag, "atespace", "a", "", "Atespace to list/get actors in. Required when getting actors; for listing, use this or -A.")
 	getActorsCmd.Flags().BoolVarP(&getActorsAllAtespaces, "all-atespaces", "A", false, "List actors across all atespaces (listing only; mutually exclusive with --atespace)")
+	getActorsCmd.Flags().BoolVarP(&getActorsWatch, "watch", "w", false, "After listing/getting the requested actor(s), watch for changes")
 	getCmd.AddCommand(getActorsCmd)
 }

cmd/kubectl-ate/internal/cmd/get_atespaces.go

@@ -15,6 +15,7 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/agent-substrate/substrate/cmd/kubectl-ate/internal/printer"
@@ -23,6 +24,8 @@ import (
 	"github.com/spf13/cobra"
 )
 
+var getAtespacesWatch bool
+
 var getAtespacesCmd = &cobra.Command{
 	Use:     "atespaces [name ...]",
 	Aliases: []string{"atespace"},
@@ -36,38 +39,41 @@ var getAtespacesCmd = &cobra.Command{
 		defer apiClient.Close()
 
 		if len(args) > 0 {
-			atespaces := make([]*ateapipb.Atespace, 0, len(args))
-			for _, atespaceName := range args {
-				resp, err := apiClient.GetAtespace(ctx, &ateapipb.GetAtespaceRequest{Atespace: &ateapipb.ObjectRef{Name: atespaceName}})
-				if err != nil {
-					return fmt.Errorf("failed to get atespace %q: %w", atespaceName, err)
+			fetch := func(ctx context.Context) ([]*ateapipb.Atespace, error) {
+				atespaces := make([]*ateapipb.Atespace, 0, len(args))
+				for _, atespaceName := range args {
+					resp, err := apiClient.GetAtespace(ctx, &ateapipb.GetAtespaceRequest{Atespace: &ateapipb.ObjectRef{Name: atespaceName}})
+					if err != nil {
+						return nil, fmt.Errorf("failed to get atespace %q: %w", atespaceName, err)
+					}
+					atespaces = append(atespaces, resp)
 				}
-				atespaces = append(atespaces, resp)
+				return atespaces, nil
 			}
-			return printer.PrintAtespaces(atespaces, outputFmt)
+			return printOrWatch(ctx, cmd.OutOrStdout(), getAtespacesWatch, outputFmt, fetch, printer.PrintAtespacesTo, atespaceWatchKey)
 		}
 
-		var allAtespaces []*ateapipb.Atespace
-		pageToken := ""
-		for {
-			resp, err := apiClient.ListAtespaces(ctx, &ateapipb.ListAtespacesRequest{
-				PageSize:  1000,
-				PageToken: pageToken,
+		fetch := func(ctx context.Context) ([]*ateapipb.Atespace, error) {
+			return fetchAllPages(ctx, func(ctx context.Context, pageToken string) ([]*ateapipb.Atespace, string, error) {
+				resp, err := apiClient.ListAtespaces(ctx, &ateapipb.ListAtespacesRequest{
+					PageSize:  1000,
+					PageToken: pageToken,
+				})
+				if err != nil {
+					return nil, "", fmt.Errorf("failed to list atespaces: %w", err)
+				}
+				return resp.GetAtespaces(), resp.GetNextPageToken(), nil
 			})
-			if err != nil {
-				return fmt.Errorf("failed to list atespaces: %w", err)
-			}
-			allAtespaces = append(allAtespaces, resp.GetAtespaces()...)
-
-			pageToken = resp.GetNextPageToken()
-			if pageToken == "" {
-				break
-			}
 		}
-		return printer.PrintAtespaces(allAtespaces, outputFmt)
+		return printOrWatch(ctx, cmd.OutOrStdout(), getAtespacesWatch, outputFmt, fetch, printer.PrintAtespacesTo, atespaceWatchKey)
 	},
 }
 
+func atespaceWatchKey(atespace *ateapipb.Atespace) string {
+	return atespace.GetMetadata().GetName()
+}
+
 func init() {
+	getAtespacesCmd.Flags().BoolVarP(&getAtespacesWatch, "watch", "w", false, "After listing/getting the requested atespace(s), watch for changes")
 	getCmd.AddCommand(getAtespacesCmd)
 }

cmd/kubectl-ate/internal/cmd/get_test.go

@@ -49,3 +49,18 @@ func TestGetCommandArgs(t *testing.T) {
 		})
 	}
 }
+
+func TestGetCommandsHaveWatchFlag(t *testing.T) {
+	commands := []*cobra.Command{getActorsCmd, getAtespacesCmd, getWorkersCmd}
+	for _, command := range commands {
+		t.Run(command.Name(), func(t *testing.T) {
+			flag := command.Flags().Lookup("watch")
+			if flag == nil {
+				t.Fatal("--watch flag is not registered")
+			}
+			if flag.Shorthand != "w" {
+				t.Errorf("--watch shorthand = %q, want w", flag.Shorthand)
+			}
+		})
+	}
+}

cmd/kubectl-ate/internal/cmd/get_workers.go

@@ -15,6 +15,7 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/agent-substrate/substrate/cmd/kubectl-ate/internal/printer"
@@ -23,6 +24,8 @@ import (
 	"github.com/spf13/cobra"
 )
 
+var getWorkersWatch bool
+
 var getWorkersCmd = &cobra.Command{
 	Use:     "workers",
 	Aliases: []string{"worker"},
@@ -36,27 +39,27 @@ var getWorkersCmd = &cobra.Command{
 		}
 		defer apiClient.Close()
 
-		var allWorkers []*ateapipb.Worker
-		pageToken := ""
-		for {
-			resp, err := apiClient.ListWorkers(ctx, &ateapipb.ListWorkersRequest{
-				PageSize:  1000,
-				PageToken: pageToken,
+		fetch := func(ctx context.Context) ([]*ateapipb.Worker, error) {
+			return fetchAllPages(ctx, func(ctx context.Context, pageToken string) ([]*ateapipb.Worker, string, error) {
+				resp, err := apiClient.ListWorkers(ctx, &ateapipb.ListWorkersRequest{
+					PageSize:  1000,
+					PageToken: pageToken,
+				})
+				if err != nil {
+					return nil, "", fmt.Errorf("failed to list workers: %w", err)
+				}
+				return resp.GetWorkers(), resp.GetNextPageToken(), nil
 			})
-			if err != nil {
-				return fmt.Errorf("failed to list workers: %w", err)
-			}
-			allWorkers = append(allWorkers, resp.GetWorkers()...)
-
-			pageToken = resp.GetNextPageToken()
-			if pageToken == "" {
-				break
-			}
 		}
-		return printer.PrintWorkers(allWorkers, outputFmt)
+		return printOrWatch(ctx, cmd.OutOrStdout(), getWorkersWatch, outputFmt, fetch, printer.PrintWorkersTo, workerWatchKey)
 	},
 }
 
+func workerWatchKey(worker *ateapipb.Worker) string {
+	return worker.GetWorkerNamespace() + "/" + worker.GetWorkerPool() + "/" + worker.GetWorkerPod()
+}
+
 func init() {
+	getWorkersCmd.Flags().BoolVarP(&getWorkersWatch, "watch", "w", false, "After listing the requested workers, watch for changes")
 	getCmd.AddCommand(getWorkersCmd)
 }

cmd/kubectl-ate/internal/cmd/pagination.go

@@ -0,0 +1,33 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import "context"
+
+func fetchAllPages[M any](ctx context.Context, fetchPage func(context.Context, string) ([]M, string, error)) ([]M, error) {
+	var all []M
+	pageToken := ""
+	for {
+		resources, nextPageToken, err := fetchPage(ctx, pageToken)
+		if err != nil {
+			return nil, err
+		}
+		all = append(all, resources...)
+		if nextPageToken == "" {
+			return all, nil
+		}
+		pageToken = nextPageToken
+	}
+}

cmd/kubectl-ate/internal/cmd/pagination_test.go

@@ -0,0 +1,55 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"slices"
+	"testing"
+)
+
+func TestFetchAllPages(t *testing.T) {
+	var tokens []string
+	got, err := fetchAllPages(context.Background(), func(_ context.Context, token string) ([]string, string, error) {
+		tokens = append(tokens, token)
+		if token == "" {
+			return []string{"one", "two"}, "next", nil
+		}
+		return []string{"three"}, "", nil
+	})
+	if err != nil {
+		t.Fatalf("fetchAllPages() error = %v", err)
+	}
+	if want := []string{"one", "two", "three"}; !slices.Equal(got, want) {
+		t.Errorf("fetchAllPages() = %v, want %v", got, want)
+	}
+	if want := []string{"", "next"}; !slices.Equal(tokens, want) {
+		t.Errorf("page tokens = %v, want %v", tokens, want)
+	}
+}
+
+func TestFetchAllPagesReturnsError(t *testing.T) {
+	wantErr := errors.New("list failed")
+	got, err := fetchAllPages(context.Background(), func(context.Context, string) ([]string, string, error) {
+		return nil, "", wantErr
+	})
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("fetchAllPages() error = %v, want %v", err, wantErr)
+	}
+	if got != nil {
+		t.Errorf("fetchAllPages() = %v, want nil", got)
+	}
+}

cmd/kubectl-ate/internal/cmd/wait.go

@@ -0,0 +1,26 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import "github.com/spf13/cobra"
+
+var waitCmd = &cobra.Command{
+	Use:   "wait",
+	Short: "Wait for a specific condition on one or more resources",
+}
+
+func init() {
+	rootCmd.AddCommand(waitCmd)
+}