Align CHASM ALLOW_ALL lifecycle with V1 (#11631)

## What changed?
- Keep attaching completion callbacks to every CHASM scheduler workflow
start, including actions whose resolved overlap policy is `ALLOW_ALL`,
so start requests remain safe across rolling upgrades.
- Exclude scheduler-wide last-completion result/failure input from new
`ALLOW_ALL` actions.
- After `StartWorkflowExecution` succeeds, remove the new `ALLOW_ALL`
`BufferedStart` in the same transaction. Separately, copy its start data
to `ScheduleInfo.RecentActions` as a start-only `RUNNING` history
record.
- Resolve an unspecified pending-start policy while building the
V1-to-V2 migration request, so every V2 dispatch and retry uses the
policy persisted on the buffered start.
- For a pre-existing callback, use the policy stamped on that start
rather than resolving against the schedule's current policy; an
unstamped V1-migrated running workflow remains tracked, as it was in V1.
- Preserve compatibility for previously persisted `ALLOW_ALL` callbacks
without updating last-completion state or `PauseOnFailure`.
- Carry start-only history through CHASM-to-V1 migration.
- Merge start-only and completion-tracked history by actual start time,
retaining the newest ten actions across both sources.
- Re-run generation after recording a start-only action, so a finite or
manual-only schedule rearms its idle timer from that start.

## Behavior

This makes CHASM match modern V1:

- A new `ALLOW_ALL` workflow starts normally, but is absent from
`DescribeSchedule.Info.RunningWorkflows` and does not consume
active-buffer capacity.
- It appears in `DescribeSchedule.Info.RecentActions` (and List's recent
actions) with its start time, workflow execution, and `RUNNING` status.
- Its completion callback remains attached for rolling-upgrade
compatibility. Because the start has already moved out of active
buffered state, the callback is ignored and success or failure cannot
change shared completion input or `PauseOnFailure`; its recent status
therefore remains `RUNNING`.
- For a final `ALLOW_ALL` action, an idle task armed before the start is
invalidated by the newer start time; generation immediately arms its
replacement, so the schedule still closes after `IdleTime`.
- If an older handler retained an `ALLOW_ALL` start, its callback
remains a compatibility path: the terminal action record is retained,
but its completion cannot update shared completion state or pause the
schedule once handled by this version.
- `DescribeSchedule.Info.RecentActions` is ordered by actual start time
and bounded to the newest ten actions across start-only and
completion-tracked history. Newer retained actions evict the oldest;
completion does not remove a start-only action.
- Non-`ALLOW_ALL` actions remain active and appear in `RunningWorkflows`
until their completion is handled.
- A V1 pending start with an unspecified override snapshots the
schedule's effective policy when its V2 migration request is built.
- A V1-migrated `RunningWorkflows` entry has an unspecified policy but
is nevertheless tracked: its completion updates sequential state and may
pause the schedule, regardless of the schedule's current `ALLOW_ALL`
default.

## Why?

`ALLOW_ALL` actions are independent executions. Tracking their
completions made scheduler-wide last result/failure and `PauseOnFailure`
depend on callback arrival order, and retaining them as active could
affect overlap and capacity behavior.

Callbacks remain attached because `StartWorkflowExecution` deduplicates
by request ID without reconciling callback differences. Keeping the
request callback-compatible prevents a mixed-version retry from
retaining a start that waits for a callback the workflow never received.

## How did you test it?

- [x] added unit coverage
- [x] added shared V1/CHASM functional coverage

Commands:

- `go test -tags test_dep ./chasm/lib/scheduler/... -count=1`
- `go test -tags test_dep ./tests -run
'^TestSchedule(CHASM|V1)/TestAllowAllDescribeContract$' -count=1`
- `make fmt-imports`
- `git diff --check`
- `env GOCACHE=/tmp/sch-038-gocache go vet -tags
disable_grpc_modules,,test_dep -vettool=.bin/errortype
-style-check=false ./chasm/lib/scheduler/...`

The functional test uses workflow signals to control completion. It
asserts counters, buffer size, recent-action status/timestamps, active
workflows, and failure/pause isolation across both backends. A CHASM
functional idle-close case verifies that a final `ALLOW_ALL` action
still closes after `IdleTime`. The migration regression and
callback-reason metric test drive real component transactions through
the CHASM test engine.

`make lint-code` currently exits before analysis with `no go files to
analyze` from its `--new-from-rev` filter, despite the Go diff; package
`go vet` is clean.

## Additional observability

Ignored callbacks are tagged as either `unrecognized_request_id` or
`already_completed`. A newly started `ALLOW_ALL` callback is expected to
be unrecognized after its buffered start moves to start-only history. An
already-completed callback is a valid redelivery (for example, after a
workflow reset). ~~Both preserve scheduler state while emitting a
warning and counter increment.~~ Missing request IDs are now metric-only
because they include expected `ALLOW_ALL` callbacks; known
`already_completed` redeliveries still emit the warning, event, and
counter.

## Potential risks

Keeping callbacks attached avoids permanently orphaning buffered starts
when old and new binaries race on the same request ID. During a rolling
upgrade, however, old and new requests still differ in whether
scheduler-wide last-completion result/failure input is included for
`ALLOW_ALL`; the request that wins deduplication determines whether that
workflow receives the legacy input. An older callback handler can also
temporarily apply the legacy completion and pause semantics. These
mixed-version differences end once the rollout completes, and the
callback ensures an older handler cannot wait indefinitely.

New `ALLOW_ALL` terminal status is intentionally not reflected in
schedule Describe/List results, as in modern V1. ~~Its expected late
callback is recorded as `unrecognized_request_id`, which adds callback
delivery plus warning/metric volume compared with omitting callbacks.~~
Its expected late callback remains recorded as
`unrecognized_request_id`, but only as a metric; warning and event
logging are suppressed until these callbacks can be safely omitted.

Migration resolves an unspecified pending-start policy at the V1-to-V2
boundary. If the schedule policy changes while that start remains
pending and the schedule then rolls back to V1, the explicit migrated
policy is preserved instead of inheriting the newer schedule policy.
This is a narrow semantic difference that keeps V2 dispatch and retry
behavior stable.

---------

Co-authored-by: David Porter <david.porter@temporal.io>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

chasm/chasmtest/test_engine.go

@@ -78,6 +78,13 @@ func WithTimeSource(ts clock.TimeSource) EngineOption {
 	}
 }
 
+// WithMetricsHandler overrides the engine's default no-op metrics handler.
+func WithMetricsHandler(handler metrics.Handler) EngineOption {
+	return func(e *Engine) {
+		e.metrics = handler
+	}
+}
+
 var defaultTransitionOptions = chasm.TransitionOptions{
 	ReusePolicy:    chasm.BusinessIDReusePolicyAllowDuplicate,
 	ConflictPolicy: chasm.BusinessIDConflictPolicyFail,

chasm/lib/scheduler/export_test.go

@@ -65,13 +65,20 @@ func (i *Invoker) RecordExecuteResult(
 	ctx chasm.MutableContext,
 	completed []*schedulespb.BufferedStart,
 	retryable []*schedulespb.BufferedStart,
-) (newlyStarted, droppedDuplicates int) {
+) (newlyStarted, droppedDuplicates int, startOnlyActions []*schedulespb.BufferedStart) {
 	return i.recordExecuteResult(ctx, &executeResult{
 		CompletedStarts: completed,
 		RetryableStarts: retryable,
 	})
 }
 
+func (s *Scheduler) RecordStartOnlyActions(
+	ctx chasm.MutableContext,
+	starts []*schedulespb.BufferedStart,
+) {
+	s.recordStartOnlyActions(ctx, starts)
+}
+
 func (b *BackfillerTaskHandler) ProcessBackfill(
 	scheduler *Scheduler,
 	backfiller *Backfiller,

chasm/lib/scheduler/helper_test.go

@@ -216,6 +216,12 @@ func withEngineTimeSource(ts *clock.EventTimeSource) engineTestOption {
 	}
 }
 
+func withEngineMetricsHandler(handler metrics.Handler) engineTestOption {
+	return func(c *engineTestConfig) {
+		c.engineOpts = append(c.engineOpts, chasmtest.WithMetricsHandler(handler))
+	}
+}
+
 func newEngineTestConfig(opts ...engineTestOption) *engineTestConfig {
 	config := &engineTestConfig{}
 	for _, opt := range opts {

chasm/lib/scheduler/internal/overlap.go

@@ -0,0 +1,24 @@
+package internal
+
+import enumspb "go.temporal.io/api/enums/v1"
+
+// ResolveOverlapPolicy applies a per-action override, then the schedule policy,
+// then the API default.
+func ResolveOverlapPolicy(
+	overlapPolicy enumspb.ScheduleOverlapPolicy,
+	schedulePolicy enumspb.ScheduleOverlapPolicy,
+) enumspb.ScheduleOverlapPolicy {
+	if overlapPolicy != enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
+		return overlapPolicy
+	}
+	if schedulePolicy != enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
+		return schedulePolicy
+	}
+	return enumspb.SCHEDULE_OVERLAP_POLICY_SKIP
+}
+
+// TracksCompletionResult reports whether an action participates in scheduler-wide
+// completion state and overlap resolution after it starts.
+func TracksCompletionResult(overlapPolicy enumspb.ScheduleOverlapPolicy) bool {
+	return overlapPolicy != enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL
+}

chasm/lib/scheduler/internal/overlap_test.go

@@ -0,0 +1,44 @@
+package internal
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
+)
+
+func TestTracksCompletionResult(t *testing.T) {
+	require.False(t, TracksCompletionResult(enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL))
+	require.True(t, TracksCompletionResult(enumspb.SCHEDULE_OVERLAP_POLICY_SKIP))
+}
+
+func TestResolveOverlapPolicy(t *testing.T) {
+	tests := []struct {
+		name           string
+		overlapPolicy  enumspb.ScheduleOverlapPolicy
+		schedulePolicy enumspb.ScheduleOverlapPolicy
+		want           enumspb.ScheduleOverlapPolicy
+	}{
+		{
+			name:           "action override",
+			overlapPolicy:  enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+			schedulePolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
+			want:           enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+		},
+		{
+			name:           "schedule policy",
+			schedulePolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
+			want:           enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
+		},
+		{
+			name: "default",
+			want: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, ResolveOverlapPolicy(tt.overlapPolicy, tt.schedulePolicy))
+		})
+	}
+}

