[Experiment] Schedule standalone activities with BUFFER_LATEST

chasm/lib/scheduler/action.go

@@ -58,7 +58,12 @@ type actionImplementation interface {
 	ParticipatesInCompletionHistory() bool
 }
 
-func implementation(_ *schedulepb.ScheduleAction) actionImplementation { return workflowAction{} }
+func implementation(action *schedulepb.ScheduleAction) actionImplementation {
+	if action.GetStartActivity() != nil {
+		return activityAction{}
+	}
+	return workflowAction{}
+}
 
 func (s *Scheduler) actionMetadata() actionMetadata {
 	return implementation(s.Schedule.GetAction()).Metadata(s.Schedule.GetAction())

chasm/lib/scheduler/action_activity.go

@@ -0,0 +1,84 @@
+package scheduler
+
+import (
+	"context"
+
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	failurepb "go.temporal.io/api/failure/v1"
+	schedulepb "go.temporal.io/api/schedule/v1"
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/api/workflowservice/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/internal"
+)
+
+type activityAction struct{}
+
+func (activityAction) Metadata(action *schedulepb.ScheduleAction) actionMetadata {
+	spec := action.GetStartActivity()
+	return actionMetadata{Kind: enumspb.EXECUTION_TYPE_ACTIVITY, Type: spec.GetActivityType().GetName(), IDBase: spec.GetActivityId(), TaskQueue: spec.GetTaskQueue().GetName(), SearchAttributes: spec.GetSearchAttributes()}
+}
+func (activityAction) Validate(action *schedulepb.ScheduleAction) error {
+	spec := action.GetStartActivity()
+	if spec == nil || spec.GetActivityId() == "" || spec.GetActivityType().GetName() == "" || spec.GetTaskQueue().GetName() == "" {
+		return serviceerror.NewInvalidArgument("scheduled activity requires ID, type, and task queue")
+	}
+	if spec.GetScheduleToCloseTimeout().AsDuration() <= 0 && spec.GetStartToCloseTimeout().AsDuration() <= 0 {
+		return serviceerror.NewInvalidArgument("scheduled activity requires a positive schedule-to-close or start-to-close timeout")
+	}
+	return nil
+}
+func (activityAction) Policies() *internal.PolicyRegistry { return internal.ActivityPolicies() }
+func (activityAction) GenerateTargetID(base string, occurrence occurrenceContext) string {
+	return internal.GenerateTimestampTargetID(base, occurrence.NominalTime)
+}
+func (activityAction) ParticipatesInCompletionHistory() bool { return false }
+func (activityAction) Start(ctx context.Context, clients actionClients, input actionStartInput) (string, error) {
+	scheduler, start := input.Scheduler, input.Occurrence
+	spec := scheduler.Schedule.GetAction().GetStartActivity()
+	reuse := enumspb.ACTIVITY_ID_REUSE_POLICY_REJECT_DUPLICATE
+	if start.Manual {
+		reuse = enumspb.ACTIVITY_ID_REUSE_POLICY_ALLOW_DUPLICATE
+	}
+	response, err := clients.Frontend.StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{
+		Namespace: scheduler.Namespace, Identity: scheduler.identity(), RequestId: start.RequestId,
+		ActivityId: internal.TargetID(start), ActivityType: spec.ActivityType, TaskQueue: spec.TaskQueue,
+		ScheduleToCloseTimeout: spec.ScheduleToCloseTimeout, ScheduleToStartTimeout: spec.ScheduleToStartTimeout,
+		StartToCloseTimeout: spec.StartToCloseTimeout, HeartbeatTimeout: spec.HeartbeatTimeout, RetryPolicy: spec.RetryPolicy,
+		Input: spec.Input, IdReusePolicy: reuse, IdConflictPolicy: enumspb.ACTIVITY_ID_CONFLICT_POLICY_FAIL,
+		SearchAttributes: spec.SearchAttributes, Header: spec.Header,
+		UserMetadata: spec.UserMetadata, Priority: spec.Priority, CompletionCallbacks: []*commonpb.Callback{input.Callback}, StartDelay: spec.StartDelay,
+	})
+	if err != nil {
+		return "", err
+	}
+	return response.RunId, nil
+}
+func (activityAction) Cancel(context.Context, actionClients, *Scheduler, *commonpb.Execution) error {
+	return serviceerror.NewInvalidArgument("scheduled activities do not support cancellation overlap policies")
+}
+func (activityAction) Terminate(ctx context.Context, clients actionClients, scheduler *Scheduler, target *commonpb.Execution) error {
+	_, err := clients.Frontend.TerminateActivityExecution(ctx, &workflowservice.TerminateActivityExecutionRequest{Namespace: scheduler.Namespace, ActivityId: target.BusinessId, RunId: target.RunId, Reason: "terminated by schedule overlap policy", Identity: scheduler.identity()})
+	return err
+}
+func (activityAction) Completion(info *persistencespb.ChasmNexusCompletion, execution *commonpb.Execution) actionCompletion {
+	status := enumspb.ACTIVITY_EXECUTION_STATUS_FAILED
+	if _, ok := info.Outcome.(*persistencespb.ChasmNexusCompletion_Success); ok {
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED
+	}
+	if failure := info.GetFailure(); failure != nil {
+		switch failure.FailureInfo.(type) {
+		case *failurepb.Failure_CanceledFailureInfo:
+			status = enumspb.ACTIVITY_EXECUTION_STATUS_CANCELED
+		case *failurepb.Failure_TimeoutFailureInfo:
+			status = enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT
+		case *failurepb.Failure_TerminatedFailureInfo:
+			status = enumspb.ACTIVITY_EXECUTION_STATUS_TERMINATED
+		default:
+			status = enumspb.ACTIVITY_EXECUTION_STATUS_FAILED
+		}
+	}
+	failed := status == enumspb.ACTIVITY_EXECUTION_STATUS_FAILED || status == enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT
+	return actionCompletion{Result: &commonpb.ActionExecutionResult{Execution: execution, Status: &commonpb.ActionExecutionResult_ActivityStatus{ActivityStatus: status}}, Failed: failed}
+}

