Add batch cancel, terminate and delete operators for standalone activities

go.mod

@@ -17,14 +17,14 @@ require (
 	github.com/stretchr/testify v1.11.1
 	github.com/temporalio/cli/cliext v0.0.0
 	github.com/temporalio/ui-server/v2 v2.52.0
-	go.temporal.io/api v1.63.0
-	go.temporal.io/sdk v1.45.0
+	go.temporal.io/api v1.63.4-0.20260720155646-7a3ca3626ec2
+	go.temporal.io/sdk v1.46.1-0.20260720184640-f34dc3da35ab
 	go.temporal.io/sdk/contrib/envconfig v1.0.2
-	go.temporal.io/server v1.32.0-158.0
-	golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f
-	golang.org/x/mod v0.35.0
-	golang.org/x/term v0.43.0
-	golang.org/x/tools v0.44.0
+	go.temporal.io/server v1.29.0-135.0.0.20260720172935-768cd1f81135
+	golang.org/x/exp v0.0.0-20260611194520-c48552f49976
+	golang.org/x/mod v0.37.0
+	golang.org/x/term v0.45.0
+	golang.org/x/tools v0.47.0
 	google.golang.org/grpc v1.81.1
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/yaml.v3 v3.0.1
@@ -78,6 +78,7 @@ require (
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cactus/go-statsd-client/statsd v0.0.0-20200423205355-cb0885a1018c // indirect
 	github.com/cactus/go-statsd-client/v5 v5.1.0 // indirect
+	github.com/caio/go-tdigest/v5 v5.0.0 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/clipperhouse/uax29/v2 v2.7.0 // indirect
@@ -129,7 +130,7 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
-	github.com/jackc/pgx/v5 v5.9.2 // indirect
+	github.com/jackc/pgx/v5 v5.10.0 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jmoiron/sqlx v1.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
@@ -182,17 +183,18 @@ require (
 	go.opentelemetry.io/contrib/detectors/gcp v1.43.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect
-	go.opentelemetry.io/otel v1.43.0 // indirect
+	go.opentelemetry.io/otel v1.44.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.43.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.57.0 // indirect
-	go.opentelemetry.io/otel/metric v1.43.0 // indirect
+	go.opentelemetry.io/otel/log v0.20.0 // indirect
+	go.opentelemetry.io/otel/metric v1.44.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.43.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.43.0 // indirect
-	go.opentelemetry.io/otel/trace v1.43.0 // indirect
+	go.opentelemetry.io/otel/trace v1.44.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
-	go.temporal.io/auto-scaled-workers v0.0.0-20260622220320-9b1e3849116d // indirect
+	go.temporal.io/auto-scaled-workers v0.0.0-20260706201056-4320b34799ee // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/dig v1.19.0 // indirect
 	go.uber.org/fx v1.24.0 // indirect
@@ -201,12 +203,12 @@ require (
 	go.uber.org/zap v1.27.1 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/crypto v0.52.0 // indirect
-	golang.org/x/net v0.55.0 // indirect
+	golang.org/x/crypto v0.54.0 // indirect
+	golang.org/x/net v0.57.0 // indirect
 	golang.org/x/oauth2 v0.36.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
-	golang.org/x/sys v0.45.0 // indirect
-	golang.org/x/text v0.37.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	golang.org/x/text v0.40.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
 	google.golang.org/api v0.276.0 // indirect
 	google.golang.org/genproto v0.0.0-20260420184626-e10c466a9529 // indirect

internal/temporalcli/commands.activity.go

@@ -1,13 +1,15 @@
 package temporalcli
 
 import (
+	"cmp"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
 
 	"github.com/fatih/color"
+	"github.com/google/uuid"
 	"github.com/temporalio/cli/internal/printer"
 	activitypb "go.temporal.io/api/activity/v1"
 	"go.temporal.io/api/batch/v1"
@@ -40,6 +42,8 @@ type (
 	}
 )
 
+const activityDeleteWarning = "WARNING: Deleting Standalone Activity Executions in a global Namespace removes them from all replicas. Requests sent to a passive cluster are forwarded to the active cluster by default; to target the passive cluster directly, specify `--grpc-meta xdc-redirection=false`."
+
 func (c *TemporalActivityStartCommand) run(cctx *CommandContext, args []string) error {
 	cl, err := dialClient(cctx, &c.Parent.ClientOptions)
 	if err != nil {
@@ -562,21 +566,107 @@ func (c *TemporalActivityCountCommand) run(cctx *CommandContext, args []string)
 	return nil
 }
 
+func (s *ActivityReferenceOrBatchOptions) activityExecOrBatch(
+	cctx *CommandContext,
+	namespace string,
+	cl client.Client,
+	yesFlag bool,
+	overrides singleOrBatchOverrides,
+) (*client.GetActivityHandleOptions, *workflowservice.StartBatchOperationRequest, error) {
+	// If activity is set, we return activity handle options with activity ID and run ID
+	if s.ActivityId != "" {
+		if s.Query != "" {
+			return nil, nil, fmt.Errorf("cannot set query when activity ID is set")
+		} else if yesFlag && !overrides.AllowYesWithActivityID {
+			return nil, nil, fmt.Errorf("cannot set 'yes' when activity ID is set")
+		} else if s.Rps != 0 {
+			return nil, nil, fmt.Errorf("cannot set rps when activity ID is set")
+		}
+		return &client.GetActivityHandleOptions{
+			ActivityID: s.ActivityId,
+			RunID:      s.RunId,
+		}, nil, nil
+	}
+
+	// Check query is set properly
+	if s.Query == "" {
+		return nil, nil, fmt.Errorf("must set either activity ID or query")
+	} else if s.ActivityId != "" { // This is redundant, but kept for completeness
+		return nil, nil, fmt.Errorf("cannot set activity ID when query is set")
+	} else if s.RunId != "" {
+		return nil, nil, fmt.Errorf("cannot set run ID when query is set")
+	}
+
+	// The count is only used in the confirmation prompt; skip the request when --yes
+	// bypasses it, so batch jobs can still proceed if the visibility API is timing out.
+	var promptMessage string
+	if yesFlag {
+		promptMessage = fmt.Sprintf("Start batch against standalone activities matching query %q? y/N", s.Query)
+	} else {
+		count, err := cl.CountActivities(cctx, client.CountActivitiesOptions{Query: s.Query})
+		if err != nil {
+			return nil, nil, fmt.Errorf("failed counting standalone activities from query: %w", err)
+		}
+		promptMessage = fmt.Sprintf("Start batch against approximately %v standalone activities(s)? y/N", count.Count)
+	}
+	isYes, err := cctx.promptYes(promptMessage, yesFlag)
+	if err != nil {
+		return nil, nil, err
+	} else if !isYes {
+		// We consider this a command failure
+		return nil, nil, fmt.Errorf("user denied confirmation")
+	}
+
+	return nil, &workflowservice.StartBatchOperationRequest{
+		MaxOperationsPerSecond: s.Rps,
+		Namespace:              namespace,
+		JobId:                  uuid.NewString(),
+		VisibilityQuery:        s.Query,
+	}, nil
+}
+
 func (c *TemporalActivityCancelCommand) run(cctx *CommandContext, args []string) error {
 	cl, err := dialClient(cctx, &c.Parent.ClientOptions)
 	if err != nil {
 		return err
 	}
 	defer cl.Close()
 
-	handle := cl.GetActivityHandle(client.GetActivityHandleOptions{
-		ActivityID: c.ActivityId,
-		RunID:      c.RunId,
-	})
-	if err := handle.Cancel(cctx, client.CancelActivityOptions{Reason: c.Reason}); err != nil {
-		return fmt.Errorf("failed to request activity cancellation: %w", err)
+	opts := ActivityReferenceOrBatchOptions{
+		ActivityId: c.ActivityId,
+		RunId:      c.RunId,
+		Query:      c.Query,
+		Rps:        c.Rps,
+	}
+
+	activityOptions, batchReq, err := opts.activityExecOrBatch(cctx, c.Parent.Namespace, cl, c.Yes, singleOrBatchOverrides{})
+	if err != nil {
+		return err
+	}
+
+	if activityOptions != nil {
+		handle := cl.GetActivityHandle(*activityOptions)
+		if err := handle.Cancel(cctx, client.CancelActivityOptions{Reason: c.Reason}); err != nil {
+			return fmt.Errorf("failed to request activity cancellation: %w", err)
+		}
+		cctx.Printer.Println("Cancellation requested")
+	} else { // batchReq != nil
+		cancelActivitiesOperation := &batch.BatchOperationCancelActivities{
+			Identity: c.Parent.Identity,
+			// do not fallback to defaultReason, to be consistent with single activity cancel
+			Reason: c.Reason,
+		}
+
+		// Reason in batch request falls back to defaultReason
+		batchReq.Reason = cmp.Or(c.Reason, defaultReason())
+		batchReq.Operation = &workflowservice.StartBatchOperationRequest_CancelActivitiesOperation{
+			CancelActivitiesOperation: cancelActivitiesOperation,
+		}
+
+		if err := startBatchJob(cctx, cl, batchReq); err != nil {
+			return err
+		}
 	}
-	cctx.Printer.Println("Cancellation requested")
 	return nil
 }
 
@@ -587,24 +677,117 @@ func (c *TemporalActivityTerminateCommand) run(cctx *CommandContext, args []stri
 	}
 	defer cl.Close()
 
-	// The CLI adds a default for terminate but not cancel.
-	// This matches the behavior for workflows.
-	reason := c.Reason
-	if reason == "" {
-		reason = defaultReason()
+	opts := ActivityReferenceOrBatchOptions{
+		ActivityId: c.ActivityId,
+		RunId:      c.RunId,
+		Query:      c.Query,
+		Rps:        c.Rps,
 	}
-	handle := cl.GetActivityHandle(client.GetActivityHandleOptions{
-		ActivityID: c.ActivityId,
-		RunID:      c.RunId,
+
+	activityOptions, batchReq, err := opts.activityExecOrBatch(cctx, c.Parent.Namespace, cl, c.Yes, singleOrBatchOverrides{})
+	if err != nil {
+		return err
+	}
+
+	// Reason for single terminate or batch request falls back to defaultReason
+	reason := cmp.Or(c.Reason, defaultReason())
+
+	if activityOptions != nil {
+		// The CLI adds a default for terminate but not cancel.
+		// This matches the behavior for workflows.
+		handle := cl.GetActivityHandle(*activityOptions)
+		// Terminate may fail if the activity doesn't exist or has already completed.
+		if err := handle.Terminate(cctx, client.TerminateActivityOptions{Reason: reason}); err != nil {
+			return fmt.Errorf("failed to terminate activity: %w", err)
+		}
+		cctx.Printer.Println("Activity terminated")
+	} else { // batchReq != nil
+		terminateActivitiesOperation := &batch.BatchOperationTerminateActivities{
+			Identity: c.Parent.Identity,
+			Reason:   reason,
+		}
+
+		batchReq.Reason = reason
+		batchReq.Operation = &workflowservice.StartBatchOperationRequest_TerminateActivitiesOperation{
+			TerminateActivitiesOperation: terminateActivitiesOperation,
+		}
+
+		if err := startBatchJob(cctx, cl, batchReq); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *TemporalActivityDeleteCommand) run(cctx *CommandContext, args []string) error {
+	cl, err := dialClient(cctx, &c.Parent.ClientOptions)
+	if err != nil {
+		return err
+	}
+	defer cl.Close()
+
+	// Only warn when the namespace is global, or can't get the namespace info
+	nsResp, nsErr := cl.WorkflowService().DescribeNamespace(cctx, &workflowservice.DescribeNamespaceRequest{
+		Namespace: c.Parent.Namespace,
 	})
-	// Terminate may fail if the activity doesn't exist or has already completed.
-	if err := handle.Terminate(cctx, client.TerminateActivityOptions{Reason: reason}); err != nil {
-		return fmt.Errorf("failed to terminate activity: %w", err)
+	if nsErr != nil || nsResp.GetIsGlobalNamespace() {
+		fmt.Fprintln(cctx.Options.Stderr, activityDeleteWarning)
 	}
-	cctx.Printer.Println("Activity terminated")
+
+	opts := ActivityReferenceOrBatchOptions{
+		ActivityId: c.ActivityId,
+		RunId:      c.RunId,
+		Query:      c.Query,
+		Rps:        c.Rps,
+	}
+
+	activityOptions, batchReq, err := opts.activityExecOrBatch(cctx, c.Parent.Namespace, cl, c.Yes, singleOrBatchOverrides{
+		AllowYesWithActivityID: true,
+	})
+	if err != nil {
+		return err
+	}
+
+	if activityOptions != nil {
+		yes, err := cctx.promptYes(activityDeleteSingleConfirmationMessage(activityOptions), c.Yes)
+		if err != nil {
+			return err
+		} else if !yes {
+			return fmt.Errorf("user denied confirmation")
+		}
+		_, err = cl.WorkflowService().DeleteActivityExecution(cctx, &workflowservice.DeleteActivityExecutionRequest{
+			Namespace:  c.Parent.Namespace,
+			ActivityId: c.ActivityId,
+			RunId:      c.RunId,
+		})
+		if err != nil {
+			return fmt.Errorf("failed to delete standalone activity: %w", err)
+		}
+		cctx.Printer.Println("Delete activity succeeded")
+	} else { // batchReq != nil
+		deleteActivitiesOperation := &batch.BatchOperationDeleteActivities{}
+		batchReq.Reason = cmp.Or(c.Reason, defaultReason())
+		batchReq.Operation = &workflowservice.StartBatchOperationRequest_DeleteActivitiesOperation{
+			DeleteActivitiesOperation: deleteActivitiesOperation,
+		}
+
+		if err := startBatchJob(cctx, cl, batchReq); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
+func activityDeleteSingleConfirmationMessage(activityOptions *client.GetActivityHandleOptions) string {
+	action := fmt.Sprintf("Delete Standalone Activity %q", activityOptions.ActivityID)
+	if activityOptions.RunID != "" {
+		action += fmt.Sprintf(" with Run ID %q", activityOptions.RunID)
+	}
+	return fmt.Sprintf("%s? y/N", action)
+}
+
 func (c *TemporalActivityCompleteCommand) run(cctx *CommandContext, args []string) error {
 	cl, err := dialClient(cctx, &c.Parent.ClientOptions)
 	if err != nil {