Skip to content

feat: add support for single SAA operator actions #1092

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
spkane31 wants to merge 8 commits into
base: main
Choose a base branch
from
Open

feat: add support for single SAA operator actions #1092

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
efcef0c
feat: add support for single SAA operator actions
spkane31 Jun 18, 2026
95b03aa
code gen and improved documentation of flags
spkane31 Jun 18, 2026
ed31767
refining
spkane31 Jun 18, 2026
f0142e3
remove the reset-heartbeats flag
spkane31 Jul 20, 2026
0d2176c
fixing tests
spkane31 Jul 20, 2026
9e99c57
wrong name
spkane31 Jul 21, 2026
57ad7a9
enable saa operator commands by DC
spkane31 Jul 29, 2026
b72bb14
add update-options, correct some documentation strings
spkane31 Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
187 changes: 124 additions & 63 deletions internal/temporalcli/commands.activity.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ type (
ScheduleToStartTimeout time.Duration
StartToCloseTimeout time.Duration
HeartbeatTimeout time.Duration
StartDelay time.Duration

InitialInterval time.Duration
BackoffCoefficient float64
Expand All @@ -44,6 +45,23 @@ 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`."

// Targeting errors for the activity operate commands. They name every flag a
// caller can use so the failure tells the user exactly how to retarget:
// - a single Activity: --activity-id, optionally with --workflow-id (workflow
// Activity) and/or --run-id (specific run; standalone when no --workflow-id)
// - a batch of workflow Activities: --query
var (
// errActivityTarget is used by the operations that support batching
// (unpause, reset, update-options).
errActivityTarget = errors.New("must specify --activity-id to target a single Activity " +
"(optionally with --workflow-id and/or --run-id), or --query to target a batch of Activities")

// errPauseActivityTarget is the pause equivalent; pause has no --query
// (batch) mode.
errPauseActivityTarget = errors.New("must specify --activity-id to pause an Activity " +
"(optionally with --workflow-id and/or --run-id)")
)

func (c *TemporalActivityStartCommand) run(cctx *CommandContext, args []string) error {
cl, err := dialClient(cctx, &c.Parent.ClientOptions)
if err != nil {
Expand Down Expand Up @@ -863,7 +881,7 @@ func (c *TemporalActivityUpdateOptionsCommand) run(cctx *CommandContext, args []

if c.Command.Flags().Changed("task-queue") {
activityOptions.TaskQueue = &taskqueuepb.TaskQueue{Name: c.TaskQueue}
updatePath = append(updatePath, "task_queue_name")
updatePath = append(updatePath, "task_queue.name")
}

if c.Command.Flags().Changed("schedule-to-close-timeout") {
Expand All @@ -886,6 +904,11 @@ func (c *TemporalActivityUpdateOptionsCommand) run(cctx *CommandContext, args []
updatePath = append(updatePath, "heartbeat_timeout")
}

if c.Command.Flags().Changed("start-delay") {
activityOptions.StartDelay = durationpb.New(c.StartDelay.Duration())
updatePath = append(updatePath, "start_delay")
}

if c.Command.Flags().Changed("retry-initial-interval") ||
c.Command.Flags().Changed("retry-maximum-interval") ||
c.Command.Flags().Changed("retry-backoff-coefficient") ||
Expand Down Expand Up @@ -924,28 +947,42 @@ func (c *TemporalActivityUpdateOptionsCommand) run(cctx *CommandContext, args []
Rps: c.Rps,
}

exec, batchReq, err := opts.workflowExecOrBatch(cctx, c.Parent.Namespace, cl, singleOrBatchOverrides{})
if err != nil {
return err
// Target a workflow Activity (--workflow-id), a standalone Activity (no
// --workflow-id; the latest run unless --run-id is set), or a batch of
// workflow Activities (--query). The server routes to standalone-activity
// handling when the workflow ID is empty, so standalone Activities take the
// single-execution path below.
var exec *common.WorkflowExecution
var batchReq *workflowservice.StartBatchOperationRequest
if c.WorkflowId == "" && c.Query == "" {
if c.ActivityId == "" {
return errActivityTarget
}
exec = &common.WorkflowExecution{RunId: c.RunId}
} else {
exec, batchReq, err = opts.workflowExecOrBatch(cctx, c.Parent.Namespace, cl, singleOrBatchOverrides{})
if err != nil {
return err
}
}

if exec != nil {
if c.ActivityId == "" {
return fmt.Errorf("either --activity-id and --workflow-id, or --query must be set")
return errActivityTarget
}
result, err := cl.WorkflowService().UpdateActivityOptions(cctx, &workflowservice.UpdateActivityOptionsRequest{
Namespace: c.Parent.Namespace,
Execution: &common.WorkflowExecution{
WorkflowId: c.WorkflowId,
RunId: c.RunId,
},
Activity: &workflowservice.UpdateActivityOptionsRequest_Id{Id: c.ActivityId},
ActivityOptions: activityOptions,
UpdateMask: &fieldmaskpb.FieldMask{
Paths: updatePath,
},
Identity: c.Parent.Identity,
})
result, err := cl.WorkflowService().UpdateActivityExecutionOptions(
cctx,
&workflowservice.UpdateActivityExecutionOptionsRequest{
Namespace: c.Parent.Namespace,
WorkflowId: c.WorkflowId,
RunId: c.RunId,
ActivityId: c.ActivityId,
ActivityOptions: activityOptions,
UpdateMask: &fieldmaskpb.FieldMask{
Paths: updatePath,
},
Identity: c.Parent.Identity,

@fretz12 fretz12 Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we need to take in --restore-original-options?

})
if err != nil {
return fmt.Errorf("unable to update Activity options: %w", err)
}
Expand All @@ -957,6 +994,7 @@ func (c *TemporalActivityUpdateOptionsCommand) run(cctx *CommandContext, args []
ScheduleToStartTimeout: result.GetActivityOptions().ScheduleToStartTimeout.AsDuration(),
StartToCloseTimeout: result.GetActivityOptions().StartToCloseTimeout.AsDuration(),
HeartbeatTimeout: result.GetActivityOptions().HeartbeatTimeout.AsDuration(),
StartDelay: result.GetActivityOptions().StartDelay.AsDuration(),

InitialInterval: result.GetActivityOptions().RetryPolicy.InitialInterval.AsDuration(),
BackoffCoefficient: result.GetActivityOptions().RetryPolicy.BackoffCoefficient,
Expand Down Expand Up @@ -988,30 +1026,32 @@ func (c *TemporalActivityUpdateOptionsCommand) run(cctx *CommandContext, args []

func (c *TemporalActivityPauseCommand) run(cctx *CommandContext, args []string) error {
if c.ActivityId == "" {
return fmt.Errorf("Activity Id must be specified")
return errPauseActivityTarget
}
// Set --workflow-id to target a workflow Activity. Omit it to target a
// standalone Activity by --activity-id (and optionally --run-id for a
// specific run); the server routes to standalone-activity handling when the
// workflow ID is empty.

cl, err := dialClient(cctx, &c.Parent.ClientOptions)
if err != nil {
return err
}
defer cl.Close()

request := &workflowservice.PauseActivityRequest{
Namespace: c.Parent.Namespace,
Execution: &common.WorkflowExecution{
WorkflowId: c.WorkflowId,
RunId: c.RunId,
},
Identity: c.Identity,
Reason: c.Reason,
Activity: &workflowservice.PauseActivityRequest_Id{Id: c.ActivityId},
request := &workflowservice.PauseActivityExecutionRequest{
Namespace: c.Parent.Namespace,
WorkflowId: c.WorkflowId,
ActivityId: c.ActivityId,
RunId: c.RunId,
Identity: c.Identity,
Reason: c.Reason,
}
if request.Identity == "" {
request.Identity = c.Parent.Identity
}

_, err = cl.WorkflowService().PauseActivity(cctx, request)
_, err = cl.WorkflowService().PauseActivityExecution(cctx, request)
if err != nil {
return fmt.Errorf("unable to pause Activity: %w", err)
}
Expand All @@ -1037,30 +1077,42 @@ func (c *TemporalActivityUnpauseCommand) run(cctx *CommandContext, args []string
Rps: c.Rps,
}

exec, batchReq, err := opts.workflowExecOrBatch(cctx, c.Parent.Namespace, cl, singleOrBatchOverrides{})
if err != nil {
return err
// Target a workflow Activity (--workflow-id), a standalone Activity (no
// --workflow-id; the latest run unless --run-id is set), or a batch of
// workflow Activities (--query). The server routes to standalone-activity
// handling when the workflow ID is empty, so standalone Activities take the
// single-execution path below.
var exec *common.WorkflowExecution
var batchReq *workflowservice.StartBatchOperationRequest
if c.WorkflowId == "" && c.Query == "" {
if c.ActivityId == "" {
return errActivityTarget
}
exec = &common.WorkflowExecution{RunId: c.RunId}
} else {
exec, batchReq, err = opts.workflowExecOrBatch(cctx, c.Parent.Namespace, cl, singleOrBatchOverrides{})
if err != nil {
return err
}
}

if exec != nil { // single workflow operation
if c.ActivityId == "" {
return fmt.Errorf("either --activity-id and --workflow-id, or --query must be set")
return errActivityTarget
}

request := &workflowservice.UnpauseActivityRequest{
Namespace: c.Parent.Namespace,
Execution: &common.WorkflowExecution{
WorkflowId: c.WorkflowId,
RunId: c.RunId,
},
request := &workflowservice.UnpauseActivityExecutionRequest{
Namespace: c.Parent.Namespace,
WorkflowId: c.WorkflowId,
ActivityId: c.ActivityId,
RunId: c.RunId,
ResetAttempts: c.ResetAttempts,
ResetHeartbeat: c.ResetHeartbeats,
Jitter: durationpb.New(c.Jitter.Duration()),
Identity: c.Parent.Identity,
Activity: &workflowservice.UnpauseActivityRequest_Id{Id: c.ActivityId},
}

_, err = cl.WorkflowService().UnpauseActivity(cctx, request)
_, err = cl.WorkflowService().UnpauseActivityExecution(cctx, request)
if err != nil {
return fmt.Errorf("unable to unpause an Activity: %w", err)
}
Expand Down Expand Up @@ -1103,49 +1155,58 @@ func (c *TemporalActivityResetCommand) run(cctx *CommandContext, args []string)
Rps: c.Rps,
}

exec, batchReq, err := opts.workflowExecOrBatch(cctx, c.Parent.Namespace, cl, singleOrBatchOverrides{})
if err != nil {
return err
// Target a workflow Activity (--workflow-id), a standalone Activity (no
// --workflow-id; the latest run unless --run-id is set), or a batch of
// workflow Activities (--query). The server routes to standalone-activity
// handling when the workflow ID is empty, so standalone Activities take the
// single-execution path below.
var exec *common.WorkflowExecution
var batchReq *workflowservice.StartBatchOperationRequest
if c.WorkflowId == "" && c.Query == "" {
if c.ActivityId == "" {
return errActivityTarget
}
exec = &common.WorkflowExecution{RunId: c.RunId}
} else {
exec, batchReq, err = opts.workflowExecOrBatch(cctx, c.Parent.Namespace, cl, singleOrBatchOverrides{})
if err != nil {
return err
}
}

if exec != nil { // single workflow operation
if c.ActivityId == "" {
return fmt.Errorf("either --activity-id and --workflow-id, or --query must be set")
return errActivityTarget
}

request := &workflowservice.ResetActivityRequest{
Activity: &workflowservice.ResetActivityRequest_Id{Id: c.ActivityId},
Namespace: c.Parent.Namespace,
Execution: &common.WorkflowExecution{
WorkflowId: c.WorkflowId,
RunId: c.RunId,
},
Identity: c.Parent.Identity,
KeepPaused: c.KeepPaused,
ResetHeartbeat: c.ResetHeartbeats,
request := &workflowservice.ResetActivityExecutionRequest{
Namespace: c.Parent.Namespace,
WorkflowId: c.WorkflowId,
ActivityId: c.ActivityId,
RunId: c.RunId,
Identity: c.Parent.Identity,
KeepPaused: c.KeepPaused,

@fretz12 fretz12 Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same thing, take in restore orig options

}

resp, err := cl.WorkflowService().ResetActivity(cctx, request)
resp, err := cl.WorkflowService().ResetActivityExecution(cctx, request)
if err != nil {
return fmt.Errorf("unable to reset an Activity: %w", err)
}

resetResponse := struct {
KeepPaused bool `json:"keepPaused"`
ResetHeartbeats bool `json:"resetHeartbeats"`
ServerResponse bool `json:"-"`
KeepPaused bool `json:"keepPaused"`
ServerResponse bool `json:"-"`
}{
ServerResponse: resp != nil,
KeepPaused: c.KeepPaused,
ResetHeartbeats: c.ResetHeartbeats,
ServerResponse: resp != nil,
KeepPaused: c.KeepPaused,
}

_ = cctx.Printer.PrintStructured(resetResponse, printer.StructuredOptions{})
} else { // batch operation
resetActivitiesOperation := &batch.BatchOperationResetActivities{
Identity: c.Parent.Identity,
ResetAttempts: c.ResetAttempts,
ResetHeartbeat: c.ResetHeartbeats,
ResetHeartbeat: true,
KeepPaused: c.KeepPaused,
Jitter: durationpb.New(c.Jitter.Duration()),
RestoreOriginalOptions: c.RestoreOriginalOptions,
Expand Down
Loading
Loading