chasm/lib/scheduler/action_completion_internal_test.go

@@ -0,0 +1,72 @@
+package scheduler
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	failurepb "go.temporal.io/api/failure/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+)
+
+func TestActionCompletion(t *testing.T) {
+	execution := &commonpb.Execution{Type: enumspb.EXECUTION_TYPE_ACTIVITY, BusinessId: "activity-id", RunId: "run-id"}
+	activityCases := []struct {
+		name   string
+		info   *persistencespb.ChasmNexusCompletion
+		status enumspb.ActivityExecutionStatus
+		failed bool
+	}{
+		{name: "success", info: &persistencespb.ChasmNexusCompletion{Outcome: &persistencespb.ChasmNexusCompletion_Success{}}, status: enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED},
+		{name: "application failure", info: completionWithFailure(&failurepb.Failure{}), status: enumspb.ACTIVITY_EXECUTION_STATUS_FAILED, failed: true},
+		{name: "timeout", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TimeoutFailureInfo{TimeoutFailureInfo: &failurepb.TimeoutFailureInfo{}}}), status: enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT, failed: true},
+		{name: "canceled", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_CanceledFailureInfo{CanceledFailureInfo: &failurepb.CanceledFailureInfo{}}}), status: enumspb.ACTIVITY_EXECUTION_STATUS_CANCELED},
+		{name: "terminated", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TerminatedFailureInfo{TerminatedFailureInfo: &failurepb.TerminatedFailureInfo{}}}), status: enumspb.ACTIVITY_EXECUTION_STATUS_TERMINATED},
+	}
+	for _, tc := range activityCases {
+		t.Run(tc.name, func(t *testing.T) {
+			completion := (activityAction{}).Completion(tc.info, execution)
+			require.Equal(t, tc.failed, completion.Failed)
+			require.Equal(t, execution, completion.Result.GetExecution())
+			require.Equal(t, tc.status, completion.Result.GetActivityStatus())
+			require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED, completion.Result.GetWorkflowStatus())
+		})
+	}
+
+	workflowExecution := &commonpb.Execution{Type: enumspb.EXECUTION_TYPE_WORKFLOW, BusinessId: "workflow-id", RunId: "run-id"}
+	workflowCases := []struct {
+		name   string
+		info   *persistencespb.ChasmNexusCompletion
+		status enumspb.WorkflowExecutionStatus
+		failed bool
+	}{
+		{name: "success", info: &persistencespb.ChasmNexusCompletion{Outcome: &persistencespb.ChasmNexusCompletion_Success{}}, status: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED},
+		{name: "application failure", info: completionWithFailure(&failurepb.Failure{}), status: enumspb.WORKFLOW_EXECUTION_STATUS_FAILED, failed: true},
+		{name: "timeout", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TimeoutFailureInfo{TimeoutFailureInfo: &failurepb.TimeoutFailureInfo{}}}), status: enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT, failed: true},
+		{name: "canceled", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_CanceledFailureInfo{CanceledFailureInfo: &failurepb.CanceledFailureInfo{}}}), status: enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED},
+		{name: "terminated", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TerminatedFailureInfo{TerminatedFailureInfo: &failurepb.TerminatedFailureInfo{}}}), status: enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED},
+	}
+	for _, tc := range workflowCases {
+		t.Run("workflow/"+tc.name, func(t *testing.T) {
+			completion := (workflowAction{}).Completion(tc.info, workflowExecution)
+			require.Equal(t, tc.failed, completion.Failed)
+			require.Equal(t, workflowExecution, completion.Result.GetExecution())
+			require.Equal(t, tc.status, completion.Result.GetWorkflowStatus())
+			require.Equal(t, enumspb.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED, completion.Result.GetActivityStatus())
+		})
+	}
+}
+
+func completionWithFailure(failure *failurepb.Failure) *persistencespb.ChasmNexusCompletion {
+	return &persistencespb.ChasmNexusCompletion{Outcome: &persistencespb.ChasmNexusCompletion_Failure{Failure: failure}}
+}
+
+func TestActionTargetID(t *testing.T) {
+	nominal := time.Date(2024, time.January, 2, 3, 4, 5, 900000000, time.FixedZone("test", -6*60*60))
+	want := "base-2024-01-02T09:04:05Z"
+	occurrence := occurrenceContext{NominalTime: nominal}
+	require.Equal(t, want, (workflowAction{}).GenerateTargetID("base", occurrence))
+	require.Equal(t, want, (activityAction{}).GenerateTargetID("base", occurrence))
+}