chasm/lib/scheduler/invoker.go

@@ -11,6 +11,8 @@ import (
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/internal"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/util"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -154,10 +156,13 @@ func (e *executeResult) Append(o executeResult) executeResult {
 
 // recordExecuteResult updates the Invoker's internal state with the results of a
 // completed InvokerExecuteTask. It returns the number of *new* actions recorded
-// (starts that transitioned from "no RunId" to "has RunId" in this call) and
-// the number of completed results that were dropped because they were previously
-// recorded.
-func (i *Invoker) recordExecuteResult(ctx chasm.MutableContext, result *executeResult) (newlyStarted, droppedDuplicates int) {
+// (starts that transitioned from "no RunId" to "has RunId" in this call), the
+// number of completed results that were dropped because they were previously
+// recorded, and starts that do not remain active while awaiting completion.
+func (i *Invoker) recordExecuteResult(
+	ctx chasm.MutableContext,
+	result *executeResult,
+) (newlyStarted, droppedDuplicates int, startOnlyActions []*schedulespb.BufferedStart) {
 	completed := make(map[string]*schedulespb.BufferedStart) // request ID -> BufferedStart with RunId/StartTime
 	failed := make(map[string]bool)                          // request ID -> is present
 	retryable := make(map[string]*schedulespb.BufferedStart) // request ID -> *BufferedStart
@@ -183,6 +188,7 @@ func (i *Invoker) recordExecuteResult(ctx chasm.MutableContext, result *executeR
 	// Remove failed (non-retryable) starts from the buffer.
 	removedStarts := 0
 	retriedStarts := 0
+	startedUntracked := make(map[string]struct{})
 	i.BufferedStarts = slices.DeleteFunc(i.GetBufferedStarts(), func(start *schedulespb.BufferedStart) bool {
 		failed := failed[start.RequestId]
 		if failed {
@@ -214,10 +220,16 @@ func (i *Invoker) recordExecuteResult(ctx chasm.MutableContext, result *executeR
 			continue
 		}
 		if completedStart, ok := completed[start.RequestId]; ok {
+			newlyStarted++
+			if !internal.TracksCompletionResult(start.GetOverlapPolicy()) {
+				startOnlyActions = append(startOnlyActions, completedStart)
+				startedUntracked[start.RequestId] = struct{}{}
+				removedStarts++
+				continue
+			}
 			start.RunId = completedStart.GetRunId()
 			start.StartTime = completedStart.GetStartTime()
 			start.HasCallback = true
-			newlyStarted++
 		}
 		if retry, ok := retryable[start.RequestId]; ok {
 			start.Attempt++
@@ -225,7 +237,10 @@ func (i *Invoker) recordExecuteResult(ctx chasm.MutableContext, result *executeR
 			retriedStarts++
 		}
 	}
-
+	i.BufferedStarts = slices.DeleteFunc(i.GetBufferedStarts(), func(start *schedulespb.BufferedStart) bool {
+		_, remove := startedUntracked[start.GetRequestId()]
+		return remove
+	})
 	i.getOrCreateEventLog(ctx).LogEvent(ctx,
 		fmt.Sprintf("recordExecuteResult kicked off %d starts, removed %d starts, retried %d starts",
 			newlyStarted,
@@ -233,7 +248,7 @@ func (i *Invoker) recordExecuteResult(ctx chasm.MutableContext, result *executeR
 			retriedStarts))
 
 	i.addTasks(ctx)
-	return newlyStarted, droppedDuplicates
+	return newlyStarted, droppedDuplicates, startOnlyActions
 }
 
 // runningWorkflowID returns the workflow ID associated with the given
@@ -260,13 +275,24 @@ func (i *Invoker) recordCompletedAction(
 	i.getOrCreateEventLog(ctx).LogEvent(ctx, fmt.Sprintf("recording completed action: %s", requestID))
 
 	// Find the BufferedStart and mark it as completed.
+	completedUntracked := ""
 	for _, start := range i.BufferedStarts {
 		if start.GetRequestId() == requestID {
 			scheduleTime = start.DesiredTime.AsTime()
-			start.Completed = completed
+			if !internal.TracksCompletionResult(start.GetOverlapPolicy()) {
+				i.Scheduler.Get(ctx).recordRecentAction(start, completed.GetStatus())
+				completedUntracked = requestID
+			} else {
+				start.Completed = completed
+			}
 			break
 		}
 	}
+	if completedUntracked != "" {
+		i.BufferedStarts = slices.DeleteFunc(i.BufferedStarts, func(start *schedulespb.BufferedStart) bool {
+			return start.GetRequestId() == completedUntracked
+		})
+	}
 
 	// Re-enable deferred starts (Attempt == -1) so they can be re-processed by
 	// ProcessBuffer now that a workflow has completed. This allows the overlap
@@ -380,7 +406,8 @@ func (i *Invoker) getEligibleBufferedStarts() []*schedulespb.BufferedStart {
 func (i *Invoker) runningWorkflowExecutions() []*commonpb.WorkflowExecution {
 	var running []*commonpb.WorkflowExecution
 	for _, start := range i.GetBufferedStarts() {
-		if start.GetRunId() != "" && start.GetCompleted() == nil {
+		if start.GetRunId() != "" && start.GetCompleted() == nil &&
+			internal.TracksCompletionResult(start.GetOverlapPolicy()) {
 			running = append(running, &commonpb.WorkflowExecution{
 				WorkflowId: start.GetWorkflowId(),
 				RunId:      start.GetRunId(),
@@ -390,11 +417,13 @@ func (i *Invoker) runningWorkflowExecutions() []*commonpb.WorkflowExecution {
 	return running
 }
 
-// recentActions returns started/completed actions as ScheduleActionResults.
-// This includes both running workflows (with status RUNNING) and completed
-// workflows (with their final status).
-func (i *Invoker) recentActions() []*schedulepb.ScheduleActionResult {
-	var results []*schedulepb.ScheduleActionResult
+// recentActions combines stored start-only actions with completion-tracked actions
+// represented by BufferedStarts.
+func (i *Invoker) recentActions(storedActions []*schedulepb.ScheduleActionResult) []*schedulepb.ScheduleActionResult {
+	results := make([]*schedulepb.ScheduleActionResult, 0, len(storedActions)+len(i.GetBufferedStarts()))
+	for _, action := range storedActions {
+		results = append(results, common.CloneProto(action))
+	}
 	for _, start := range i.GetBufferedStarts() {
 		// Only include workflows that have been started (have a RunId).
 		if start.GetRunId() == "" {
@@ -414,7 +443,28 @@ func (i *Invoker) recentActions() []*schedulepb.ScheduleActionResult {
 			StartWorkflowStatus: status,
 		})
 	}
-	return results
+	slices.SortFunc(results, func(a, b *schedulepb.ScheduleActionResult) int {
+		return a.GetActualTime().AsTime().Compare(b.GetActualTime().AsTime())
+	})
+	return util.SliceTail(results, recentActionCount)
+}
+
+// bufferedStartsCount returns the actions whose successful StartWorkflowExecution
+// result has not yet been recorded. BufferedStarts also retains running and completed
+// actions for lifecycle tracking and history, so its length is not the API buffer size.
+// This preserves V1's distinction: V1 removes selected starts from BufferedStarts
+// before recording them as running or recent, while CHASM uses a recorded RunId as
+// the durable boundary between those states.
+// Count starts without a RunId directly because recent actions include start-only
+// ALLOW_ALL records stored outside BufferedStarts and are capped independently.
+func (i *Invoker) bufferedStartsCount() int {
+	count := 0
+	for _, start := range i.GetBufferedStarts() {
+		if start.GetRunId() == "" {
+			count++
+		}
+	}
+	return count
 }
 
 // applyCompletedRetention removes the oldest completed BufferedStarts beyond

chasm/lib/scheduler/invoker_execute_task_test.go

@@ -10,16 +10,19 @@ import (
 	commonpb "go.temporal.io/api/common/v1"
 	deploymentpb "go.temporal.io/api/deployment/v1"
 	enumspb "go.temporal.io/api/enums/v1"
+	failurepb "go.temporal.io/api/failure/v1"
 	"go.temporal.io/api/serviceerror"
 	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/api/historyservicemock/v1"
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/chasm/chasmtest"
 	"go.temporal.io/server/chasm/lib/scheduler"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1"
+	"go.temporal.io/server/common/testing/testlogger"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
@@ -150,6 +153,10 @@ func executeTaskOnce(t *testing.T, env *invokerExecuteTestEnv, ctx chasm.Mutable
 // Execute success case.
 func TestExecuteTask_Basic(t *testing.T) {
 	env := newInvokerExecuteTestEnv(t)
+	env.Scheduler.LastCompletionResult = chasm.NewDataField(env.MutableContext(), &schedulerpb.LastCompletionResult{
+		Success: &commonpb.Payload{Data: []byte("previous-result")},
+		Failure: &failurepb.Failure{Message: "previous-failure"},
+	})
 	startTime := timestamppb.New(env.TimeSource.Now())
 	bufferedStarts := []*schedulespb.BufferedStart{
 		{
@@ -175,21 +182,153 @@ func TestExecuteTask_Basic(t *testing.T) {
 	// Expect both buffered starts to result in workflow executions.
 	env.mockFrontendClient.EXPECT().
 		StartWorkflowExecution(gomock.Any(), gomock.Any()).
-		Times(2).
-		Return(&workflowservice.StartWorkflowExecutionResponse{
-			RunId: "run-id",
-		}, nil)
+		DoAndReturn(func(_ context.Context, req *workflowservice.StartWorkflowExecutionRequest, _ ...grpc.CallOption) (*workflowservice.StartWorkflowExecutionResponse, error) {
+			require.Len(t, req.GetCompletionCallbacks(), 1)
+			require.Empty(t, req.GetLastCompletionResult().GetPayloads())
+			require.Nil(t, req.GetContinuedFailure())
+			return &workflowservice.StartWorkflowExecutionResponse{RunId: "run-id"}, nil
+		}).
+		Times(2)
 
-	// After execution, both BufferedStarts are kept (with RunId set).
-	// They become "running" workflows.
 	runExecuteTestCase(t, env, &executeTestCase{
 		InitialBufferedStarts:    bufferedStarts,
-		ExpectedBufferedStarts:   2, // kept after starting
-		ExpectedRunningWorkflows: 2,
+		ExpectedBufferedStarts:   0,
+		ExpectedRunningWorkflows: 0,
 		ExpectedActionCount:      2,
+		ValidateInvoker: func(t *testing.T, _ *scheduler.Invoker, env *invokerExecuteTestEnv) {
+			listInfo := env.Scheduler.ListInfo(env.ReadContext())
+			require.Len(t, listInfo.GetRecentActions(), 2)
+			for _, action := range listInfo.GetRecentActions() {
+				require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, action.GetStartWorkflowStatus())
+			}
+		},
 	})
 }
 
+func TestExecuteTask_AllowAllRemovedAcrossEngineTransaction(t *testing.T) {
+	logger := testlogger.NewTestLogger(t, testlogger.FailOnExpectedErrorOnly)
+	engine, engineCtx := newTestEngineContext(t, logger)
+	result, err := chasm.StartExecution(
+		engineCtx,
+		chasm.ExecutionKey{NamespaceID: namespaceID, BusinessID: scheduleID},
+		scheduler.CreateScheduler,
+		&schedulerpb.CreateScheduleRequest{
+			NamespaceId: namespaceID,
+			FrontendRequest: &workflowservice.CreateScheduleRequest{
+				Namespace:  namespace,
+				ScheduleId: scheduleID,
+				Schedule:   defaultSchedule(),
+			},
+		},
+	)
+	require.NoError(t, err)
+	rootRef := chasm.NewComponentRef[*scheduler.Scheduler](result.ExecutionKey)
+
+	startTime := timestamppb.Now()
+	var invoker *scheduler.Invoker
+	_, _, err = chasm.UpdateComponent(
+		engineCtx,
+		rootRef,
+		func(sched *scheduler.Scheduler, ctx chasm.MutableContext, _ struct{}) (struct{}, error) {
+			invoker = sched.Invoker.Get(ctx)
+			invoker.LastProcessedTime = startTime
+			invoker.BufferedStarts = []*schedulespb.BufferedStart{{
+				NominalTime: startTime, ActualTime: startTime, DesiredTime: startTime,
+				RequestId: "allow-all-request", WorkflowId: "allow-all-workflow",
+				OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, Attempt: 1,
+			}}
+			return struct{}{}, nil
+		},
+		struct{}{},
+	)
+	require.NoError(t, err)
+
+	ctrl := gomock.NewController(t)
+	frontendClient := workflowservicemock.NewMockWorkflowServiceClient(ctrl)
+	frontendClient.EXPECT().StartWorkflowExecution(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, req *workflowservice.StartWorkflowExecutionRequest, _ ...grpc.CallOption) (*workflowservice.StartWorkflowExecutionResponse, error) {
+			require.Len(t, req.GetCompletionCallbacks(), 1)
+			return &workflowservice.StartWorkflowExecutionResponse{RunId: "allow-all-run"}, nil
+		})
+	handler := scheduler.NewInvokerExecuteTaskHandler(scheduler.InvokerTaskHandlerOptions{
+		Config:         defaultConfig(),
+		MetricsHandler: metrics.NoopMetricsHandler,
+		BaseLogger:     logger,
+		FrontendClient: frontendClient,
+	})
+
+	dropped, err := chasmtest.ExecuteSideEffectTask(
+		context.Background(),
+		engine,
+		invoker,
+		handler,
+		chasm.TaskAttributes{},
+		&schedulerpb.InvokerExecuteTask{},
+	)
+	require.NoError(t, err)
+	require.False(t, dropped)
+
+	_, err = chasm.ReadComponent(
+		engineCtx,
+		rootRef,
+		func(sched *scheduler.Scheduler, ctx chasm.Context, _ struct{}) (struct{}, error) {
+			persistedInvoker := sched.Invoker.Get(ctx)
+			require.Empty(t, persistedInvoker.GetBufferedStarts())
+			require.Len(t, sched.Info.GetRecentActions(), 1)
+			require.Equal(t, "allow-all-run", sched.Info.GetRecentActions()[0].GetStartWorkflowResult().GetRunId())
+			describe, describeErr := sched.Describe(ctx, &schedulerpb.DescribeScheduleRequest{}, newLegacySpecBuilder(0, 0))
+			require.NoError(t, describeErr)
+			require.Empty(t, describe.GetFrontendResponse().GetInfo().GetRunningWorkflows())
+			require.Len(t, describe.GetFrontendResponse().GetInfo().GetRecentActions(), 1)
+			require.Zero(t, describe.GetFrontendResponse().GetInfo().GetBufferSize())
+			return struct{}{}, nil
+		},
+		struct{}{},
+	)
+	require.NoError(t, err)
+}
+
+func TestRecordExecuteResult_AllowAllRecentActionsBounded(t *testing.T) {
+	env := newTestEnv(t)
+	ctx := env.MutableContext()
+	invoker := env.Scheduler.Invoker.Get(ctx)
+	base := env.TimeSource.Now().Add(-time.Hour)
+	var completed []*schedulespb.BufferedStart
+	for idx := range scheduler.RecentActionCount + 2 {
+		startTime := timestamppb.New(base.Add(time.Duration(idx) * time.Minute))
+		start := &schedulespb.BufferedStart{
+			NominalTime: startTime, ActualTime: startTime, DesiredTime: startTime,
+			RequestId: fmt.Sprintf("req-%d", idx), WorkflowId: fmt.Sprintf("wf-%d", idx),
+			OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, Attempt: 1,
+		}
+		invoker.BufferedStarts = append(invoker.BufferedStarts, start)
+		result := proto.Clone(start).(*schedulespb.BufferedStart)
+		result.RunId = fmt.Sprintf("run-%d", idx)
+		result.StartTime = startTime
+		completed = append(completed, result)
+	}
+
+	newlyStarted, droppedDuplicates, startOnlyActions := invoker.RecordExecuteResult(ctx, completed, nil)
+	env.Scheduler.RecordStartOnlyActions(ctx, startOnlyActions)
+	require.Equal(t, scheduler.RecentActionCount+2, newlyStarted)
+	require.Zero(t, droppedDuplicates)
+	require.Empty(t, invoker.GetBufferedStarts())
+	require.Len(t, env.Scheduler.Info.GetRecentActions(), scheduler.RecentActionCount)
+	require.Equal(t, "run-2", env.Scheduler.Info.GetRecentActions()[0].GetStartWorkflowResult().GetRunId())
+	require.Equal(t, "run-11", env.Scheduler.Info.GetRecentActions()[scheduler.RecentActionCount-1].GetStartWorkflowResult().GetRunId())
+	recentRunIDs := make(map[string]struct{}, scheduler.RecentActionCount)
+	for _, action := range env.Scheduler.Info.GetRecentActions() {
+		recentRunIDs[action.GetStartWorkflowResult().GetRunId()] = struct{}{}
+	}
+	require.NotContains(t, recentRunIDs, "run-0")
+	bufferedRequestIDs := make(map[string]struct{}, len(invoker.GetBufferedStarts()))
+	for _, start := range invoker.GetBufferedStarts() {
+		bufferedRequestIDs[start.GetRequestId()] = struct{}{}
+	}
+	require.NotContains(t, bufferedRequestIDs, "req-0")
+	require.Len(t, env.Scheduler.ListInfo(ctx).GetRecentActions(), 5)
+}
+
 func TestExecuteTask_ForwardsVersioningOverride(t *testing.T) {
 	tests := map[string]struct {
 		override *workflowpb.VersioningOverride
@@ -259,8 +398,8 @@ func TestExecuteTask_ForwardsVersioningOverride(t *testing.T) {
 					OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
 					Attempt:       1,
 				}},
-				ExpectedBufferedStarts:   1,
-				ExpectedRunningWorkflows: 1,
+				ExpectedBufferedStarts:   0,
+				ExpectedRunningWorkflows: 0,
 				ExpectedActionCount:      1,
 			})
 		})
@@ -311,6 +450,33 @@ func TestExecuteTask_DistinctRequestsCanReuseCompletedWorkflowID(t *testing.T) {
 	})
 }
 
+func TestExecuteTask_UsesBufferedOverlapPolicy(t *testing.T) {
+	env := newInvokerExecuteTestEnv(t)
+	env.Scheduler.Schedule.Policies.OverlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL
+	startTime := timestamppb.New(env.TimeSource.Now())
+
+	env.mockFrontendClient.EXPECT().
+		StartWorkflowExecution(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, req *workflowservice.StartWorkflowExecutionRequest, _ ...grpc.CallOption) (*workflowservice.StartWorkflowExecutionResponse, error) {
+			require.Len(t, req.GetCompletionCallbacks(), 1)
+			return &workflowservice.StartWorkflowExecutionResponse{RunId: "run-id"}, nil
+		})
+
+	runExecuteTestCase(t, env, &executeTestCase{
+		InitialBufferedStarts: []*schedulespb.BufferedStart{{
+			NominalTime: startTime, ActualTime: startTime, DesiredTime: startTime,
+			RequestId: "request-id", WorkflowId: "workflow-id", Attempt: 1,
+			OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
+		}},
+		ExpectedBufferedStarts:   1,
+		ExpectedRunningWorkflows: 1,
+		ExpectedActionCount:      1,
+		ValidateInvoker: func(t *testing.T, invoker *scheduler.Invoker, _ *invokerExecuteTestEnv) {
+			require.Equal(t, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, invoker.BufferedStarts[0].GetOverlapPolicy())
+		},
+	})
+}
+
 // Execute is scheduled with an empty buffer.
 func TestExecuteTask_Empty(t *testing.T) {
 	env := newInvokerExecuteTestEnv(t)
@@ -361,11 +527,11 @@ func TestExecuteTask_RetryableFailure(t *testing.T) {
 
 	// After execution:
 	// - Failed start stays in buffer with backoff (pending)
-	// - Successful start stays in buffer with RunId set (running)
+	// - Successful ALLOW_ALL start moves to recent-action history
 	runExecuteTestCase(t, env, &executeTestCase{
 		InitialBufferedStarts:    bufferedStarts,
-		ExpectedBufferedStarts:   2, // both kept: 1 failed (backoff) + 1 running
-		ExpectedRunningWorkflows: 1,
+		ExpectedBufferedStarts:   1,
+		ExpectedRunningWorkflows: 0,
 		ExpectedActionCount:      1,
 		ValidateInvoker: func(t *testing.T, invoker *scheduler.Invoker, env *invokerExecuteTestEnv) {
 			// Find the failed start (no RunId, has backoff).
@@ -461,8 +627,8 @@ func TestExecuteTask_SucceedsOnFinalAttempt(t *testing.T) {
 			OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
 			Attempt:       scheduler.InvokerMaxStartAttempts,
 		}},
-		ExpectedBufferedStarts:   1,
-		ExpectedRunningWorkflows: 1,
+		ExpectedBufferedStarts:   0,
+		ExpectedRunningWorkflows: 0,
 		ExpectedActionCount:      1,
 	})
 }
@@ -659,11 +825,11 @@ func TestExecuteTask_ExceedsMaxActionsPerExecution(t *testing.T) {
 			RunId: "run-id",
 		}, nil)
 
-	// All BufferedStarts are kept: maxStarts get RunId set (running), the rest stay pending.
+	// Started ALLOW_ALL actions move to recent history; only the unexecuted half remains buffered.
 	runExecuteTestCase(t, env, &executeTestCase{
 		InitialBufferedStarts:    bufferedStarts,
-		ExpectedBufferedStarts:   maxStarts * 2, // all kept: started + pending
-		ExpectedRunningWorkflows: maxStarts,     // only started ones
+		ExpectedBufferedStarts:   maxStarts,
+		ExpectedRunningWorkflows: 0,
 		ExpectedActionCount:      int64(maxStarts),
 	})
 }
@@ -700,9 +866,10 @@ func TestExecuteTask_RecordResultIdempotentOnRace(t *testing.T) {
 		StartTime: loserStartTime,
 	}}
 
-	newlyStarted, droppedDuplicates := invoker.RecordExecuteResult(ctx, loser, nil)
+	newlyStarted, droppedDuplicates, startOnlyActions := invoker.RecordExecuteResult(ctx, loser, nil)
 	require.Equal(t, 0, newlyStarted, "duplicate RunId-set start must not be counted")
 	require.Equal(t, 1, droppedDuplicates, "the dropped completion must be reported for observability")
+	require.Empty(t, startOnlyActions)
 	live := invoker.BufferedStarts[0]
 	require.Equal(t, winning, live.RunId, "live RunId must not be stomped")
 	require.Equal(t, startTime.AsTime(), live.StartTime.AsTime(), "live StartTime must not be stomped")
@@ -723,9 +890,10 @@ func TestExecuteTask_RecordResultIdempotentOnRace(t *testing.T) {
 		RunId:     "first-run",
 		StartTime: startTime,
 	}}
-	newlyStarted, droppedDuplicates = invoker.RecordExecuteResult(ctx, first, nil)
+	newlyStarted, droppedDuplicates, startOnlyActions = invoker.RecordExecuteResult(ctx, first, nil)
 	require.Equal(t, 1, newlyStarted, "first-time RunId assignment must be counted")
 	require.Equal(t, 0, droppedDuplicates, "no duplicate was dropped")
+	require.Empty(t, startOnlyActions)
 	freshlyStarted := invoker.BufferedStarts[1]
 	require.Equal(t, "first-run", freshlyStarted.RunId)
 	require.Equal(t, startTime.AsTime(), freshlyStarted.StartTime.AsTime())
@@ -763,9 +931,10 @@ func TestExecuteTask_RecordResultIdempotentOnRetryableRace(t *testing.T) {
 		BackoffTime: loserBackoff,
 	}}
 
-	newlyStarted, droppedDuplicates := invoker.RecordExecuteResult(ctx, nil, retryable)
+	newlyStarted, droppedDuplicates, startOnlyActions := invoker.RecordExecuteResult(ctx, nil, retryable)
 	require.Equal(t, 0, newlyStarted)
 	require.Equal(t, 0, droppedDuplicates, "retryable drops aren't counted as duplicates - only completed-side drops are")
+	require.Empty(t, startOnlyActions)
 	live := invoker.BufferedStarts[0]
 	require.Equal(t, int64(1), live.Attempt, "Attempt must not be incremented on a started entry")
 	require.Nil(t, live.BackoffTime, "BackoffTime must not be set on a started entry")

chasm/lib/scheduler/invoker_tasks.go

@@ -16,6 +16,7 @@ import (
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
@@ -243,7 +244,8 @@ func (h *InvokerExecuteTaskHandler) Execute(
 			s := i.Scheduler.Get(ctx)
 			// Use newlyStarted (not len(result.CompletedStarts)) so a concurrent
 			// ExecuteTask's duplicate StartWorkflow can't inflate ActionCount.
-			newlyStarted, droppedDuplicates := i.recordExecuteResult(ctx, &result)
+			newlyStarted, droppedDuplicates, startOnlyActions := i.recordExecuteResult(ctx, &result)
+			s.recordStartOnlyActions(ctx, startOnlyActions)
 			s.recordActionResult(&schedulerActionResult{actionCount: int64(newlyStarted)})
 			if droppedDuplicates > 0 {
 				h.recordDuplicateExecuteDrops(s, droppedDuplicates)
@@ -378,8 +380,8 @@ func (h *InvokerExecuteTaskHandler) startWorkflows(
 			break
 		}
 
-		// Clone start before concurrent access. The clone will have RunId/StartTime
-		// set by startWorkflow, then copied back to the original in recordExecuteResult.
+		// Clone start before concurrent access. Buffered starts carry the policy
+		// resolved when they entered CHASM, including through migration.
 		start = common.CloneProto(start)
 
 		// Run all starts concurrently.
@@ -660,14 +662,18 @@ func (h *InvokerExecuteTaskHandler) startWorkflow(
 		reusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE
 	}
 
+	tracksCompletionResult := internal.TracksCompletionResult(start.GetOverlapPolicy())
 	var lcr []*commonpb.Payload
-	if lastCompletionState.Success != nil {
+	continuedFailure := lastCompletionState.Failure
+	if tracksCompletionResult && lastCompletionState.Success != nil {
 		lcr = append(lcr, lastCompletionState.Success)
 	}
-	// Build the completion callback with this start's request ID packed into its token, so the
-	// completion is matched by a request ID that rides in the callback header and survives
-	// continue-as-new, rather than the started workflow's callback state which is re-stamped on each
-	// new run.
+	if !tracksCompletionResult {
+		continuedFailure = nil
+	}
+	// Always attach callbacks so StartWorkflowExecution requests remain compatible
+	// across mixed server versions. ALLOW_ALL callbacks become harmless late deliveries
+	// after their starts move to start-only history.
 	callback, err := chasm.GenerateNexusCallback(schedulerRef, start.RequestId, h.config.EncodeInternalTokenWithEnvelope(scheduler.Namespace))
 	if err != nil {
 		return err
@@ -691,7 +697,7 @@ func (h *InvokerExecuteTaskHandler) startWorkflow(
 		WorkflowTaskTimeout:      requestSpec.WorkflowTaskTimeout,
 		WorkflowType:             requestSpec.WorkflowType,
 		Priority:                 requestSpec.Priority,
-		ContinuedFailure:         lastCompletionState.Failure,
+		ContinuedFailure:         continuedFailure,
 		LastCompletionResult: &commonpb.Payloads{
 			Payloads: lcr,
 		},

chasm/lib/scheduler/migration/identity_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -24,6 +25,7 @@ func TestSameTimePendingStartsReceiveUniqueIdentities(t *testing.T) {
 		"schedule-id",
 		1,
 		"workflow-id",
+		enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
 	)
 	require.Len(t, converted, 2)
 	require.NotEqual(t, converted[0].GetRequestId(), converted[1].GetRequestId(),
@@ -51,6 +53,7 @@ func TestMigratedStartsPreserveExistingIdentities(t *testing.T) {
 		"schedule-id",
 		1,
 		"workflow-id",
+		enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
 	)
 	require.Len(t, converted, 2)
 	for i, start := range converted {
@@ -60,3 +63,49 @@ func TestMigratedStartsPreserveExistingIdentities(t *testing.T) {
 			"identities carried over from V1 must not be suffixed")
 	}
 }
+
+func TestMigratedStartsResolveOverlapPolicy(t *testing.T) {
+	when := timestamppb.New(time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC))
+	tests := []struct {
+		name           string
+		startPolicy    enumspb.ScheduleOverlapPolicy
+		schedulePolicy enumspb.ScheduleOverlapPolicy
+		want           enumspb.ScheduleOverlapPolicy
+	}{
+		{
+			name:           "explicit start policy",
+			startPolicy:    enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
+			schedulePolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+			want:           enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
+		},
+		{
+			name:           "inherited schedule policy",
+			schedulePolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+			want:           enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+		},
+		{
+			name: "default policy",
+			want: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			converted := convertBufferedStartsLegacyToCHASM(
+				[]*schedulespb.BufferedStart{{
+					NominalTime:   when,
+					ActualTime:    when,
+					OverlapPolicy: tt.startPolicy,
+				}},
+				"namespace-id",
+				"schedule-id",
+				1,
+				"workflow-id",
+				tt.schedulePolicy,
+			)
+
+			require.Len(t, converted, 1)
+			require.Equal(t, tt.want, converted[0].GetOverlapPolicy())
+		})
+	}
+}

chasm/lib/scheduler/migration/migration.go

@@ -3,6 +3,7 @@ package migration
 import (
 	"fmt"
 	"maps"
+	"slices"
 	"strconv"
 	"time"
 
@@ -15,9 +16,12 @@ import (
 	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/searchattribute/sadefs"
+	"go.temporal.io/server/common/util"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
+const legacyRecentActionCount = 10
+
 // LegacyToCreateFromMigrationStateRequest converts legacy (workflow-backed) scheduler
 // state to a CreateFromMigrationStateRequest proto. This is the primary V1-to-V2
 // migration function.
@@ -37,8 +41,8 @@ import (
 //   - High water mark (becomes Generator.LastProcessedTime)
 //   - Search attributes and memo
 //
-// Note: In V2, RunningWorkflows and RecentActions are computed on-demand from
-// BufferedStarts by the Invoker, rather than being stored separately in ScheduleInfo.
+// Note: In V2, completion-tracked RunningWorkflows and RecentActions are computed
+// on-demand from BufferedStarts. Start-only recent actions remain in ScheduleInfo.
 func LegacyToCreateFromMigrationStateRequest(
 	schedule *schedulepb.Schedule,
 	info *schedulepb.ScheduleInfo,
@@ -47,7 +51,7 @@ func LegacyToCreateFromMigrationStateRequest(
 	memo *commonpb.Memo,
 	migrationTime time.Time,
 ) *schedulerpb.CreateFromMigrationStateRequest {
-	// V2 computes RunningWorkflows/RecentActions on-demand from BufferedStarts
+	// Imported recent actions are represented by BufferedStarts in V2.
 	infoClone := common.CloneProto(info)
 	infoClone.RunningWorkflows = nil
 	infoClone.RecentActions = nil
@@ -73,6 +77,7 @@ func LegacyToCreateFromMigrationStateRequest(
 		state.ScheduleId,
 		state.ConflictToken,
 		getWorkflowID(schedule),
+		schedule.GetPolicies().GetOverlapPolicy(),
 	)
 
 	runningBufferedStarts := convertRunningWorkflowsToBufferedStarts(
@@ -164,6 +169,17 @@ func CHASMToLegacyStartScheduleArgs(
 		invokerBuffered = invoker.GetBufferedStarts()
 	}
 	bufferedStarts, running, recent := splitBufferedStartsForLegacy(invokerBuffered)
+	if len(info.GetRecentActions()) > 0 {
+		storedRecent := make([]*schedulepb.ScheduleActionResult, 0, len(info.GetRecentActions()))
+		for _, action := range info.GetRecentActions() {
+			storedRecent = append(storedRecent, common.CloneProto(action))
+		}
+		recent = append(storedRecent, recent...)
+		slices.SortFunc(recent, func(a, b *schedulepb.ScheduleActionResult) int {
+			return a.GetActualTime().AsTime().Compare(b.GetActualTime().AsTime())
+		})
+		recent = util.SliceTail(recent, legacyRecentActionCount)
+	}
 	ongoingBackfills, triggerStarts := convertBackfillersCHASMToLegacy(backfillers, migrationTime)
 	bufferedStarts = append(bufferedStarts, triggerStarts...)
 
@@ -208,6 +224,7 @@ func convertBufferedStartsLegacyToCHASM(
 	namespaceID, scheduleID string,
 	conflictToken int64,
 	baseWorkflowID string,
+	scheduleOverlapPolicy enumspb.ScheduleOverlapPolicy,
 ) []*schedulespb.BufferedStart {
 	if len(v1Starts) == 0 {
 		return nil
@@ -250,6 +267,10 @@ func convertBufferedStartsLegacyToCHASM(
 
 		v2Start.Attempt = 0
 		v2Start.BackoffTime = nil
+		v2Start.OverlapPolicy = schedulerinternal.ResolveOverlapPolicy(
+			v2Start.GetOverlapPolicy(),
+			scheduleOverlapPolicy,
+		)
 
 		v2Starts[i] = v2Start
 	}
@@ -466,7 +487,7 @@ func splitBufferedStartsForLegacy(
 		// to later non-ALLOW_ALL starts. They still appear in RecentActions above,
 		// matching V1.
 		if start.GetCompleted() == nil &&
-			start.GetOverlapPolicy() != enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL {
+			schedulerinternal.TracksCompletionResult(start.GetOverlapPolicy()) {
 			running = append(running, &commonpb.WorkflowExecution{
 				WorkflowId: start.GetWorkflowId(),
 				RunId:      start.GetRunId(),

chasm/lib/scheduler/migration/migration_test.go

@@ -232,6 +232,15 @@ func TestCHASMToLegacyStartScheduleArgs(t *testing.T) {
 		Info:          &schedulepb.ScheduleInfo{ActionCount: 12},
 	}
 	generator := &schedulerpb.GeneratorState{LastProcessedTime: timestamppb.New(now.Add(-time.Minute))}
+	scheduler.Info.RecentActions = []*schedulepb.ScheduleActionResult{{
+		ScheduleTime: timestamppb.New(now.Add(-4 * time.Minute)),
+		ActualTime:   timestamppb.New(now.Add(-4 * time.Minute)),
+		StartWorkflowResult: &commonpb.WorkflowExecution{
+			WorkflowId: "wf-start-only",
+			RunId:      "run-start-only",
+		},
+		StartWorkflowStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
+	}}
 	invoker := &schedulerpb.InvokerState{
 		BufferedStarts: []*schedulespb.BufferedStart{
 			{
@@ -299,7 +308,8 @@ func TestCHASMToLegacyStartScheduleArgs(t *testing.T) {
 	require.Equal(t, "wf-running", args.Info.RunningWorkflows[0].WorkflowId)
 	require.Equal(t, "run-running", args.Info.RunningWorkflows[0].RunId)
 
-	require.Len(t, args.Info.RecentActions, 2)
+	require.Len(t, args.Info.RecentActions, 3)
+	require.Equal(t, "run-start-only", args.Info.RecentActions[2].GetStartWorkflowResult().GetRunId())
 	require.Len(t, args.State.BufferedStarts, 2) // pending + trigger
 	require.Len(t, args.State.OngoingBackfills, 1)
 	require.Equal(t, backfillProgress.AsTime(), args.State.OngoingBackfills[0].StartTime.AsTime())

chasm/lib/scheduler/scheduler.go

@@ -18,6 +18,7 @@ import (
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/contextutil"
 	"go.temporal.io/server/common/log/tag"
@@ -67,6 +68,11 @@ var (
 	_ (chasm.VisibilityMemoProvider)             = (*Scheduler)(nil)
 )
 
+const (
+	callbackIgnoredUnrecognizedRequest metrics.ReasonString = "unrecognized_request_id"
+	callbackIgnoredAlreadyCompleted    metrics.ReasonString = "already_completed"
+)
+
 var (
 	executionStatusRunning   = "Running"
 	executionStatusCompleted = "Completed"
@@ -475,18 +481,14 @@ func (s *Scheduler) identity() string {
 }
 
 func (s *Scheduler) overlapPolicy() enumspb.ScheduleOverlapPolicy {
-	policy := s.Schedule.GetPolicies().GetOverlapPolicy()
-	if policy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
-		policy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP
-	}
-	return policy
+	return internal.ResolveOverlapPolicy(
+		enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,
+		s.Schedule.GetPolicies().GetOverlapPolicy(),
+	)
 }
 
 func (s *Scheduler) resolveOverlapPolicy(overlapPolicy enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy {
-	if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
-		overlapPolicy = s.overlapPolicy()
-	}
-	return overlapPolicy
+	return internal.ResolveOverlapPolicy(overlapPolicy, s.overlapPolicy())
 }
 
 // validateCachedState clears cached fields whenever the Scheduler's
@@ -520,7 +522,7 @@ func (s *Scheduler) getLastEventTime(ctx chasm.Context) time.Time {
 	)
 
 	// The recentActions list is unsorted.
-	for _, a := range s.Invoker.Get(ctx).recentActions() {
+	for _, a := range s.recentActions(ctx) {
 		latest = util.MaxTime(latest, a.GetActualTime().AsTime())
 	}
 
@@ -581,13 +583,51 @@ type schedulerActionResult struct {
 }
 
 // recordActionResult updates the Scheduler's customer-facing metrics.
-// RunningWorkflows and RecentActions are computed from BufferedStarts.
+// RunningWorkflows are computed from BufferedStarts. RecentActions also includes
+// start-only records for actions that are removed from active state after starting.
 func (s *Scheduler) recordActionResult(result *schedulerActionResult) {
 	s.Info.ActionCount += result.actionCount
 	s.Info.OverlapSkipped += result.overlapSkipped
 	s.Info.MissedCatchupWindow += result.missedCatchupWindow
 }
 
+func (s *Scheduler) recordRecentAction(
+	start *schedulespb.BufferedStart,
+	status enumspb.WorkflowExecutionStatus,
+) {
+	s.Info.RecentActions = append(s.Info.RecentActions, &schedulepb.ScheduleActionResult{
+		ScheduleTime: start.GetActualTime(),
+		ActualTime:   start.GetStartTime(),
+		StartWorkflowResult: &commonpb.WorkflowExecution{
+			WorkflowId: start.GetWorkflowId(),
+			RunId:      start.GetRunId(),
+		},
+		StartWorkflowStatus: status,
+	})
+	slices.SortFunc(s.Info.RecentActions, func(a, b *schedulepb.ScheduleActionResult) int {
+		return a.GetActualTime().AsTime().Compare(b.GetActualTime().AsTime())
+	})
+	s.Info.RecentActions = util.SliceTail(s.Info.RecentActions, recentActionCount)
+}
+
+func (s *Scheduler) recordStartOnlyActions(
+	ctx chasm.MutableContext,
+	starts []*schedulespb.BufferedStart,
+) {
+	for _, start := range starts {
+		s.recordRecentAction(start, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING)
+	}
+	if len(starts) > 0 {
+		// These starts have no completion callback to rearm an idle task after
+		// their start-only history advances the idle deadline.
+		s.Generator.Get(ctx).Generate(ctx)
+	}
+}
+
+func (s *Scheduler) recentActions(ctx chasm.Context) []*schedulepb.ScheduleActionResult {
+	return s.Invoker.Get(ctx).recentActions(s.Info.GetRecentActions())
+}
+
 var _ chasm.NexusCompletionHandler = &Scheduler{}
 
 func executionStatusFromFailure(failure *failurepb.Failure) enumspb.WorkflowExecutionStatus {
@@ -613,6 +653,18 @@ func countsAsFailureForPause(status enumspb.WorkflowExecutionStatus) bool {
 	}
 }
 
+func (s *Scheduler) recordIgnoredCallback(
+	ctx chasm.MutableContext,
+	metricsHandler metrics.Handler,
+	requestID string,
+	reason metrics.ReasonString,
+	message string,
+) {
+	s.getOrCreateEventLog(ctx).LogEvent(ctx, fmt.Sprintf("%s: %s", message, requestID))
+	ctx.Logger().Warn(message, tag.RequestID(requestID), tag.ScheduleID(s.ScheduleId))
+	metricsHandler.Counter(metrics.ScheduleCallbackIgnored.Name()).Record(1, metrics.ReasonTag(reason))
+}
+
 // HandleNexusCompletion allows Scheduler to record workflow completions from
 // worfklows started by the same scheduler tree's Invoker.
 func (s *Scheduler) HandleNexusCompletion(
@@ -622,19 +674,37 @@ func (s *Scheduler) HandleNexusCompletion(
 	invoker := s.Invoker.Get(ctx)
 	metricsHandler := newTaggedMetricsHandler(ctx.MetricsHandler(), s)
 
-	workflowID := invoker.runningWorkflowID(info.RequestId)
-	if workflowID == "" {
-		// If the request ID was removed, the request must have already been processed;
-		// fast-succeed.
-		msg := "handled Nexus completion with an unrecognized request ID"
-		s.getOrCreateEventLog(ctx).LogEvent(ctx,
-			fmt.Sprintf("%s: %s", msg, info.RequestId))
-		ctx.Logger().Warn(msg,
-			tag.RequestID(info.RequestId),
-			tag.ScheduleID(s.ScheduleId))
-		metricsHandler.Counter(metrics.ScheduleCallbackIgnored.Name()).Record(1)
+	var start *schedulespb.BufferedStart
+	for _, bufferedStart := range invoker.GetBufferedStarts() {
+		if bufferedStart.GetRequestId() == info.RequestId {
+			start = bufferedStart
+			break
+		}
+	}
+	if start == nil {
+		// Missing request IDs are expected for start-only ALLOW_ALL actions because
+		// their callbacks remain attached for rolling-upgrade compatibility.
+		// TODO: Restore warning and event logging once those callbacks can be safely omitted.
+		metricsHandler.Counter(metrics.ScheduleCallbackIgnored.Name()).Record(
+			1,
+			metrics.ReasonTag(callbackIgnoredUnrecognizedRequest),
+		)
 		return nil
 	}
+	if start.GetCompleted() != nil {
+		// Completion callbacks may be validly redelivered, for example after a workflow reset.
+		// Preserve state but keep the duplicate observable through the log and metric.
+		s.recordIgnoredCallback(
+			ctx,
+			metricsHandler,
+			info.RequestId,
+			callbackIgnoredAlreadyCompleted,
+			"handled Nexus completion for an already-completed buffered start",
+		)
+		return nil
+	}
+	workflowID := start.GetWorkflowId()
+	tracksCompletionResult := internal.TracksCompletionResult(start.GetOverlapPolicy())
 
 	// Record how long it took for the callback to arrive after the action completed.
 	// Use ctx.Now instead of time.Since to use a consistent time source across nodes,
@@ -650,23 +720,27 @@ func (s *Scheduler) HandleNexusCompletion(
 	var wfStatus enumspb.WorkflowExecutionStatus
 	switch outcome := info.Outcome.(type) {
 	case *persistencespb.ChasmNexusCompletion_Failure:
-		previousResult := s.LastCompletionResult.Get(ctx) // Most-recent success is kept after failure.
 		wfStatus = executionStatusFromFailure(outcome.Failure)
-		s.LastCompletionResult = chasm.NewDataField(ctx, &schedulerpb.LastCompletionResult{
-			Failure: outcome.Failure,
-			Success: previousResult.Success,
-		})
+		if tracksCompletionResult {
+			previousResult := s.LastCompletionResult.Get(ctx) // Most-recent success is kept after failure.
+			s.LastCompletionResult = chasm.NewDataField(ctx, &schedulerpb.LastCompletionResult{
+				Failure: outcome.Failure,
+				Success: previousResult.Success,
+			})
+		}
 	case *persistencespb.ChasmNexusCompletion_Success:
 		wfStatus = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
-		s.LastCompletionResult = chasm.NewDataField(ctx, &schedulerpb.LastCompletionResult{
-			Success: outcome.Success,
-		})
+		if tracksCompletionResult {
+			s.LastCompletionResult = chasm.NewDataField(ctx, &schedulerpb.LastCompletionResult{
+				Success: outcome.Success,
+			})
+		}
 	default:
 		wfStatus = enumspb.WORKFLOW_EXECUTION_STATUS_FAILED
 	}
 
 	// Handle pause-on-failure.
-	if countsAsFailureForPause(wfStatus) &&
+	if tracksCompletionResult && countsAsFailureForPause(wfStatus) &&
 		s.Schedule.Policies.PauseOnFailure && !s.Schedule.State.Paused {
 		s.Schedule.State.Paused = true
 		s.Schedule.State.Notes = fmt.Sprintf(
@@ -735,11 +809,10 @@ func (s *Scheduler) Describe(
 	invoker := s.Invoker.Get(ctx)
 	info := common.CloneProto(s.Info)
 	info.RunningWorkflows = invoker.runningWorkflowExecutions()
-	info.RecentActions = invoker.recentActions()
+	info.RecentActions = s.recentActions(ctx)
 	info.FutureActionTimes = futureActionTimes
-	// BufferedStarts holds waiting, running, and recently-completed entries; only the
-	// waiting portion (those not yet surfaced via RecentActions) counts as buffered.
-	info.BufferSize = int64(len(invoker.GetBufferedStarts()) - len(info.RecentActions))
+	// Only starts that have not reached StartWorkflowExecution count as buffered.
+	info.BufferSize = int64(invoker.bufferedStartsCount())
 
 	executionInfo := ctx.ExecutionInfo()
 	info.StateSizeBytes = int64(executionInfo.ApproximateStateSize)
@@ -1017,7 +1090,7 @@ func (s *Scheduler) SearchAttributes(ctx chasm.Context) []chasm.SearchAttributeK
 
 		invoker := s.Invoker.Get(ctx)
 		runningWorkflowCount := int64(len(invoker.runningWorkflowExecutions()))
-		bufferedStartsCount := int64(len(invoker.GetBufferedStarts()) - len(invoker.recentActions()))
+		bufferedStartsCount := int64(invoker.bufferedStartsCount())
 
 		// Emitted even when zero so that exact and range queries both work.
 		out = append(out,
@@ -1054,11 +1127,10 @@ func (s *Scheduler) ListInfo(
 	spec.StructuredCalendar = util.SliceHead(spec.StructuredCalendar, listInfoSpecFieldLimit)
 
 	generator := s.Generator.Get(ctx)
-	invoker := s.Invoker.Get(ctx)
 
 	// Hard-cap the memo's recent-action list by length after sorting by actual time
 	// (ascending towards most recent).
-	recentActions := invoker.recentActions()
+	recentActions := s.recentActions(ctx)
 	slices.SortFunc(recentActions, func(a, b *schedulepb.ScheduleActionResult) int {
 		return a.GetActualTime().AsTime().Compare(b.GetActualTime().AsTime())
 	})

chasm/lib/scheduler/scheduler_idle_tasks_test.go

@@ -5,10 +5,13 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
 	schedulepb "go.temporal.io/api/schedule/v1"
+	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/metrics/metricstest"
@@ -99,6 +102,59 @@ func TestIdleTask_ExecuteInitializesEventLogMissingFromOlderTree(t *testing.T) {
 	require.Equal(t, "schedule closed from idle timer", eventLog.Events[0].Message)
 }
 
+func TestIdleTask_AllowAllStartRearmsIdleTimer(t *testing.T) {
+	base := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+	timeSource := clock.NewEventTimeSource()
+	timeSource.Update(base)
+
+	schedule := defaultSchedule()
+	schedule.Spec = &schedulepb.ScheduleSpec{}
+	testEngine := newSchedulerTestEngine(t, schedule, withEngineTimeSource(timeSource))
+
+	_, err := testEngine.engine.FirePureTasks(testEngine.rootRef, base)
+	require.NoError(t, err)
+	oldIdleDeadline := base.Add(scheduler.DefaultTweakables.IdleTime)
+	require.NoError(t, testEngine.readScheduler(func(s *scheduler.Scheduler, _ chasm.Context) error {
+		require.Equal(t, oldIdleDeadline, s.IdleCloseTime.AsTime())
+		return nil
+	}))
+
+	startTime := base.Add(time.Minute)
+	timeSource.Update(startTime)
+	require.NoError(t, testEngine.updateScheduler(func(s *scheduler.Scheduler, ctx chasm.MutableContext) error {
+		invoker := s.Invoker.Get(ctx)
+		invoker.BufferedStarts = append(invoker.BufferedStarts, &schedulespb.BufferedStart{
+			RequestId:     "allow-all-request",
+			WorkflowId:    "allow-all-workflow",
+			NominalTime:   timestamppb.New(startTime),
+			ActualTime:    timestamppb.New(startTime),
+			DesiredTime:   timestamppb.New(startTime),
+			OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+			Attempt:       1,
+		})
+		_, _, startOnlyActions := invoker.RecordExecuteResult(ctx, []*schedulespb.BufferedStart{{
+			RequestId:     "allow-all-request",
+			RunId:         "allow-all-run",
+			StartTime:     timestamppb.New(startTime),
+			OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+		}}, nil)
+		s.RecordStartOnlyActions(ctx, startOnlyActions)
+		return nil
+	}))
+
+	_, err = testEngine.engine.FirePureTasks(testEngine.rootRef, oldIdleDeadline)
+	require.NoError(t, err)
+
+	newIdleDeadline := startTime.Add(scheduler.DefaultTweakables.IdleTime)
+	timeSource.Update(newIdleDeadline)
+	_, err = testEngine.engine.FirePureTasks(testEngine.rootRef, newIdleDeadline)
+	require.NoError(t, err)
+	require.NoError(t, testEngine.readScheduler(func(s *scheduler.Scheduler, _ chasm.Context) error {
+		require.True(t, s.Closed)
+		return nil
+	}))
+}
+
 func TestIdleTask_Validate_SchedulerNotIdle(t *testing.T) {
 	env := newTestEnv(t)
 	now := env.TimeSource.Now()

chasm/lib/scheduler/scheduler_nexus_completion_test.go

@@ -9,12 +9,17 @@ import (
 	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/workflowservice/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"
 	schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/migration"
+	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/metrics/metricstest"
+	"go.temporal.io/server/common/testing/testlogger"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -128,6 +133,167 @@ func TestHandleNexusCompletion_Success(t *testing.T) {
 	executeNexusCompletion(t, tc)
 }
 
+func TestHandleNexusCompletion_ExistingAllowAllDoesNotUpdateCompletionState(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		manual bool
+	}{
+		{name: "generator", manual: false},
+		{name: "backfiller", manual: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			sched, ctx, node := setupSchedulerForTest(t)
+			initial := &schedulerpb.LastCompletionResult{Success: &commonpb.Payload{Data: []byte("previous-result")}}
+			sched.LastCompletionResult = chasm.NewDataField(ctx, initial)
+			sched.Schedule.Policies.PauseOnFailure = true
+			sched.Invoker.Get(ctx).BufferedStarts = []*schedulespb.BufferedStart{{
+				RequestId: "req-1", WorkflowId: "wf-1", RunId: "run-1", Attempt: 1,
+				OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+				ActualTime:    timestamppb.New(time.Now().Add(-time.Minute)),
+				StartTime:     timestamppb.New(time.Now().Add(-30 * time.Second)),
+				Manual:        tc.manual,
+			}}
+
+			err := sched.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				RequestId: "req-1",
+				Outcome: &persistencespb.ChasmNexusCompletion_Failure{
+					Failure: &failurepb.Failure{Message: "allow-all failure"},
+				},
+				CloseTime: timestamppb.Now(),
+			})
+			require.NoError(t, err)
+			_, err = node.CloseTransaction()
+			require.NoError(t, err)
+
+			readCtx := chasm.NewContext(context.Background(), node)
+			require.Equal(t, initial, sched.LastCompletionResult.Get(readCtx))
+			require.False(t, sched.Schedule.State.Paused)
+			invoker := sched.Invoker.Get(readCtx)
+			require.Empty(t, invoker.GetBufferedStarts())
+			require.Len(t, sched.Info.GetRecentActions(), 1)
+			require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_FAILED, sched.Info.GetRecentActions()[0].GetStartWorkflowStatus())
+		})
+	}
+}
+
+func TestHandleNexusCompletion_IgnoredReasonMetric(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		setup  func(*scheduler.Invoker)
+		reason string
+	}{
+		{
+			name:   "unrecognized request",
+			reason: "unrecognized_request_id",
+		},
+		{
+			name: "already completed",
+			setup: func(invoker *scheduler.Invoker) {
+				invoker.BufferedStarts = []*schedulespb.BufferedStart{{
+					RequestId: "request-id",
+					Completed: &schedulespb.CompletedResult{
+						Status: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
+					},
+				}}
+			},
+			reason: "already_completed",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			recorder := metricstest.NewCaptureHandler()
+			capture := recorder.StartCapture()
+			defer recorder.StopCapture(capture)
+
+			logger := testlogger.NewTestLogger(t, testlogger.FailOnExpectedErrorOnly)
+			_, engineCtx := newTestEngineContext(t, logger, withEngineMetricsHandler(recorder))
+			_, err := chasm.StartExecution(
+				engineCtx,
+				chasm.ExecutionKey{NamespaceID: namespaceID, BusinessID: scheduleID},
+				scheduler.CreateScheduler,
+				&schedulerpb.CreateScheduleRequest{
+					NamespaceId: namespaceID,
+					FrontendRequest: &workflowservice.CreateScheduleRequest{
+						Namespace: namespace, ScheduleId: scheduleID, Schedule: defaultSchedule(), RequestId: "create-request",
+					},
+				},
+			)
+			require.NoError(t, err)
+			rootRef := chasm.NewComponentRef[*scheduler.Scheduler](chasm.ExecutionKey{NamespaceID: namespaceID, BusinessID: scheduleID})
+
+			_, _, err = chasm.UpdateComponent(engineCtx, rootRef,
+				func(s *scheduler.Scheduler, ctx chasm.MutableContext, _ struct{}) (struct{}, error) {
+					if tc.setup != nil {
+						tc.setup(s.Invoker.Get(ctx))
+					}
+					return struct{}{}, s.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{RequestId: "request-id"})
+				}, struct{}{})
+			require.NoError(t, err)
+
+			recordings := capture.Snapshot()[metrics.ScheduleCallbackIgnored.Name()]
+			require.Len(t, recordings, 1)
+			require.Equal(t, tc.reason, recordings[0].Tags["reason"])
+		})
+	}
+}
+
+func TestHandleNexusCompletion_MigratedRunningWorkflowKeepsCompletionState(t *testing.T) {
+	now := time.Now().UTC()
+	v1Schedule := defaultSchedule()
+	v1Schedule.Policies.OverlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL
+	v1Schedule.Policies.PauseOnFailure = true
+	v1Info := &schedulepb.ScheduleInfo{
+		RunningWorkflows: []*commonpb.WorkflowExecution{{WorkflowId: "wf-tracked", RunId: "run-tracked"}},
+	}
+	v1State := &schedulespb.InternalState{
+		Namespace:         namespace,
+		NamespaceId:       namespaceID,
+		ScheduleId:        scheduleID,
+		ConflictToken:     42,
+		LastProcessedTime: timestamppb.New(now),
+	}
+	req := migration.LegacyToCreateFromMigrationStateRequest(v1Schedule, v1Info, v1State, nil, nil, now)
+
+	var migrated *schedulespb.BufferedStart
+	for _, start := range req.GetState().GetInvokerState().GetBufferedStarts() {
+		if start.GetWorkflowId() == "wf-tracked" {
+			migrated = start
+			break
+		}
+	}
+	require.NotNil(t, migrated)
+	require.Equal(t, enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED, migrated.GetOverlapPolicy())
+
+	logger := testlogger.NewTestLogger(t, testlogger.FailOnExpectedErrorOnly)
+	_, engineCtx := newTestEngineContext(t, logger)
+	handler := scheduler.NewTestHandler(logger)
+	_, err := handler.TestCreateFromMigrationState(engineCtx, req)
+	require.NoError(t, err)
+
+	rootRef := chasm.NewComponentRef[*scheduler.Scheduler](chasm.ExecutionKey{
+		NamespaceID: namespaceID,
+		BusinessID:  scheduleID,
+	})
+	_, _, err = chasm.UpdateComponent(engineCtx, rootRef,
+		func(s *scheduler.Scheduler, ctx chasm.MutableContext, _ struct{}) (struct{}, error) {
+			return struct{}{}, s.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				RequestId: migrated.GetRequestId(),
+				Outcome: &persistencespb.ChasmNexusCompletion_Failure{
+					Failure: &failurepb.Failure{Message: "tracked workflow failed"},
+				},
+				CloseTime: timestamppb.New(now),
+			})
+		}, struct{}{})
+	require.NoError(t, err)
+
+	_, err = chasm.ReadComponent(engineCtx, rootRef,
+		func(s *scheduler.Scheduler, ctx chasm.Context, _ struct{}) (struct{}, error) {
+			require.Equal(t, "tracked workflow failed", s.LastCompletionResult.Get(ctx).GetFailure().GetMessage())
+			require.True(t, s.Schedule.State.Paused)
+			return struct{}{}, nil
+		}, struct{}{})
+	require.NoError(t, err)
+}
+
 // TestHandleNexusCompletion_Failure verifies that a failed workflow completion
 // is properly recorded with the failure payload and workflow status is updated.
 func TestHandleNexusCompletion_Failure(t *testing.T) {

chasm/lib/scheduler/scheduler_test.go

@@ -156,6 +156,70 @@ func TestListInfo_RecentActionsCapped(t *testing.T) {
 	}
 }
 
+func TestDescribe_RecentActionsCappedAcrossSources(t *testing.T) {
+	logger := testlogger.NewTestLogger(t, testlogger.FailOnExpectedErrorOnly)
+	_, engineCtx := newTestEngineContext(t, logger)
+	result, err := chasm.StartExecution(
+		engineCtx,
+		chasm.ExecutionKey{NamespaceID: namespaceID, BusinessID: scheduleID},
+		scheduler.CreateScheduler,
+		&schedulerpb.CreateScheduleRequest{
+			NamespaceId: namespaceID,
+			FrontendRequest: &workflowservice.CreateScheduleRequest{
+				Namespace:  namespace,
+				ScheduleId: scheduleID,
+				Schedule:   defaultSchedule(),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	rootRef := chasm.NewComponentRef[*scheduler.Scheduler](result.ExecutionKey)
+	base := time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC)
+	_, _, err = chasm.UpdateComponent(engineCtx, rootRef,
+		func(sched *scheduler.Scheduler, ctx chasm.MutableContext, _ struct{}) (struct{}, error) {
+			invoker := sched.Invoker.Get(ctx)
+			for i := range scheduler.RecentActionCount {
+				storedTime := timestamppb.New(base.Add(time.Duration(2*i) * time.Minute))
+				sched.Info.RecentActions = append(sched.Info.RecentActions, &schedulepb.ScheduleActionResult{
+					ScheduleTime: storedTime,
+					ActualTime:   storedTime,
+					StartWorkflowResult: &commonpb.WorkflowExecution{
+						WorkflowId: fmt.Sprintf("stored-workflow-%d", i),
+						RunId:      fmt.Sprintf("stored-run-%d", i),
+					},
+				})
+
+				trackedTime := timestamppb.New(base.Add(time.Duration(2*i+1) * time.Minute))
+				invoker.BufferedStarts = append(invoker.BufferedStarts, &schedulespb.BufferedStart{
+					WorkflowId: fmt.Sprintf("tracked-workflow-%d", i),
+					RunId:      fmt.Sprintf("tracked-run-%d", i),
+					StartTime:  trackedTime,
+				})
+			}
+			return struct{}{}, nil
+		}, struct{}{})
+	require.NoError(t, err)
+
+	_, err = chasm.ReadComponent(engineCtx, rootRef,
+		func(sched *scheduler.Scheduler, ctx chasm.Context, _ struct{}) (struct{}, error) {
+			response, err := sched.Describe(ctx, &schedulerpb.DescribeScheduleRequest{}, newLegacySpecBuilder(0, 0))
+			require.NoError(t, err)
+			actions := response.GetFrontendResponse().GetInfo().GetRecentActions()
+			require.Len(t, actions, scheduler.RecentActionCount)
+			for i, action := range actions {
+				wantIndex := i/2 + scheduler.RecentActionCount/2
+				if i%2 == 0 {
+					require.Equal(t, fmt.Sprintf("stored-run-%d", wantIndex), action.GetStartWorkflowResult().GetRunId())
+				} else {
+					require.Equal(t, fmt.Sprintf("tracked-run-%d", wantIndex), action.GetStartWorkflowResult().GetRunId())
+				}
+			}
+			return struct{}{}, nil
+		}, struct{}{})
+	require.NoError(t, err)
+}
+
 func TestCreateSchedulerFromMigration(t *testing.T) {
 	now := time.Now().UTC()
 	_, _, node := setupSchedulerForTest(t)

tests/schedule_test.go

@@ -521,6 +521,7 @@ func runSharedScheduleTests(t *testing.T, newContext contextFactory) {
 	t.Run("TestBasics", func(t *testing.T) { t.Parallel(); testBasics(t, newContext) })
 	t.Run("TestInput", func(t *testing.T) { t.Parallel(); testInput(t, newContext) })
 	t.Run("TestLastCompletionAndError", func(t *testing.T) { t.Parallel(); testLastCompletionAndError(t, newContext) })
+	t.Run("TestAllowAllDescribeContract", func(t *testing.T) { t.Parallel(); testAllowAllDescribeContract(t, newContext) })
 	t.Run("TestScheduleContinuesAfterWorkflowRetryFailure", func(t *testing.T) { t.Parallel(); testScheduleContinuesAfterWorkflowRetryFailure(t, newContext) })
 	t.Run("TestListSchedulesReturnsWorkflowStatus", func(t *testing.T) { t.Parallel(); testListSchedulesReturnsWorkflowStatus(t, newContext) })
 	t.Run("TestListSchedulesRecentActionsCapped", func(t *testing.T) { t.Parallel(); testListSchedulesRecentActionsCapped(t, newContext) })
@@ -567,6 +568,127 @@ func runSharedScheduleTests(t *testing.T, newContext contextFactory) {
 	t.Run("TestBufferOneDeferredFiresAfterCompletion", func(t *testing.T) { t.Parallel(); testBufferOneDeferredFiresAfterCompletion(t, newContext) })
 }
 
+// testAllowAllDescribeContract verifies the customer-facing Describe state shared by V1 and CHASM.
+// ALLOW_ALL executions appear in RecentActions but not RunningWorkflows; sequential executions remain active.
+func testAllowAllDescribeContract(t *testing.T, newContext contextFactory) {
+	s := newScheduleEnv(t, scheduleCommonOpts(t)...)
+	sid := testcore.RandomizeStr("sched-allow-all-active")
+	wid := testcore.RandomizeStr("sched-allow-all-active-wf")
+	wt := testcore.RandomizeStr("sched-allow-all-active-wt")
+
+	var runs atomic.Int32
+	s.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
+		_ = workflow.SideEffect(ctx, func(workflow.Context) any { runs.Add(1); return 0 })
+		failed := false
+		selector := workflow.NewSelector(ctx)
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "complete"), func(c workflow.ReceiveChannel, _ bool) {
+			c.Receive(ctx, nil)
+		})
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "fail"), func(c workflow.ReceiveChannel, _ bool) {
+			c.Receive(ctx, nil)
+			failed = true
+		})
+		selector.Select(ctx)
+		if failed {
+			return errors.New("allow-all failure")
+		}
+		return nil
+	}, workflow.RegisterOptions{Name: wt})
+
+	ctx := newContext(testcore.NewContext())
+	createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
+		Spec:     &schedulepb.ScheduleSpec{},
+		Action:   startWorkflowAction(s, wid, wt),
+		Policies: &schedulepb.SchedulePolicies{PauseOnFailure: true},
+	})
+
+	patchSchedule(ctx, t, s, sid, triggerPatch(enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL))
+	var allowAllRun *commonpb.WorkflowExecution
+	var allowAllDescribe *workflowservice.DescribeScheduleResponse
+	require.Eventually(t, func() bool {
+		desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
+			Namespace: s.Namespace().String(), ScheduleId: sid,
+		})
+		if err != nil || runs.Load() != 1 || desc.GetInfo().GetActionCount() != 1 ||
+			desc.GetInfo().GetBufferSize() != 0 || len(desc.GetInfo().GetRecentActions()) != 1 ||
+			len(desc.GetInfo().GetRunningWorkflows()) != 0 {
+			return false
+		}
+		recent := desc.GetInfo().GetRecentActions()[0]
+		if recent.GetStartWorkflowStatus() != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING ||
+			recent.GetScheduleTime() == nil || recent.GetActualTime() == nil {
+			return false
+		}
+		allowAllRun = recent.GetStartWorkflowResult()
+		if allowAllRun.GetRunId() == "" {
+			return false
+		}
+		allowAllDescribe = desc
+		return true
+	}, awaitTimeout, pollInterval, "ALLOW_ALL Describe state should be recent, running, and not active")
+	require.Empty(t, allowAllDescribe.GetInfo().GetRunningWorkflows())
+	require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, allowAllDescribe.GetInfo().GetRecentActions()[0].GetStartWorkflowStatus())
+
+	allowAllNominal, err := time.Parse(time.RFC3339, strings.TrimPrefix(allowAllRun.GetWorkflowId(), wid+"-"))
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		return time.Now().UTC().Truncate(time.Second).After(allowAllNominal)
+	}, awaitTimeout, pollInterval, "next trigger should receive a distinct timestamp-based workflow ID")
+
+	patchSchedule(ctx, t, s, sid, triggerPatch(enumspb.SCHEDULE_OVERLAP_POLICY_SKIP))
+	var sequentialRun *commonpb.WorkflowExecution
+	var sequentialDescribe *workflowservice.DescribeScheduleResponse
+	require.Eventually(t, func() bool {
+		desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
+			Namespace: s.Namespace().String(), ScheduleId: sid,
+		})
+		if err != nil || runs.Load() != 2 || desc.GetInfo().GetActionCount() != 2 ||
+			desc.GetInfo().GetBufferSize() != 0 || len(desc.GetInfo().GetRecentActions()) != 2 ||
+			len(desc.GetInfo().GetRunningWorkflows()) != 1 {
+			return false
+		}
+		sequentialRun = desc.GetInfo().GetRunningWorkflows()[0]
+		if sequentialRun.GetRunId() == "" || sequentialRun.GetRunId() == allowAllRun.GetRunId() {
+			return false
+		}
+		for _, recent := range desc.GetInfo().GetRecentActions() {
+			if recent.GetStartWorkflowStatus() != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING ||
+				recent.GetScheduleTime() == nil || recent.GetActualTime() == nil {
+				return false
+			}
+		}
+		sequentialDescribe = desc
+		return true
+	}, awaitTimeout, pollInterval, "ALLOW_ALL execution should not block a sequential trigger")
+	require.Equal(t, []*commonpb.WorkflowExecution{sequentialRun}, sequentialDescribe.GetInfo().GetRunningWorkflows())
+	require.ElementsMatch(t, []*commonpb.WorkflowExecution{allowAllRun, sequentialRun}, []*commonpb.WorkflowExecution{
+		sequentialDescribe.GetInfo().GetRecentActions()[0].GetStartWorkflowResult(),
+		sequentialDescribe.GetInfo().GetRecentActions()[1].GetStartWorkflowResult(),
+	})
+
+	require.NoError(t, s.SdkClient().SignalWorkflow(ctx, allowAllRun.GetWorkflowId(), allowAllRun.GetRunId(), "fail", nil))
+	require.Eventually(t, func() bool {
+		resp, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: s.Namespace().String(), Execution: allowAllRun,
+		})
+		return err == nil && resp.GetWorkflowExecutionInfo().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_FAILED
+	}, awaitTimeout, pollInterval, "ALLOW_ALL workflow should fail")
+
+	desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
+		Namespace: s.Namespace().String(), ScheduleId: sid,
+	})
+	require.NoError(t, err)
+	require.False(t, desc.GetSchedule().GetState().GetPaused())
+	require.Equal(t, int64(2), desc.GetInfo().GetActionCount())
+	require.Zero(t, desc.GetInfo().GetBufferSize())
+	require.Equal(t, []*commonpb.WorkflowExecution{sequentialRun}, desc.GetInfo().GetRunningWorkflows())
+	require.Len(t, desc.GetInfo().GetRecentActions(), 2)
+	for _, recent := range desc.GetInfo().GetRecentActions() {
+		require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, recent.GetStartWorkflowStatus())
+	}
+	require.NoError(t, s.SdkClient().SignalWorkflow(ctx, sequentialRun.GetWorkflowId(), sequentialRun.GetRunId(), "complete", nil))
+}
+
 // testBufferSizeReportedWhenBuffered verifies that ScheduleInfo.BufferSize is
 // populated by both the V1 and V2 (CHASM) schedulers when at least one fire is
 // queued behind a still-running workflow. A schedule with a 1s interval and
@@ -4842,6 +4964,7 @@ type scheduleClosesCase struct {
 	name         string
 	prefix       string
 	state        *schedulepb.ScheduleState
+	policies     *schedulepb.SchedulePolicies
 	expectedRuns int32
 
 	// buildSpec receives the current time at the moment the schedule is created
@@ -4892,6 +5015,17 @@ func testScheduleClosesFromIdle(t *testing.T, newContext contextFactory) {
 			state:          &schedulepb.ScheduleState{LimitedActions: true, RemainingActions: 2},
 			strictRunCount: true,
 		},
+		{
+			name:         "FinalAllowAllAction",
+			prefix:       "sched-final-allow-all-closes",
+			expectedRuns: 1,
+			buildSpec: func(_ time.Time) *schedulepb.ScheduleSpec {
+				return intervalSpec(fastInterval)
+			},
+			state:          &schedulepb.ScheduleState{LimitedActions: true, RemainingActions: 1},
+			policies:       &schedulepb.SchedulePolicies{OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL},
+			strictRunCount: true,
+		},
 		{
 			name:         "IntervalEndTime",
 			prefix:       "sched-interval-end-closes",
@@ -4923,9 +5057,10 @@ func runScheduleClosesFromIdleCase(t *testing.T, newContext contextFactory, c sc
 
 	ctx := newContext(s.Context())
 	createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
-		Spec:   c.buildSpec(time.Now().UTC()),
-		State:  c.state,
-		Action: startWorkflowAction(s, wid, wt),
+		Spec:     c.buildSpec(time.Now().UTC()),
+		Policies: c.policies,
+		State:    c.state,
+		Action:   startWorkflowAction(s, wid, wt),
 	})
 
 	// A hard action budget must land on exactly expectedRuns; time-bounded specs