implement Pause/UnpauseActivityExecution for standalone activities (#9851)

## What changed?
Implement `PauseActivityExecution` and `UnpauseActivityExecution` for
standalone activities. Previously both handlers returned Unimplemented
for the SAA path. They now use chasm.UpdateComponent to apply
pause/unpause state directly to the CHASM Activity component, matching
the semantics of
the existing workflow-activity implementation.

- Proto (`activity_state.proto`): Added `ActivityPauseState` message
(`pause_time`, `identity`, `reason`) and a `pause_state` field on
`ActivityState`.
- `handlePauseRequested`: Sets `PauseState` on the component. If the
activity is in `SCHEDULED` state, increments the attempt stamp so the
existing `ActivityDispatchTask` is invalidated — preventing the activity
from being dispatched to a worker while paused. For `STARTED` activities
the stamp is left unchanged; the worker retains a valid token and
receives `ActivityPaused: true` on its next heartbeat.
- `handleUnpauseRequested`: Clears `PauseState`, optionally resets the
attempt count and/or heartbeat details, and if the activity is
`SCHEDULED` bumps the stamp and enqueues a new `ActivityDispatchTask`
with optional jitter.
- `RecordHeartbeat`: Wires up the `ActivityPaused` response field.
- `buildActivityExecutionInfo`: Maps pause state to
`PENDING_ACTIVITY_STATE_PAUSED` (activity is scheduled but not running)
or `PENDING_ACTIVITY_STATE_PAUSE_REQUESTED` (activity is running on the
worker) in the `RunState` field of `DescribeActivityExecution`.

## Why?
`PauseActivityExecution` / `UnpauseActivityExecution` were already
implemented for workflow-embedded activities via the history service.
Standalone activities had stub handlers that returned `Unimplemented`,
this brings SAA to feature parity with workflow activities for the
pause/unpause lifecycle operations.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [X] added new functional test(s)

## Potential risks
Minimal, this is a translation of an existing api
(Pause/UnpauseActivity)

Author: spkane31

chasm/lib/activity/activity.go

@@ -3,6 +3,7 @@ package activity
 import (
 	"errors"
 	"fmt"
+	"math/rand"
 	"slices"
 	"time"
 
@@ -93,6 +94,19 @@ type RespondCancelledEvent struct {
 	Token   *tokenspb.Task
 }
 
+func (a *Activity) isTerminal() bool {
+	switch a.GetStatus() {
+	case activitypb.ACTIVITY_EXECUTION_STATUS_COMPLETED,
+		activitypb.ACTIVITY_EXECUTION_STATUS_FAILED,
+		activitypb.ACTIVITY_EXECUTION_STATUS_CANCELED,
+		activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED,
+		activitypb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT:
+		return true
+	default:
+		return false
+	}
+}
+
 // LifecycleState implements the chasm.Component interface.
 func (a *Activity) LifecycleState(_ chasm.Context) chasm.LifecycleState {
 	switch a.Status {
@@ -590,8 +604,11 @@ func (a *Activity) handleCancellationRequested(ctx chasm.MutableContext, request
 		return &activitypb.RequestCancelActivityExecutionResponse{}, nil
 	}
 
-	// If in scheduled state, cancel immediately right after marking cancel requested
-	isCancelImmediately := a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED
+	// SCHEDULED and PAUSED activities have no active worker token so cancel immediately.
+	// STARTED and CANCEL_REQUESTED activities wait for the worker to respond.
+	originalStatus := a.GetStatus()
+	isCancelImmediately := originalStatus == activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED ||
+		originalStatus == activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED
 
 	if err := TransitionCancelRequested.Apply(a, ctx, req); err != nil {
 		return nil, err
@@ -611,7 +628,7 @@ func (a *Activity) handleCancellationRequested(ctx chasm.MutableContext, request
 		err = TransitionCanceled.Apply(a, ctx, cancelEvent{
 			details:    details,
 			handler:    metricsHandler,
-			fromStatus: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, // if we're here the original status was scheduled
+			fromStatus: originalStatus,
 		})
 		if err != nil {
 			return nil, err
@@ -621,6 +638,125 @@ func (a *Activity) handleCancellationRequested(ctx chasm.MutableContext, request
 	return &activitypb.RequestCancelActivityExecutionResponse{}, nil
 }
 
+func (a *Activity) handlePauseRequested(ctx chasm.MutableContext, req *activitypb.PauseActivityExecutionRequest) (
+	*activitypb.PauseActivityExecutionResponse, error,
+) {
+	if a.isTerminal() {
+		return nil, serviceerror.NewFailedPreconditionf("activity is in terminal state %v", a.GetStatus())
+	}
+	if a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED {
+		return nil, serviceerror.NewFailedPrecondition("cannot pause an activity with a pending cancellation")
+	}
+	if a.PauseState != nil {
+		return &activitypb.PauseActivityExecutionResponse{}, nil
+	}
+
+	metricsHandler, err := a.enrichMetricsHandler(ctx, metrics.ActivityPausedScope)
+	if err != nil {
+		return nil, err
+	}
+
+	if TransitionPaused.Possible(a) {
+		// SCHEDULED → real PAUSED status; stamp bumped to invalidate the pending dispatch task.
+		if err := TransitionPaused.Apply(a, ctx, pauseEvent{
+			req:            req.GetFrontendRequest(),
+			metricsHandler: metricsHandler,
+		}); err != nil {
+			return nil, err
+		}
+		return &activitypb.PauseActivityExecutionResponse{}, nil
+	}
+	// STARTED → flag-only pause. Status stays STARTED so the worker's token remains valid.
+	// The worker will see ActivityPaused=true on the next heartbeat.
+	a.pause(ctx, pauseEvent{req.GetFrontendRequest(), metricsHandler})
+	return &activitypb.PauseActivityExecutionResponse{}, nil
+}
+
+func (a *Activity) handleUnpauseRequested(ctx chasm.MutableContext, req *activitypb.UnpauseActivityExecutionRequest) (
+	*activitypb.UnpauseActivityExecutionResponse, error,
+) {
+	if a.isTerminal() {
+		return nil, serviceerror.NewFailedPreconditionf("activity is in terminal state %v", a.GetStatus())
+	}
+	// Not paused → no-op.
+	if a.PauseState == nil {
+		return &activitypb.UnpauseActivityExecutionResponse{}, nil
+	}
+
+	metricsHandler, err := a.enrichMetricsHandler(ctx, metrics.ActivityUnpausedScope)
+	if err != nil {
+		return nil, err
+	}
+
+	if TransitionUnpaused.Possible(a) {
+		if err := TransitionUnpaused.Apply(a, ctx, unpauseEvent{
+			req:            req.GetFrontendRequest(),
+			metricsHandler: metricsHandler,
+		}); err != nil {
+			return nil, err
+		}
+		return &activitypb.UnpauseActivityExecutionResponse{}, nil
+	}
+
+	// Flag-based pause (status is STARTED, CANCEL_REQUESTED, or SCHEDULED after retry while paused).
+	if a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_STARTED ||
+		a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED {
+		// Worker continues with its existing token — no stamp bump needed, no dispatch task.
+		// Cancel takes precedence over pause. Unpause clears the pause flag but does not re-dispatch;
+		// the activity remains CANCEL_REQUESTED and will be cancelled when the worker responds.
+		a.PauseState = nil
+		a.emitOnUnpausedMetrics(metricsHandler)
+		return &activitypb.UnpauseActivityExecutionResponse{}, nil
+	}
+	a.unpause(ctx, unpauseEvent{
+		req:            req.GetFrontendRequest(),
+		metricsHandler: metricsHandler,
+	})
+	return &activitypb.UnpauseActivityExecutionResponse{}, nil
+}
+
+func (a *Activity) unpause(
+	ctx chasm.MutableContext,
+	event unpauseEvent,
+) {
+	a.PauseState = nil
+	attempt := a.LastAttempt.Get(ctx)
+	if event.req.GetResetAttempts() {
+		attempt.Count = 1
+	}
+	if event.req.GetResetHeartbeat() {
+		a.LastHeartbeat = chasm.NewDataField(ctx, &activitypb.ActivityHeartbeatState{})
+	}
+	attempt.Stamp++
+	attempt.CurrentRetryInterval = nil
+	scheduleTime := ctx.Now(a)
+	if jitter := event.req.GetJitter().AsDuration(); jitter > 0 {
+		scheduleTime = scheduleTime.Add(time.Duration(rand.Int63n(int64(jitter)))) //nolint:gosec
+	}
+	if timeout := a.GetScheduleToStartTimeout().AsDuration(); timeout > 0 {
+		ctx.AddTask(
+			a,
+			chasm.TaskAttributes{ScheduledTime: scheduleTime.Add(timeout)},
+			&activitypb.ScheduleToStartTimeoutTask{Stamp: attempt.GetStamp()})
+	}
+	ctx.AddTask(
+		a,
+		chasm.TaskAttributes{ScheduledTime: scheduleTime},
+		&activitypb.ActivityDispatchTask{Stamp: attempt.GetStamp()})
+	a.emitOnUnpausedMetrics(event.metricsHandler)
+}
+func (a *Activity) pause(
+	ctx chasm.MutableContext,
+	event pauseEvent,
+) {
+	a.PauseState = &activitypb.ActivityPauseState{
+		PauseTime: timestamppb.New(ctx.Now(a)),
+		Identity:  event.req.GetIdentity(),
+		Reason:    event.req.GetReason(),
+	}
+	a.emitOnPausedMetrics(event.metricsHandler)
+}
+
 // recordScheduleToStartOrCloseTimeoutFailure records schedule-to-start or schedule-to-close timeouts. Such timeouts are not retried so we
 // set the outcome failure directly and leave the attempt failure as is.
 func (a *Activity) recordScheduleToStartOrCloseTimeoutFailure(ctx chasm.MutableContext, timeoutType enumspb.TimeoutType) error {
@@ -671,7 +807,9 @@ func (a *Activity) recordFailedAttempt(
 }
 
 // tryReschedule attempts to reschedule the activity for retry. Returns true if rescheduled, false
-// if retry is not possible.
+// if retry is not possible. When the activity has PauseState set (flag-based pause from STARTED),
+// the retry transitions to SCHEDULED normally but the dispatch task is blocked by the pause flag
+// until the activity is unpaused.
 func (a *Activity) tryReschedule(
 	ctx chasm.MutableContext,
 	overridingRetryInterval time.Duration,
@@ -767,7 +905,8 @@ func (a *Activity) RecordHeartbeat(
 	}
 	return &historyservice.RecordActivityTaskHeartbeatResponse{
 		CancelRequested: a.Status == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
-		// TODO(saa-preview): ActivityPaused, ActivityReset
+		ActivityPaused:  a.PauseState != nil,
+		// TODO(saa-preview): ActivityReset
 	}, nil
 }
 
@@ -776,7 +915,8 @@ func InternalStatusToAPIStatus(status activitypb.ActivityExecutionStatus) enumsp
 	switch status {
 	case activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
 		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
-		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED:
+		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
+		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED:
 		return enumspb.ACTIVITY_EXECUTION_STATUS_RUNNING
 	case activitypb.ACTIVITY_EXECUTION_STATUS_COMPLETED:
 		return enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED
@@ -803,6 +943,8 @@ func internalStatusToRunState(status activitypb.ActivityExecutionStatus) enumspb
 		return enumspb.PENDING_ACTIVITY_STATE_STARTED
 	case activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED:
 		return enumspb.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED:
+		return enumspb.PENDING_ACTIVITY_STATE_PAUSED
 	case activitypb.ACTIVITY_EXECUTION_STATUS_COMPLETED,
 		activitypb.ACTIVITY_EXECUTION_STATUS_FAILED,
 		activitypb.ACTIVITY_EXECUTION_STATUS_CANCELED,
@@ -816,9 +958,20 @@ func internalStatusToRunState(status activitypb.ActivityExecutionStatus) enumspb
 }
 
 func (a *Activity) buildActivityExecutionInfo(ctx chasm.Context) *apiactivitypb.ActivityExecutionInfo {
-	// TODO(saa-preview): support pause states
 	status := InternalStatusToAPIStatus(a.GetStatus())
-	runState := internalStatusToRunState(a.GetStatus())
+	// Derive the external run state with hybrid pause logic:
+	//   PAUSED status (real) → PAUSED
+	//   STARTED + PauseState != nil (pause requested while running) → PAUSE_REQUESTED
+	//   SCHEDULED + PauseState != nil (retry while paused flag set) → PAUSED
+	//   All other cases → derived from internal status directly
+	var runState enumspb.PendingActivityState
+	if a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_STARTED && a.PauseState != nil {
+		runState = enumspb.PENDING_ACTIVITY_STATE_PAUSE_REQUESTED
+	} else if a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED && a.PauseState != nil {
+		runState = enumspb.PENDING_ACTIVITY_STATE_PAUSED
+	} else {
+		runState = internalStatusToRunState(a.GetStatus())
+	}
 
 	requestData := a.RequestData.Get(ctx)
 	attempt := a.LastAttempt.Get(ctx)
@@ -1104,6 +1257,20 @@ func (a *Activity) emitOnTimedOutMetrics(
 	metrics.ActivityTimeout.With(handler).Record(1, timeoutTag)
 }
 
+func (a *Activity) emitOnPausedMetrics(
+	handler metrics.Handler,
+) {
+	metrics.ActivityPauseRequests.With(handler).Record(1)
+	metrics.ActivityPause.With(handler).Record(1)
+}
+
+func (a *Activity) emitOnUnpausedMetrics(
+	handler metrics.Handler,
+) {
+	metrics.ActivityUnpauseRequests.With(handler).Record(1)
+	metrics.ActivityUnpause.With(handler).Record(1)
+}
+
 // SearchAttributes implements chasm.VisibilitySearchAttributesProvider interface.
 // Returns the current search attribute values for this activity execution.
 func (a *Activity) SearchAttributes(_ chasm.Context) []chasm.SearchAttributeKeyValue {

chasm/lib/activity/activity_tasks.go

@@ -36,7 +36,11 @@ func (h *activityDispatchTaskHandler) Validate(
 	task *activitypb.ActivityDispatchTask,
 ) (bool, error) {
 	// TODO(saa-preview): make sure we handle resets when we support them, as they will reset the attempt count
+	// Do not dispatch while the activity has a pause flag set (SCHEDULED + PauseState from a retry
+	// while a STARTED activity was flag-paused). TransitionStarted.Possible already returns false for
+	// real PAUSED status activities (source must be SCHEDULED, and PAUSED → SCHEDULED via unpause).
 	return (TransitionStarted.Possible(activity) &&
+		activity.PauseState == nil &&
 		task.Stamp == activity.LastAttempt.Get(ctx).GetStamp()), nil
 }
 
@@ -93,7 +97,9 @@ func (h *scheduleToStartTimeoutTaskHandler) Validate(
 	_ chasm.TaskAttributes,
 	task *activitypb.ScheduleToStartTimeoutTask,
 ) (bool, error) {
+	// Do not time out a SCHEDULED activity that has the pause flag set (retry while paused).
 	return (activity.Status == activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED &&
+		activity.PauseState == nil &&
 		task.Stamp == activity.LastAttempt.Get(ctx).GetStamp()), nil
 }
 

chasm/lib/activity/frontend.go

@@ -468,7 +468,15 @@ func (h *frontendHandler) PauseActivityExecution(
 		return nil, ErrStandaloneActivityDisabled
 	}
 
-	// TODO: validate request fields (e.g. namespace, identity length)
+	if err := validatePauseActivityExecutionRequest(
+		req,
+		h.config.MaxIDLengthLimit(),
+		h.config.BlobSizeLimitError,
+		h.config.BlobSizeLimitWarn,
+		h.logger); err != nil {
+		return nil, err
+	}
+
 	namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(req.GetNamespace()))
 	if err != nil {
 		return nil, err
@@ -492,7 +500,10 @@ func (h *frontendHandler) UnpauseActivityExecution(
 		return nil, ErrStandaloneActivityDisabled
 	}
 
-	// TODO: validate request fields (e.g. namespace, identity length)
+	if err := validateUnpauseActivityExecutionRequest(req, h.config.MaxIDLengthLimit()); err != nil {
+		return nil, err
+	}
+
 	namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(req.GetNamespace()))
 	if err != nil {
 		return nil, err