chasm/lib/scheduler/action_results.go

@@ -1,12 +1,18 @@
 package scheduler
 
 import (
+	"fmt"
+	"strings"
+
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	schedulepb "go.temporal.io/api/schedule/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
 	schedulespb "go.temporal.io/server/api/schedule/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/metrics"
 )
 
 func actionResult(start *schedulespb.BufferedStart) *schedulepb.ScheduleActionResult {
@@ -17,3 +23,32 @@ func actionResult(start *schedulespb.BufferedStart) *schedulepb.ScheduleActionRe
 	}
 	return result
 }
+
+func (s *Scheduler) recordActivityCompletion(ctx chasm.MutableContext, invoker *Invoker, start *schedulespb.BufferedStart, info *persistencespb.ChasmNexusCompletion) error {
+	execution := common.CloneProto(internal.Execution(start))
+	for _, link := range info.GetLinks() {
+		if activity := link.GetActivity(); activity != nil && activity.GetActivityId() == execution.BusinessId {
+			if execution.RunId != "" && execution.RunId != activity.GetRunId() {
+				return nil
+			}
+			execution.RunId = activity.GetRunId()
+		}
+	}
+	completion := implementation(s.Schedule.GetAction()).Completion(info, execution)
+	start.Completion = completion.Result
+	start.CompletionTime = info.CloseTime
+	if start.GetStartTime() == nil && info.StartTime != nil {
+		start.StartTime = info.StartTime
+	}
+	if internal.TracksExecution(start) && completion.Failed && s.Schedule.GetPolicies().GetPauseOnFailure() && !s.Schedule.GetState().GetPaused() {
+		s.Schedule.State.Paused = true
+		s.Schedule.State.Notes = fmt.Sprintf("paused, activity %s: %s", strings.ToLower(completion.Result.GetActivityStatus().String()), execution.BusinessId)
+		s.updateConflictToken()
+	}
+	if closeTime := info.GetCloseTime().AsTime(); !closeTime.IsZero() {
+		newTaggedMetricsHandler(ctx.MetricsHandler(), s).Timer(metrics.ScheduleCallbackLatency.Name()).Record(max(0, ctx.Now(s).Sub(closeTime)))
+	}
+	invoker.releaseWaiting(ctx, info.CloseTime)
+	s.Generator.Get(ctx).Generate(ctx)
+	return nil
+}

chasm/lib/scheduler/internal/policy.go

@@ -133,6 +133,25 @@ func BuiltinPolicy(policy enumspb.ScheduleOverlapPolicy) PolicyDefinition {
 	}
 }
 
+func BufferLatestPolicy() PolicyDefinition {
+	return PolicyDefinition{Identity: PolicyIdentity{Custom: BufferLatestPolicyName}, Plan: func(s PolicySnapshot) PolicyDecision {
+		if len(s.Running) == 0 && s.Selected == nil {
+			return PolicyDecision{Start: true}
+		}
+		decision := PolicyDecision{Wait: true}
+		for _, waiting := range s.Waiting {
+			if waiting.CustomOverlapPolicy != BufferLatestPolicyName {
+				continue
+			}
+			if waiting.ActualTime.After(s.Occurrence.ActualTime) || waiting.ActualTime.Equal(s.Occurrence.ActualTime) && waiting.Occurrence > s.Occurrence.Occurrence {
+				return PolicyDecision{}
+			}
+			decision.Replace = append(decision.Replace, waiting)
+		}
+		return decision
+	}}
+}
+
 func WorkflowPolicies() *PolicyRegistry {
 	definitions := make([]PolicyDefinition, 0, 6)
 	for _, policy := range []enumspb.ScheduleOverlapPolicy{enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL} {
@@ -145,6 +164,19 @@ func WorkflowPolicies() *PolicyRegistry {
 	return r
 }
 
+func ActivityPolicies() *PolicyRegistry {
+	definitions := make([]PolicyDefinition, 0, 6)
+	for _, policy := range []enumspb.ScheduleOverlapPolicy{enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL} {
+		definitions = append(definitions, BuiltinPolicy(policy))
+	}
+	definitions = append(definitions, BufferLatestPolicy())
+	r, err := NewPolicyRegistry(definitions, PolicyIdentity{}, ExecutionOperations{Terminate: true})
+	if err != nil {
+		log.NewCLILogger().Fatal("invalid scheduler overlap policy registry", tag.Error(err))
+	}
+	return r
+}
+
 func (r *PolicyRegistry) clone() *PolicyRegistry {
 	return &PolicyRegistry{policies: maps.Clone(r.policies), defaultPolicy: r.defaultPolicy}
 }

chasm/lib/scheduler/internal/policy_test.go

@@ -25,6 +25,51 @@ func TestPolicyRegistrySubsetWithoutDefault(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestActivityPolicySelections(t *testing.T) {
+	registry := ActivityPolicies()
+	for _, policy := range []enumspb.ScheduleOverlapPolicy{enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER} {
+		_, err := registry.Resolve(PolicyIdentity{}, PolicyIdentity{Builtin: policy})
+		require.NoError(t, err)
+	}
+	for _, selection := range []PolicyIdentity{{}, {Builtin: enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER}, {Custom: "unknown"}, {Builtin: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, Custom: BufferLatestPolicyName}} {
+		_, err := registry.Resolve(PolicyIdentity{}, selection)
+		require.Error(t, err)
+	}
+	_, err := registry.Resolve(PolicyIdentity{}, PolicyIdentity{Custom: BufferLatestPolicyName})
+	require.NoError(t, err)
+	_, err = WorkflowPolicies().Resolve(PolicyIdentity{}, PolicyIdentity{Custom: BufferLatestPolicyName})
+	require.Error(t, err)
+}
+
+func TestBufferLatestReplacesOnlyOlderWaitingOccurrences(t *testing.T) {
+	now := time.Date(2026, 9, 4, 12, 0, 0, 0, time.UTC)
+	latest := func(id string, at time.Time) BufferedStartSnapshot {
+		return BufferedStartSnapshot{RequestID: id, ActualTime: at, CustomOverlapPolicy: BufferLatestPolicyName}
+	}
+	for _, active := range []bool{false, true} {
+		t.Run(map[bool]string{false: "selected", true: "running"}[active], func(t *testing.T) {
+			starts := []BufferedStartSnapshot{latest("first", now), latest("second", now.Add(time.Second)), {RequestID: "other-policy", ActualTime: now, OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL}, latest("older", now.Add(-time.Second)), latest("tie-newer-insertion", now.Add(time.Second))}
+			snapshot := BufferProcessingSnapshot{Starts: starts, Policies: ActivityPolicies(), CatchupWindow: time.Hour, LimitedActions: true, RemainingActions: 3}
+			if active {
+				snapshot.RunningExecutions = []ExecutionSnapshot{{TargetID: "active", RunID: "run"}}
+			}
+			plan := PlanBufferProcessing(snapshot, now)
+			require.EqualValues(t, map[bool]int{false: 2, true: 3}[active], plan.OverlapSkipped)
+			require.Equal(t, plan.OverlapSkipped, plan.OverlapSkippedByCustomPolicy[BufferLatestPolicyName])
+			decisions := make(map[string]BufferDecision)
+			for _, decision := range plan.Decisions {
+				decisions[decision.RequestID] = decision
+			}
+			require.Equal(t, BufferDecisionDefer, decisions["tie-newer-insertion"].Action)
+			require.Equal(t, BufferDecisionDefer, decisions["other-policy"].Action)
+			require.Equal(t, BufferDecisionDiscard, decisions["older"].Action)
+			require.False(t, decisions["second"].ConsumesScheduledAction)
+			require.EqualValues(t, 3, snapshot.RemainingActions)
+			require.Equal(t, starts, snapshot.Starts)
+		})
+	}
+}
+
 func TestPolicyPlannerReceivesDetachedSnapshots(t *testing.T) {
 	id := PolicyIdentity{Custom: "test.only"}
 	registry, err := NewPolicyRegistry([]PolicyDefinition{{Identity: id, Plan: func(snapshot PolicySnapshot) PolicyDecision {

chasm/lib/scheduler/scheduler.go

@@ -739,6 +739,9 @@ func (s *Scheduler) HandleNexusCompletion(
 		)
 		return nil
 	}
+	if s.Schedule.GetAction().GetStartActivity() != nil {
+		return s.recordActivityCompletion(ctx, invoker, start, info)
+	}
 	workflowID := start.GetWorkflowId()
 	tracksCompletionResult := implementation(s.Schedule.GetAction()).ParticipatesInCompletionHistory() && internal.TracksExecution(start)