diff --git a/chasm/lib/scheduler/buffer_planner.go b/chasm/lib/scheduler/buffer_planner.go
new file mode 100644
index 00000000000..ce0c32bd7f0
--- /dev/null
+++ b/chasm/lib/scheduler/buffer_planner.go
@@ -0,0 +1,207 @@
+package scheduler
+
+import (
+	"maps"
+	"slices"
+	"time"
+
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	schedulespb "go.temporal.io/server/api/schedule/v1"
+	"go.temporal.io/server/chasm"
+	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
+)
+
+// appliedBufferPlan collects the live-state result and the number of decisions invalidated during revalidation.
+type appliedBufferPlan struct {
+	result               processBufferResult
+	invalidatedDecisions int64
+}
+
+// newBufferProcessingSnapshot projects live CHASM state into values that the pure planner cannot mutate.
+func newBufferProcessingSnapshot(invoker *Invoker, scheduler *Scheduler, catchupWindow time.Duration) schedulerinternal.BufferProcessingSnapshot {
+	state := scheduler.Schedule.GetState()
+	snapshot := schedulerinternal.BufferProcessingSnapshot{
+		Starts:               make([]schedulerinternal.BufferedStartSnapshot, 0, len(invoker.GetBufferedStarts())),
+		DefaultOverlapPolicy: scheduler.overlapPolicy(),
+		CatchupWindow:        catchupWindow,
+		MinimumCatchupWindow: startWorkflowMinDeadline,
+		Paused:               state.GetPaused(),
+		LimitedActions:       state.GetLimitedActions(),
+		RemainingActions:     state.GetRemainingActions(),
+	}
+	for _, start := range invoker.GetBufferedStarts() {
+		projected := projectBufferedStart(start)
+		snapshot.Starts = append(snapshot.Starts, projected)
+		if projected.RunID != "" && !projected.Completed {
+			snapshot.RunningWorkflows = append(snapshot.RunningWorkflows, schedulerinternal.WorkflowExecutionSnapshot{
+				WorkflowID: projected.WorkflowID,
+				RunID:      projected.RunID,
+			})
+		}
+	}
+	return snapshot
+}
+
+func projectBufferedStart(start *schedulespb.BufferedStart) schedulerinternal.BufferedStartSnapshot {
+	return schedulerinternal.BufferedStartSnapshot{
+		RequestID:     start.GetRequestId(),
+		WorkflowID:    start.GetWorkflowId(),
+		RunID:         start.GetRunId(),
+		Attempt:       start.GetAttempt(),
+		Manual:        start.GetManual(),
+		OverlapPolicy: start.GetOverlapPolicy(),
+		ActualTime:    start.GetActualTime().AsTime(),
+		DesiredTime:   start.GetDesiredTime().AsTime(),
+		Completed:     start.GetCompleted() != nil,
+	}
+}
+
+// applyBufferPlan revalidates a plan, resolves its value decisions back to live
+// BufferedStart pointers, and applies only decisions whose expected state still matches.
+func applyBufferPlan(
+	ctx chasm.MutableContext,
+	scheduler *Scheduler,
+	invoker *Invoker,
+	plan schedulerinternal.BufferPlan,
+) appliedBufferPlan {
+	applied := newAppliedBufferPlan()
+	currentSnapshot := newBufferProcessingSnapshot(invoker, scheduler, plan.Snapshot.CatchupWindow)
+	if !bufferProcessingSnapshotsEqual(plan.Snapshot, currentSnapshot) {
+		return invalidateBufferPlan(plan, applied)
+	}
+	startsByRequestID := make(map[string][]*schedulespb.BufferedStart)
+	for _, start := range invoker.GetBufferedStarts() {
+		startsByRequestID[start.GetRequestId()] = append(startsByRequestID[start.GetRequestId()], start)
+	}
+
+	for _, decision := range plan.Decisions {
+		if !decision.MutatesState() {
+			continue
+		}
+		start, ok := popMatchingBufferedStart(startsByRequestID, decision)
+		if !ok {
+			applied.recordInvalidatedDecision()
+			continue
+		}
+
+		if decision.ConsumesScheduledAction && !scheduler.consumeScheduledAction() {
+			applied.recordInvalidatedDecision()
+			continue
+		}
+
+		applyBufferDecision(&applied.result, decision, start)
+	}
+	if applied.invalidatedDecisions == 0 {
+		applied.result.overlapSkipped = plan.OverlapSkipped
+		applied.result.overlapSkippedByPolicy = maps.Clone(plan.OverlapSkippedByPolicy)
+	}
+
+	for _, target := range plan.TerminateWorkflows {
+		if currentRunningWorkflow(invoker, target) {
+			applied.result.terminateWorkflows = append(applied.result.terminateWorkflows, workflowExecutionFromSnapshot(target))
+		}
+	}
+	for _, target := range plan.CancelWorkflows {
+		if currentRunningWorkflow(invoker, target) {
+			applied.result.cancelWorkflows = append(applied.result.cancelWorkflows, workflowExecutionFromSnapshot(target))
+		}
+	}
+
+	var totalMissedCatchup int64
+	for _, count := range applied.result.missedCatchupByActionRunning {
+		totalMissedCatchup += count
+	}
+	scheduler.recordActionResult(&schedulerActionResult{
+		overlapSkipped:      applied.result.overlapSkipped,
+		missedCatchupWindow: totalMissedCatchup,
+	})
+	invoker.recordProcessBufferResult(ctx, &applied.result)
+	return applied
+}
+
+func newAppliedBufferPlan() appliedBufferPlan {
+	return appliedBufferPlan{result: processBufferResult{
+		overlapSkippedByPolicy:       make(map[enumspb.ScheduleOverlapPolicy]int64),
+		missedCatchupByActionRunning: make(map[bool]int64),
+		processedStarts:              make(map[string]bool),
+	}}
+}
+
+func invalidateBufferPlan(plan schedulerinternal.BufferPlan, applied appliedBufferPlan) appliedBufferPlan {
+	for _, decision := range plan.Decisions {
+		if decision.MutatesState() {
+			applied.recordInvalidatedDecision()
+		}
+	}
+	return applied
+}
+
+func (a *appliedBufferPlan) recordInvalidatedDecision() {
+	a.invalidatedDecisions++
+}
+
+// popMatchingBufferedStart resolves one decision to its live protobuf pointer.
+// Removing the match ensures duplicate request IDs cannot reuse the same start.
+func popMatchingBufferedStart(
+	startsByRequestID map[string][]*schedulespb.BufferedStart,
+	decision schedulerinternal.BufferDecision,
+) (*schedulespb.BufferedStart, bool) {
+	matches := startsByRequestID[decision.RequestID]
+	for index, start := range matches {
+		if projectBufferedStart(start) == decision.Expected {
+			startsByRequestID[decision.RequestID] = append(matches[:index], matches[index+1:]...)
+			return start, true
+		}
+	}
+	return nil, false
+}
+
+func applyBufferDecision(result *processBufferResult, decision schedulerinternal.BufferDecision, start *schedulespb.BufferedStart) {
+	result.processedStarts[decision.RequestID] = true
+	switch decision.Action {
+	case schedulerinternal.BufferDecisionExecute:
+		result.startWorkflows = append(result.startWorkflows, start)
+	case schedulerinternal.BufferDecisionDiscard:
+		result.discardStarts = append(result.discardStarts, start)
+	default:
+	}
+	recordAppliedDecisionMetrics(result, decision)
+}
+
+func recordAppliedDecisionMetrics(result *processBufferResult, decision schedulerinternal.BufferDecision) {
+	switch decision.Reason {
+	case schedulerinternal.BufferDecisionReasonMissedCatchupWindow:
+		result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedMissedCatchup)
+	case schedulerinternal.BufferDecisionReasonPausedOrLimited:
+		result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedPausedOrLimited)
+	default:
+	}
+	if decision.MissedCatchupMetric {
+		result.missedCatchupByActionRunning[decision.MissedCatchupActionRunning]++
+	}
+}
+
+func bufferProcessingSnapshotsEqual(left, right schedulerinternal.BufferProcessingSnapshot) bool {
+	return left.DefaultOverlapPolicy == right.DefaultOverlapPolicy &&
+		left.CatchupWindow == right.CatchupWindow &&
+		left.MinimumCatchupWindow == right.MinimumCatchupWindow &&
+		left.Paused == right.Paused &&
+		left.LimitedActions == right.LimitedActions &&
+		left.RemainingActions == right.RemainingActions &&
+		slices.Equal(left.Starts, right.Starts) &&
+		slices.Equal(left.RunningWorkflows, right.RunningWorkflows)
+}
+
+func currentRunningWorkflow(invoker *Invoker, target schedulerinternal.WorkflowExecutionSnapshot) bool {
+	for _, start := range invoker.GetBufferedStarts() {
+		if start.GetWorkflowId() == target.WorkflowID && start.GetRunId() == target.RunID && start.GetCompleted() == nil {
+			return true
+		}
+	}
+	return false
+}
+
+func workflowExecutionFromSnapshot(execution schedulerinternal.WorkflowExecutionSnapshot) *commonpb.WorkflowExecution {
+	return &commonpb.WorkflowExecution{WorkflowId: execution.WorkflowID, RunId: execution.RunID}
+}
diff --git a/chasm/lib/scheduler/buffer_processor_legacy_test.go b/chasm/lib/scheduler/buffer_processor_legacy_test.go
new file mode 100644
index 00000000000..621de6f513d
--- /dev/null
+++ b/chasm/lib/scheduler/buffer_processor_legacy_test.go
@@ -0,0 +1,97 @@
+package scheduler
+
+import (
+	enumspb "go.temporal.io/api/enums/v1"
+	schedulespb "go.temporal.io/server/api/schedule/v1"
+	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/common/util"
+	legacyscheduler "go.temporal.io/server/service/worker/scheduler"
+)
+
+func (h *InvokerProcessBufferTaskHandler) processBufferLegacy(
+	ctx chasm.MutableContext,
+	invoker *Invoker,
+	scheduler *Scheduler,
+) (result processBufferResult) {
+	runningWorkflows := invoker.runningWorkflowExecutions()
+	isRunning := len(runningWorkflows) > 0
+	result.missedCatchupByActionRunning = make(map[bool]int64)
+
+	// Processing ignores starts that are already executing or backing off. An existing
+	// deferred BUFFER_ONE start still participates so it can reject later starts.
+	pendingBufferedStarts := util.FilterSlice(invoker.GetBufferedStarts(), func(start *schedulespb.BufferedStart) bool {
+		return start.Attempt == 0 ||
+			(start.Attempt == -1 && scheduler.resolveOverlapPolicy(start.GetOverlapPolicy()) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE)
+	})
+
+	// Resolve overlap policies and trim BufferedStarts that are skipped by policy.
+	action := legacyscheduler.ProcessBuffer(pendingBufferedStarts, isRunning, scheduler.resolveOverlapPolicy)
+
+	// ProcessBuffer will drop starts by omitting them from NewBuffer. Start with the
+	// diff between the input and NewBuffer, and add any executing starts.
+	keepStarts := make(map[string]struct{}) // request ID -> is present
+	for _, start := range action.NewBuffer {
+		keepStarts[start.GetRequestId()] = struct{}{}
+	}
+
+	// Combine all available starts.
+	readyStarts := action.OverlappingStarts
+	if action.NonOverlappingStart != nil {
+		readyStarts = append(readyStarts, action.NonOverlappingStart)
+	}
+
+	// Update result metrics.
+	result.overlapSkipped = action.OverlapSkipped
+	result.overlapSkippedByPolicy = action.OverlapSkippedByPolicy
+
+	// Add starting workflows to result, trim others. Catchup-window expiry is
+	// checked before consumeScheduledAction so that a start past its catchup
+	// window doesn't consume a LimitedActions slot.
+	for _, start := range readyStarts {
+		deadline := h.startWorkflowDeadline(ctx, scheduler, start)
+		if ctx.Now(invoker).After(deadline) {
+			// Action was buffered in time but expired before execution
+			// (e.g., due to overlap deferral, retries, or system delay).
+			// Only emit the metric if the schedule would have run this
+			// start -- skip paused or action-exhausted schedules.
+			if start.Manual || scheduler.canTakeScheduledAction() {
+				// Determine if a running action contributed: either one is still
+				// running, or the previous action's CloseTime (stored in DesiredTime)
+				// was already past this start's deadline.
+				// Note: if no prior action completed, DesiredTime is zero-valued,
+				// so After(deadline) is false, correctly yielding actionRunning=false.
+				actionRunning := isRunning ||
+					start.GetDesiredTime().AsTime().After(deadline)
+				result.missedCatchupByActionRunning[actionRunning]++
+			}
+			result.discardStarts = append(result.discardStarts, start)
+			result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedMissedCatchup)
+			continue
+		}
+
+		// Ensure we can take more actions. Manual actions are always allowed.
+		if !start.Manual && !scheduler.consumeScheduledAction() {
+			// Drop buffered automated actions while paused or out of actions.
+			result.discardStarts = append(result.discardStarts, start)
+			result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedPausedOrLimited)
+			continue
+		}
+
+		keepStarts[start.GetRequestId()] = struct{}{}
+		result.startWorkflows = append(result.startWorkflows, start)
+	}
+
+	result.discardStarts = util.FilterSlice(pendingBufferedStarts, func(start *schedulespb.BufferedStart) bool {
+		_, keep := keepStarts[start.GetRequestId()]
+		return !keep
+	})
+
+	// Terminate overrides cancel if both are requested.
+	if action.NeedTerminate {
+		result.terminateWorkflows = runningWorkflows
+	} else if action.NeedCancel {
+		result.cancelWorkflows = runningWorkflows
+	}
+
+	return
+}
diff --git a/chasm/lib/scheduler/export_test.go b/chasm/lib/scheduler/export_test.go
index 8880a4efd63..175da50faa8 100644
--- a/chasm/lib/scheduler/export_test.go
+++ b/chasm/lib/scheduler/export_test.go
@@ -7,7 +7,10 @@ import (
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/metrics"
+	queueerrors "go.temporal.io/server/service/history/queues/errors"
 	legacyscheduler "go.temporal.io/server/service/worker/scheduler"
 )
 
@@ -72,6 +75,47 @@ func (i *Invoker) RecordExecuteResult(
 	})
 }
 
+func (h *InvokerProcessBufferTaskHandler) PlanBufferProcessingForTest(
+	invoker *Invoker,
+	scheduler *Scheduler,
+	now time.Time,
+) func(chasm.MutableContext, *Scheduler, *Invoker) int64 {
+	tweakables := h.config.Tweakables(scheduler.Namespace)
+	snapshot := newBufferProcessingSnapshot(invoker, scheduler, catchupWindow(scheduler, tweakables))
+	plan := schedulerinternal.PlanBufferProcessing(snapshot, now)
+	return func(ctx chasm.MutableContext, scheduler *Scheduler, invoker *Invoker) int64 {
+		return applyBufferPlan(ctx, scheduler, invoker, plan).invalidatedDecisions
+	}
+}
+
+func (h *InvokerProcessBufferTaskHandler) ExecuteProcessBufferLegacyForTest(
+	ctx chasm.MutableContext,
+	invoker *Invoker,
+) error {
+	scheduler := invoker.Scheduler.Get(ctx)
+	newTaggedMetricsHandler(h.metricsHandler, scheduler).
+		Counter(metrics.ScheduleInvokerProcessBufferTask.Name()).
+		Record(1, metrics.OutcomeTag(outcomeFired), metrics.ReasonTag(reasonNone))
+
+	invoker.getOrCreateEventLog(ctx).LogEvent(ctx, "processBufferTask executed")
+	if scheduler.Schedule.GetAction().GetStartWorkflow() == nil {
+		return queueerrors.NewUnprocessableTaskError("schedules must have an Action set")
+	}
+
+	result := h.processBufferLegacy(ctx, invoker, scheduler)
+	var totalMissedCatchup int64
+	for _, count := range result.missedCatchupByActionRunning {
+		totalMissedCatchup += count
+	}
+	scheduler.recordActionResult(&schedulerActionResult{
+		overlapSkipped:      result.overlapSkipped,
+		missedCatchupWindow: totalMissedCatchup,
+	})
+	invoker.recordProcessBufferResult(ctx, &result)
+	h.recordBufferProcessingMetrics(scheduler, result)
+	return nil
+}
+
 func (b *BackfillerTaskHandler) ProcessBackfill(
 	scheduler *Scheduler,
 	backfiller *Backfiller,
diff --git a/chasm/lib/scheduler/internal/buffer_planner.go b/chasm/lib/scheduler/internal/buffer_planner.go
new file mode 100644
index 00000000000..ed193b43fef
--- /dev/null
+++ b/chasm/lib/scheduler/internal/buffer_planner.go
@@ -0,0 +1,323 @@
+package internal
+
+import (
+	"slices"
+	"time"
+
+	enumspb "go.temporal.io/api/enums/v1"
+)
+
+// BufferDecisionAction describes how the apply phase should treat one buffered start.
+type BufferDecisionAction int
+
+const (
+	BufferDecisionRetain BufferDecisionAction = iota
+	BufferDecisionExecute
+	BufferDecisionDiscard
+	BufferDecisionDefer
+	BufferDecisionRetry
+	BufferDecisionRunning
+	BufferDecisionCompleted
+)
+
+// BufferDecisionReason records why the planner selected an action.
+type BufferDecisionReason int
+
+const (
+	BufferDecisionReasonNone BufferDecisionReason = iota
+	BufferDecisionReasonAlreadyProcessed
+	BufferDecisionReasonOverlapPolicy
+	BufferDecisionReasonCancelOrTerminate
+	BufferDecisionReasonMissedCatchupWindow
+	BufferDecisionReasonPausedOrLimited
+)
+
+// BufferedStartSnapshot is the value-only planner projection of a persisted BufferedStart.
+// Keeping protobuf pointers out of the planner prevents planning from mutating live CHASM state.
+type BufferedStartSnapshot struct {
+	RequestID     string
+	WorkflowID    string
+	RunID         string
+	Attempt       int64
+	Manual        bool
+	OverlapPolicy enumspb.ScheduleOverlapPolicy
+	ActualTime    time.Time
+	DesiredTime   time.Time
+	Completed     bool
+}
+
+// WorkflowExecutionSnapshot identifies a running workflow without retaining a protobuf pointer.
+type WorkflowExecutionSnapshot struct {
+	WorkflowID string
+	RunID      string
+}
+
+// BufferProcessingSnapshot contains all state read by one buffer-planning pass.
+type BufferProcessingSnapshot struct {
+	Starts               []BufferedStartSnapshot
+	RunningWorkflows     []WorkflowExecutionSnapshot
+	DefaultOverlapPolicy enumspb.ScheduleOverlapPolicy
+	CatchupWindow        time.Duration
+	MinimumCatchupWindow time.Duration
+	Paused               bool
+	LimitedActions       bool
+	RemainingActions     int64
+}
+
+// BufferDecision describes the planned outcome for one exact BufferedStart snapshot.
+type BufferDecision struct {
+	RequestID string
+	// Expected is matched against current persisted state before any mutation is applied.
+	Expected BufferedStartSnapshot
+	Action   BufferDecisionAction
+	Reason   BufferDecisionReason
+	// ConsumesScheduledAction defers the live capacity decrement until after revalidation.
+	ConsumesScheduledAction    bool
+	MissedCatchupMetric        bool
+	MissedCatchupActionRunning bool
+}
+
+// MutatesState reports whether applying the decision changes the persisted buffer.
+func (d BufferDecision) MutatesState() bool {
+	return d.Action == BufferDecisionExecute || d.Action == BufferDecisionDiscard || d.Action == BufferDecisionDefer
+}
+
+// BufferPlan is an ordered, value-based description of a buffer-processing pass.
+// Snapshot anchors whole-plan validation; each decision also carries its exact expected start.
+type BufferPlan struct {
+	Snapshot               BufferProcessingSnapshot
+	Decisions              []BufferDecision
+	CancelWorkflows        []WorkflowExecutionSnapshot
+	TerminateWorkflows     []WorkflowExecutionSnapshot
+	OverlapSkipped         int64
+	OverlapSkippedByPolicy map[enumspb.ScheduleOverlapPolicy]int64
+}
+
+type overlapAction struct {
+	OverlappingStarts      []BufferedStartSnapshot
+	NonOverlappingStart    BufferedStartSnapshot
+	NewBuffer              []BufferedStartSnapshot
+	NeedCancel             bool
+	NeedTerminate          bool
+	OverlapSkipped         int64
+	OverlapSkippedByPolicy map[enumspb.ScheduleOverlapPolicy]int64
+}
+
+// PlanBufferProcessing computes buffer outcomes without mutating snapshot or live scheduler state.
+func PlanBufferProcessing(snapshot BufferProcessingSnapshot, now time.Time) BufferPlan {
+	plan := BufferPlan{
+		Snapshot:               cloneBufferProcessingSnapshot(snapshot),
+		OverlapSkippedByPolicy: make(map[enumspb.ScheduleOverlapPolicy]int64),
+	}
+	resolveOverlapPolicy := func(policy enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy {
+		if policy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
+			return snapshot.DefaultOverlapPolicy
+		}
+		return policy
+	}
+
+	pending := pendingBufferedStarts(snapshot.Starts, resolveOverlapPolicy)
+	action := planOverlapActions(pending, len(snapshot.RunningWorkflows) > 0, resolveOverlapPolicy)
+	plan.OverlapSkipped = action.OverlapSkipped
+	plan.OverlapSkippedByPolicy = action.OverlapSkippedByPolicy
+	if action.NeedTerminate {
+		plan.TerminateWorkflows = append(plan.TerminateWorkflows, snapshot.RunningWorkflows...)
+	} else if action.NeedCancel {
+		plan.CancelWorkflows = append(plan.CancelWorkflows, snapshot.RunningWorkflows...)
+	}
+
+	keep := make(map[string]struct{}, len(action.NewBuffer)+len(action.OverlappingStarts)+1)
+	for _, start := range action.NewBuffer {
+		keep[start.RequestID] = struct{}{}
+	}
+	readyStarts := slices.Clone(action.OverlappingStarts)
+	if action.NonOverlappingStart.RequestID != "" {
+		readyStarts = append(readyStarts, action.NonOverlappingStart)
+	}
+
+	// All ready decisions share this task-wide budget. Passing its address preserves
+	// legacy ready-start ordering while keeping the input snapshot immutable.
+	remainingActions := snapshot.RemainingActions
+	ready := make(map[string]struct{}, len(readyStarts))
+	readyDecisions := make([]BufferDecision, 0, len(readyStarts))
+	readyOccurrences := make(map[BufferedStartSnapshot]int, len(readyStarts))
+	for _, start := range readyStarts {
+		decision := planReadyBufferDecision(start, snapshot, now, &remainingActions)
+		if decision.Action == BufferDecisionExecute {
+			keep[start.RequestID] = struct{}{}
+			ready[start.RequestID] = struct{}{}
+		}
+		readyDecisions = append(readyDecisions, decision)
+		readyOccurrences[start]++
+	}
+
+	for _, decision := range readyDecisions {
+		decision.Action = finalPendingBufferAction(decision.RequestID, keep, ready)
+		plan.Decisions = append(plan.Decisions, decision)
+	}
+	for _, start := range snapshot.Starts {
+		if !isPendingBufferedStart(start, resolveOverlapPolicy) {
+			plan.Decisions = append(plan.Decisions, alreadyProcessedBufferDecision(start))
+			continue
+		}
+		if readyOccurrences[start] > 0 {
+			readyOccurrences[start]--
+			continue
+		}
+		decision := BufferDecision{
+			RequestID: start.RequestID,
+			Expected:  start,
+			Action:    finalPendingBufferAction(start.RequestID, keep, ready),
+			Reason:    BufferDecisionReasonOverlapPolicy,
+		}
+		if decision.Action == BufferDecisionDiscard && (action.NeedCancel || action.NeedTerminate) {
+			decision.Reason = BufferDecisionReasonCancelOrTerminate
+		}
+		plan.Decisions = append(plan.Decisions, decision)
+	}
+	return plan
+}
+
+func planOverlapActions(
+	buffer []BufferedStartSnapshot,
+	isRunning bool,
+	resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy,
+) overlapAction {
+	action := overlapAction{OverlapSkippedByPolicy: make(map[enumspb.ScheduleOverlapPolicy]int64)}
+	for _, start := range buffer {
+		overlapPolicy := resolve(start.OverlapPolicy)
+		if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL {
+			action.OverlappingStarts = append(action.OverlappingStarts, start)
+			continue
+		}
+		if !isRunning && action.NonOverlappingStart.RequestID == "" {
+			action.NonOverlappingStart = start
+			continue
+		}
+		switch overlapPolicy {
+		case enumspb.SCHEDULE_OVERLAP_POLICY_SKIP:
+			action.OverlapSkipped++
+			action.OverlapSkippedByPolicy[overlapPolicy]++
+		case enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE:
+			if len(action.NewBuffer) == 0 {
+				action.NewBuffer = append(action.NewBuffer, start)
+			} else {
+				action.OverlapSkipped++
+				action.OverlapSkippedByPolicy[overlapPolicy]++
+			}
+		case enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL:
+			action.NewBuffer = append(action.NewBuffer, start)
+		case enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER:
+			if isRunning {
+				action.NeedCancel = true
+				action.NewBuffer = append(action.NewBuffer, start)
+			} else {
+				action.NonOverlappingStart = start
+			}
+		case enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER:
+			if isRunning {
+				action.NeedTerminate = true
+				action.NewBuffer = append(action.NewBuffer, start)
+			} else {
+				action.NonOverlappingStart = start
+			}
+		default:
+		}
+	}
+	if action.NeedCancel || action.NeedTerminate {
+		action.OverlappingStarts = nil
+	}
+	return action
+}
+
+func pendingBufferedStarts(
+	starts []BufferedStartSnapshot,
+	resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy,
+) []BufferedStartSnapshot {
+	pending := make([]BufferedStartSnapshot, 0, len(starts))
+	for _, start := range starts {
+		if isPendingBufferedStart(start, resolve) {
+			pending = append(pending, start)
+		}
+	}
+	return pending
+}
+
+func isPendingBufferedStart(
+	start BufferedStartSnapshot,
+	resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy,
+) bool {
+	return start.Attempt == bufferedStartUnprocessedAttempt ||
+		(start.Attempt == bufferedStartDeferredAttempt &&
+			resolve(start.OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE)
+}
+
+func alreadyProcessedBufferDecision(start BufferedStartSnapshot) BufferDecision {
+	decision := BufferDecision{
+		RequestID: start.RequestID,
+		Expected:  start,
+		Action:    BufferDecisionRetain,
+		Reason:    BufferDecisionReasonAlreadyProcessed,
+	}
+	switch {
+	case start.Completed:
+		decision.Action = BufferDecisionCompleted
+	case start.RunID != "":
+		decision.Action = BufferDecisionRunning
+	case start.Attempt > bufferedStartFirstExecutionAttempt:
+		decision.Action = BufferDecisionRetry
+	default:
+	}
+	return decision
+}
+
+func planReadyBufferDecision(
+	start BufferedStartSnapshot,
+	snapshot BufferProcessingSnapshot,
+	now time.Time,
+	remainingActions *int64,
+) BufferDecision {
+	decision := BufferDecision{RequestID: start.RequestID, Expected: start}
+	deadline := start.ActualTime.Add(max(snapshot.CatchupWindow, snapshot.MinimumCatchupWindow))
+	canTakeScheduledAction := !snapshot.Paused && (!snapshot.LimitedActions || *remainingActions > 0)
+	if !start.Manual && now.After(deadline) {
+		decision.Action = BufferDecisionDiscard
+		decision.Reason = BufferDecisionReasonMissedCatchupWindow
+		if canTakeScheduledAction {
+			decision.MissedCatchupMetric = true
+			decision.MissedCatchupActionRunning = len(snapshot.RunningWorkflows) > 0 || start.DesiredTime.After(deadline)
+		}
+		return decision
+	}
+	if !start.Manual && !canTakeScheduledAction {
+		decision.Action = BufferDecisionDiscard
+		decision.Reason = BufferDecisionReasonPausedOrLimited
+		return decision
+	}
+	if !start.Manual && snapshot.LimitedActions {
+		*remainingActions--
+	}
+	decision.Action = BufferDecisionExecute
+	decision.ConsumesScheduledAction = !start.Manual
+	return decision
+}
+
+func finalPendingBufferAction(
+	requestID string,
+	keep map[string]struct{},
+	ready map[string]struct{},
+) BufferDecisionAction {
+	if _, ok := ready[requestID]; ok {
+		return BufferDecisionExecute
+	}
+	if _, ok := keep[requestID]; ok {
+		return BufferDecisionDefer
+	}
+	return BufferDecisionDiscard
+}
+
+func cloneBufferProcessingSnapshot(snapshot BufferProcessingSnapshot) BufferProcessingSnapshot {
+	snapshot.Starts = slices.Clone(snapshot.Starts)
+	snapshot.RunningWorkflows = slices.Clone(snapshot.RunningWorkflows)
+	return snapshot
+}
diff --git a/chasm/lib/scheduler/internal/buffer_planner_test.go b/chasm/lib/scheduler/internal/buffer_planner_test.go
new file mode 100644
index 00000000000..b0290d3254e
--- /dev/null
+++ b/chasm/lib/scheduler/internal/buffer_planner_test.go
@@ -0,0 +1,170 @@
+package internal
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	enumspb "go.temporal.io/api/enums/v1"
+)
+
+func TestPlanBufferProcessing_OverlapPolicies(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	tests := []struct {
+		name           string
+		policy         enumspb.ScheduleOverlapPolicy
+		action         BufferDecisionAction
+		reason         BufferDecisionReason
+		cancelCount    int
+		terminateCount int
+		overlapSkipped int64
+	}{
+		{name: "allow all", policy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, action: BufferDecisionExecute},
+		{name: "skip", policy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, action: BufferDecisionDiscard, reason: BufferDecisionReasonOverlapPolicy, overlapSkipped: 1},
+		{name: "buffer one", policy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy},
+		{name: "buffer all", policy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy},
+		{name: "cancel other", policy: enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy, cancelCount: 1},
+		{name: "terminate other", policy: enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy, terminateCount: 1},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			snapshot := BufferProcessingSnapshot{
+				Starts: []BufferedStartSnapshot{{
+					RequestID:     "request",
+					OverlapPolicy: test.policy,
+					ActualTime:    now,
+					DesiredTime:   now,
+				}},
+				RunningWorkflows:     []WorkflowExecutionSnapshot{{WorkflowID: "running", RunID: "run"}},
+				DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
+				CatchupWindow:        time.Hour,
+				MinimumCatchupWindow: 5 * time.Second,
+			}
+
+			plan := PlanBufferProcessing(snapshot, now)
+
+			require.Len(t, plan.Decisions, 1)
+			require.Equal(t, test.action, plan.Decisions[0].Action)
+			require.Equal(t, test.reason, plan.Decisions[0].Reason)
+			require.Len(t, plan.CancelWorkflows, test.cancelCount)
+			require.Len(t, plan.TerminateWorkflows, test.terminateCount)
+			require.Equal(t, test.overlapSkipped, plan.OverlapSkipped)
+		})
+	}
+}
+
+func TestPlanBufferProcessing_TimeAndCapacity(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	snapshot := BufferProcessingSnapshot{
+		Starts: []BufferedStartSnapshot{
+			{RequestID: "expired", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now.Add(-2 * time.Hour), DesiredTime: now.Add(-2 * time.Hour)},
+			{RequestID: "automatic", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now, DesiredTime: now},
+			{RequestID: "limited", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now, DesiredTime: now},
+			{RequestID: "manual", Manual: true, OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now.Add(-24 * time.Hour), DesiredTime: now},
+		},
+		DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
+		CatchupWindow:        time.Hour,
+		MinimumCatchupWindow: 5 * time.Second,
+		LimitedActions:       true,
+		RemainingActions:     1,
+	}
+
+	plan := PlanBufferProcessing(snapshot, now)
+
+	require.Equal(t, []BufferDecisionAction{
+		BufferDecisionDiscard,
+		BufferDecisionExecute,
+		BufferDecisionDiscard,
+		BufferDecisionExecute,
+	}, decisionActions(plan.Decisions))
+	require.Equal(t, BufferDecisionReasonMissedCatchupWindow, plan.Decisions[0].Reason)
+	require.Equal(t, BufferDecisionReasonPausedOrLimited, plan.Decisions[2].Reason)
+	require.True(t, plan.Decisions[0].MissedCatchupMetric)
+	require.False(t, plan.Decisions[0].MissedCatchupActionRunning)
+}
+
+func TestPlanBufferProcessing_PreservesLegacyReadyOrder(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	snapshot := BufferProcessingSnapshot{
+		Starts: []BufferedStartSnapshot{
+			{RequestID: "non-overlapping", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, ActualTime: now},
+			{RequestID: "allow-all", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now},
+		},
+		DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
+		CatchupWindow:        time.Hour,
+		MinimumCatchupWindow: 5 * time.Second,
+		LimitedActions:       true,
+		RemainingActions:     1,
+	}
+
+	plan := PlanBufferProcessing(snapshot, now)
+
+	require.Len(t, plan.Decisions, 2)
+	require.Equal(t, "allow-all", plan.Decisions[0].RequestID)
+	require.Equal(t, BufferDecisionExecute, plan.Decisions[0].Action)
+	require.True(t, plan.Decisions[0].ConsumesScheduledAction)
+	require.Equal(t, "non-overlapping", plan.Decisions[1].RequestID)
+	require.Equal(t, BufferDecisionDiscard, plan.Decisions[1].Action)
+	require.Equal(t, BufferDecisionReasonPausedOrLimited, plan.Decisions[1].Reason)
+}
+
+func TestPlanBufferProcessing_PausedAndProcessedStates(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	snapshot := BufferProcessingSnapshot{
+		Starts: []BufferedStartSnapshot{
+			{RequestID: "automatic", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now},
+			{RequestID: "manual", Manual: true, OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now},
+			{RequestID: "backing-off", Attempt: 2, ActualTime: now},
+			{RequestID: "running", Attempt: 1, RunID: "run", ActualTime: now},
+			{RequestID: "completed", Attempt: 1, RunID: "run", Completed: true, ActualTime: now},
+		},
+		DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
+		CatchupWindow:        time.Hour,
+		MinimumCatchupWindow: 5 * time.Second,
+		Paused:               true,
+	}
+
+	plan := PlanBufferProcessing(snapshot, now)
+
+	require.Equal(t, []BufferDecisionAction{
+		BufferDecisionDiscard,
+		BufferDecisionExecute,
+		BufferDecisionRetry,
+		BufferDecisionRunning,
+		BufferDecisionCompleted,
+	}, decisionActions(plan.Decisions))
+}
+
+func TestPlanBufferProcessing_DoesNotMutateOrAliasInput(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	snapshot := BufferProcessingSnapshot{
+		Starts: []BufferedStartSnapshot{{
+			RequestID: "request", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now,
+		}},
+		DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP,
+		CatchupWindow:        time.Hour,
+		MinimumCatchupWindow: 5 * time.Second,
+	}
+	want := BufferProcessingSnapshot{
+		Starts:               append([]BufferedStartSnapshot(nil), snapshot.Starts...),
+		DefaultOverlapPolicy: snapshot.DefaultOverlapPolicy,
+		CatchupWindow:        snapshot.CatchupWindow,
+		MinimumCatchupWindow: snapshot.MinimumCatchupWindow,
+	}
+
+	plan := PlanBufferProcessing(snapshot, now)
+
+	require.True(t, reflect.DeepEqual(want, snapshot))
+	plan.Decisions[0].Expected.RequestID = "changed"
+	plan.Snapshot.Starts[0].RequestID = "changed-again"
+	require.Equal(t, "request", snapshot.Starts[0].RequestID)
+}
+
+func decisionActions(decisions []BufferDecision) []BufferDecisionAction {
+	actions := make([]BufferDecisionAction, 0, len(decisions))
+	for _, decision := range decisions {
+		actions = append(actions, decision.Action)
+	}
+	return actions
+}
diff --git a/chasm/lib/scheduler/internal/buffered_start.go b/chasm/lib/scheduler/internal/buffered_start.go
index 8a316469796..b60a3340ea2 100644
--- a/chasm/lib/scheduler/internal/buffered_start.go
+++ b/chasm/lib/scheduler/internal/buffered_start.go
@@ -20,12 +20,18 @@ const (
 	BufferedStartStateCompleted
 )
 
+const (
+	bufferedStartDeferredAttempt       int64 = -1
+	bufferedStartUnprocessedAttempt    int64 = 0
+	bufferedStartFirstExecutionAttempt int64 = 1
+)
+
 // ClassifyBufferedStart returns the lifecycle state encoded by start at retryEvaluationTime.
 func ClassifyBufferedStart(
 	start *schedulespb.BufferedStart,
 	retryEvaluationTime time.Time,
 ) BufferedStartState {
-	if start == nil || start.GetAttempt() < -1 {
+	if start == nil || start.GetAttempt() < bufferedStartDeferredAttempt {
 		return BufferedStartStateInvalid
 	}
 
@@ -34,12 +40,12 @@ func ClassifyBufferedStart(
 	hasBackoff := start.GetBackoffTime() != nil
 
 	switch start.GetAttempt() {
-	case -1:
+	case bufferedStartDeferredAttempt:
 		if hasRun || hasCompletion || hasBackoff {
 			return BufferedStartStateInvalid
 		}
 		return BufferedStartStateDeferred
-	case 0:
+	case bufferedStartUnprocessedAttempt:
 		if hasRun || hasCompletion || hasBackoff {
 			return BufferedStartStateInvalid
 		}
@@ -63,7 +69,7 @@ func ClassifyBufferedStart(
 
 // MarkStartUnprocessed marks a start for its initial overlap-policy pass.
 func MarkStartUnprocessed(start *schedulespb.BufferedStart) {
-	start.Attempt = 0
+	start.Attempt = bufferedStartUnprocessedAttempt
 }
 
 // MarkStartMigratedUnprocessed clears V2 retry state from a pending V1 start.
@@ -74,12 +80,12 @@ func MarkStartMigratedUnprocessed(start *schedulespb.BufferedStart) {
 
 // MarkStartDeferred marks a start as waiting for an overlapping action to complete.
 func MarkStartDeferred(start *schedulespb.BufferedStart) {
-	start.Attempt = -1
+	start.Attempt = bufferedStartDeferredAttempt
 }
 
 // MarkStartReady marks a start as ready for its first execution attempt.
 func MarkStartReady(start *schedulespb.BufferedStart) {
-	start.Attempt = 1
+	start.Attempt = bufferedStartFirstExecutionAttempt
 }
 
 // MarkStartRetrying advances a start to its next attempt and backoff deadline.
diff --git a/chasm/lib/scheduler/invoker.go b/chasm/lib/scheduler/invoker.go
index 91bdc663e16..b7277e26c30 100644
--- a/chasm/lib/scheduler/invoker.go
+++ b/chasm/lib/scheduler/invoker.go
@@ -12,6 +12,7 @@ import (
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
 	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
+	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/util"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -93,6 +94,11 @@ type processBufferResult struct {
 	// Number of buffered starts dropped from missing the catchup window,
 	// bucketed by whether a running action contributed to the miss.
 	missedCatchupByActionRunning map[bool]int64
+	bufferedStartDropReasons     []metrics.ReasonString
+
+	// processedStarts limits lifecycle transitions to identities covered by a
+	// revalidated plan. A nil map preserves the legacy processor's behavior.
+	processedStarts map[string]bool
 }
 
 // recordProcessBufferResult updates the Invoker's internal state based on result, as well as the
@@ -120,7 +126,7 @@ func (i *Invoker) recordProcessBufferResult(ctx chasm.MutableContext, result *pr
 		if ready[start.RequestId] && start.Attempt < 1 {
 			schedulerinternal.MarkStartReady(start)
 			readiedStarts++
-		} else if start.Attempt == 0 {
+		} else if start.Attempt == 0 && (result.processedStarts == nil || result.processedStarts[start.RequestId]) {
 			// Start was processed but deferred (e.g., BUFFER_ONE policy with running workflow).
 			// Mark as deferred (-1) to distinguish from newly-enqueued starts so addTasks
 			// won't schedule an immediate ProcessBuffer task for them - they wait on
diff --git a/chasm/lib/scheduler/invoker_process_buffer_task_test.go b/chasm/lib/scheduler/invoker_process_buffer_task_test.go
index 87735003005..8dc6742d640 100644
--- a/chasm/lib/scheduler/invoker_process_buffer_task_test.go
+++ b/chasm/lib/scheduler/invoker_process_buffer_task_test.go
@@ -2,13 +2,17 @@ package scheduler_test
 
 import (
 	"context"
+	"fmt"
 	"slices"
+	"sort"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 	commonpb "go.temporal.io/api/common/v1"
 	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/chasmtest"
@@ -16,8 +20,10 @@ import (
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/metrics/metricstest"
 	"go.temporal.io/server/common/util"
 	"go.temporal.io/server/service/history/tasks"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -678,6 +684,728 @@ func TestProcessBufferTask_PausedDropsAutomatedKeepsManual(t *testing.T) {
 	})
 }
 
+type bufferProcessingComparisonInput struct {
+	name                 string
+	schedule             *schedulepb.Schedule
+	bufferedStarts       []*schedulespb.BufferedStart
+	cancelWorkflows      []*commonpb.WorkflowExecution
+	terminateWorkflows   []*commonpb.WorkflowExecution
+	lastProcessedTime    time.Time
+	workflowMigration    bool
+	initialConflictToken int64
+}
+
+type normalizedBufferProcessing struct {
+	schedulerState     *schedulerpb.SchedulerState
+	invokerState       *schedulerpb.InvokerState
+	tasks              []string
+	actionRequests     []string
+	metrics            []string
+	outcomes           []string
+	remainingDelta     int64
+	conflictTokenDelta int64
+}
+
+func TestProcessBufferTask_LegacyAndPlannerDifferentialCorpus(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	makeStart := func(id string, policy enumspb.ScheduleOverlapPolicy) *schedulespb.BufferedStart {
+		return &schedulespb.BufferedStart{
+			NominalTime:   timestamppb.New(now),
+			ActualTime:    timestamppb.New(now),
+			DesiredTime:   timestamppb.New(now),
+			RequestId:     id,
+			WorkflowId:    "workflow-" + id,
+			OverlapPolicy: policy,
+		}
+	}
+	running := makeStart("running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+	running.Attempt = 1
+	running.RunId = "running-run"
+
+	var corpus []bufferProcessingComparisonInput
+	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,
+	} {
+		corpus = append(corpus, bufferProcessingComparisonInput{
+			name:              "overlap " + policy.String(),
+			schedule:          defaultSchedule(),
+			bufferedStarts:    []*schedulespb.BufferedStart{running, makeStart("pending", policy)},
+			lastProcessedTime: now,
+		})
+	}
+
+	pausedSchedule := defaultSchedule()
+	pausedSchedule.State.Paused = true
+	pausedManual := makeStart("manual", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+	pausedManual.Manual = true
+	corpus = append(corpus, bufferProcessingComparisonInput{
+		name: "paused automatic and manual", schedule: pausedSchedule,
+		bufferedStarts: []*schedulespb.BufferedStart{
+			makeStart("automatic", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL), pausedManual,
+		}, lastProcessedTime: now,
+	})
+	for _, remaining := range []int64{0, 1, 3} {
+		limited := defaultSchedule()
+		limited.State.LimitedActions = true
+		limited.State.RemainingActions = remaining
+		corpus = append(corpus, bufferProcessingComparisonInput{
+			name: fmt.Sprintf("limited actions %d", remaining), schedule: limited,
+			bufferedStarts: []*schedulespb.BufferedStart{
+				makeStart("first", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL),
+				makeStart("second", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL),
+			}, lastProcessedTime: now, initialConflictToken: 17,
+		})
+	}
+	unlimitedManual := makeStart("manual", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+	unlimitedManual.Manual = true
+	corpus = append(corpus, bufferProcessingComparisonInput{
+		name: "unlimited and manual", schedule: defaultSchedule(),
+		bufferedStarts: []*schedulespb.BufferedStart{
+			makeStart("automatic", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL), unlimitedManual,
+		}, lastProcessedTime: now,
+	})
+	expired := makeStart("expired", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+	expired.ActualTime = timestamppb.New(now.Add(-2 * defaultCatchupWindow))
+	corpus = append(corpus, bufferProcessingComparisonInput{
+		name: "expired catchup window", schedule: defaultSchedule(),
+		bufferedStarts: []*schedulespb.BufferedStart{expired}, lastProcessedTime: now,
+	})
+	deferred := makeStart("deferred", enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE)
+	deferred.Attempt = -1
+	corpus = append(corpus, bufferProcessingComparisonInput{
+		name: "running and deferred", schedule: defaultSchedule(),
+		bufferedStarts: []*schedulespb.BufferedStart{running, deferred}, lastProcessedTime: now,
+	})
+	for _, backoffDelta := range []time.Duration{-time.Second, 0, time.Second} {
+		retry := makeStart(fmt.Sprintf("retry-%s", backoffDelta), enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+		retry.Attempt = 2
+		retry.BackoffTime = timestamppb.New(now.Add(backoffDelta))
+		corpus = append(corpus, bufferProcessingComparisonInput{
+			name: fmt.Sprintf("retry boundary %s", backoffDelta), schedule: defaultSchedule(),
+			bufferedStarts: []*schedulespb.BufferedStart{retry}, lastProcessedTime: now,
+		})
+	}
+	completed := makeStart("duplicate", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+	completed.Attempt = 1
+	completed.RunId = "completed-run"
+	completed.Completed = &schedulespb.CompletedResult{}
+	corpus = append(corpus, bufferProcessingComparisonInput{
+		name: "duplicate and completed entries", schedule: defaultSchedule(),
+		bufferedStarts:    []*schedulespb.BufferedStart{completed, proto.Clone(completed).(*schedulespb.BufferedStart)},
+		lastProcessedTime: now,
+	})
+	duplicatePending := makeStart("duplicate-pending", enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE)
+	corpus = append(corpus, bufferProcessingComparisonInput{
+		name: "duplicate pending entries", schedule: defaultSchedule(),
+		bufferedStarts: []*schedulespb.BufferedStart{
+			running,
+			duplicatePending,
+			proto.Clone(duplicatePending).(*schedulespb.BufferedStart),
+		},
+		lastProcessedTime: now,
+	})
+	corpus = append(corpus,
+		bufferProcessingComparisonInput{
+			name: "existing cancel work", schedule: defaultSchedule(),
+			bufferedStarts:    []*schedulespb.BufferedStart{running},
+			cancelWorkflows:   []*commonpb.WorkflowExecution{{WorkflowId: "cancel", RunId: "cancel-run"}},
+			lastProcessedTime: now,
+		},
+		bufferProcessingComparisonInput{
+			name: "existing terminate work", schedule: defaultSchedule(),
+			bufferedStarts:     []*schedulespb.BufferedStart{running},
+			terminateWorkflows: []*commonpb.WorkflowExecution{{WorkflowId: "terminate", RunId: "terminate-run"}},
+			lastProcessedTime:  now,
+		},
+		bufferProcessingComparisonInput{
+			name: "migration state", schedule: defaultSchedule(),
+			bufferedStarts:    []*schedulespb.BufferedStart{makeStart("migration", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)},
+			lastProcessedTime: now, workflowMigration: true,
+		},
+	)
+	completionRace := makeStart("completion-race", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+	completionRace.Attempt = 1
+	completionRace.Completed = &schedulespb.CompletedResult{}
+	corpus = append(corpus, bufferProcessingComparisonInput{
+		name: "completion before start result", schedule: defaultSchedule(),
+		bufferedStarts: []*schedulespb.BufferedStart{completionRace}, lastProcessedTime: now,
+	})
+
+	for _, input := range corpus {
+		t.Run(input.name, func(t *testing.T) {
+			compareBufferProcessing(t, input)
+		})
+	}
+}
+
+func TestProcessBufferTask_LegacyAndPlannerDifferentialOverlapStateSpace(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	policies := []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,
+	}
+	capacityCases := []struct {
+		name      string
+		limited   bool
+		remaining int64
+	}{
+		{name: "unlimited"},
+		{name: "exhausted", limited: true},
+		{name: "one", limited: true, remaining: 1},
+		{name: "two", limited: true, remaining: 2},
+		{name: "three", limited: true, remaining: 3},
+	}
+
+	comparisons := 0
+	for length := 0; length <= 4; length++ {
+		forEachOverlapPolicySequence(policies, length, func(sequence []enumspb.ScheduleOverlapPolicy) {
+			for _, running := range []bool{false, true} {
+				for _, capacity := range capacityCases {
+					schedule := defaultSchedule()
+					schedule.Policies.OverlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP
+					schedule.State.LimitedActions = capacity.limited
+					schedule.State.RemainingActions = capacity.remaining
+					starts := make([]*schedulespb.BufferedStart, 0, len(sequence)+1)
+					if running {
+						start := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+						start.Attempt = 1
+						start.RunId = "running-run"
+						starts = append(starts, start)
+					}
+					for index, policy := range sequence {
+						starts = append(starts, newBufferProcessingComparisonStart(now, fmt.Sprintf("pending-%d", index), policy))
+					}
+					compareBufferProcessing(t, bufferProcessingComparisonInput{
+						name: fmt.Sprintf(
+							"overlap state length=%d policies=%v running=%t capacity=%s",
+							length,
+							sequence,
+							running,
+							capacity.name,
+						),
+						schedule:             schedule,
+						bufferedStarts:       starts,
+						lastProcessedTime:    now,
+						initialConflictToken: 17,
+					})
+					comparisons++
+				}
+			}
+		})
+	}
+	require.Equal(t, 15550, comparisons)
+}
+
+func TestProcessBufferTask_LegacyAndPlannerDifferentialDuplicateIdentityStateSpace(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	policies := []enumspb.ScheduleOverlapPolicy{
+		enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,
+		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,
+	}
+	requestIDPatterns := []struct {
+		name string
+		id   func(int) string
+	}{
+		{name: "same", id: func(int) string { return "duplicate" }},
+		{name: "alternating", id: func(index int) string { return fmt.Sprintf("duplicate-%d", index%2) }},
+	}
+	comparisons := 0
+	for length := 1; length <= 3; length++ {
+		forEachOverlapPolicySequence(policies, length, func(sequence []enumspb.ScheduleOverlapPolicy) {
+			for _, requestIDs := range requestIDPatterns {
+				for _, running := range []bool{false, true} {
+					for remaining := int64(0); remaining <= 3; remaining++ {
+						schedule := defaultSchedule()
+						schedule.Policies.OverlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP
+						schedule.State.LimitedActions = true
+						schedule.State.RemainingActions = remaining
+						starts := make([]*schedulespb.BufferedStart, 0, len(sequence)+1)
+						if running {
+							start := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+							start.Attempt = 1
+							start.RunId = "running-run"
+							starts = append(starts, start)
+						}
+						for index, policy := range sequence {
+							starts = append(starts, newBufferProcessingComparisonStart(now, requestIDs.id(index), policy))
+						}
+						compareBufferProcessing(t, bufferProcessingComparisonInput{
+							name: fmt.Sprintf(
+								"duplicate identities pattern=%s policies=%v running=%t remaining=%d",
+								requestIDs.name,
+								sequence,
+								running,
+								remaining,
+							),
+							schedule:             schedule,
+							bufferedStarts:       starts,
+							lastProcessedTime:    now,
+							initialConflictToken: 17,
+						})
+						comparisons++
+					}
+				}
+			}
+		})
+	}
+	require.Equal(t, 6384, comparisons)
+}
+
+func TestProcessBufferTask_LegacyAndPlannerDifferentialDefaultPolicyStateSpace(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	policies := []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,
+	}
+	comparisons := 0
+	for _, defaultPolicy := range policies {
+		for unspecifiedIndex := range 3 {
+			forEachOverlapPolicySequence(policies, 2, func(surrounding []enumspb.ScheduleOverlapPolicy) {
+				sequence := make([]enumspb.ScheduleOverlapPolicy, 3)
+				sequence[unspecifiedIndex] = enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED
+				for index, policy := range surrounding {
+					if index >= unspecifiedIndex {
+						index++
+					}
+					sequence[index] = policy
+				}
+				for _, running := range []bool{false, true} {
+					for remaining := int64(0); remaining <= 3; remaining++ {
+						schedule := defaultSchedule()
+						schedule.Policies.OverlapPolicy = defaultPolicy
+						schedule.State.LimitedActions = true
+						schedule.State.RemainingActions = remaining
+						starts := make([]*schedulespb.BufferedStart, 0, len(sequence)+1)
+						if running {
+							start := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+							start.Attempt = 1
+							start.RunId = "running-run"
+							starts = append(starts, start)
+						}
+						for index, policy := range sequence {
+							starts = append(starts, newBufferProcessingComparisonStart(now, fmt.Sprintf("pending-%d", index), policy))
+						}
+						compareBufferProcessing(t, bufferProcessingComparisonInput{
+							name: fmt.Sprintf(
+								"default policy=%s policies=%v running=%t remaining=%d",
+								defaultPolicy,
+								sequence,
+								running,
+								remaining,
+							),
+							schedule:             schedule,
+							bufferedStarts:       starts,
+							lastProcessedTime:    now,
+							initialConflictToken: 17,
+						})
+						comparisons++
+					}
+				}
+			})
+		}
+	}
+	require.Equal(t, 5184, comparisons)
+}
+
+func TestProcessBufferTask_LegacyAndPlannerDifferentialTimeCapacityStateSpace(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	catchupWindows := []time.Duration{time.Second, 5 * time.Second, time.Hour}
+	timeRelations := []struct {
+		name   string
+		offset time.Duration
+	}{
+		{name: "before", offset: time.Nanosecond},
+		{name: "equal"},
+		{name: "after", offset: -time.Nanosecond},
+	}
+	capacityCases := []struct {
+		name      string
+		limited   bool
+		remaining int64
+	}{
+		{name: "unlimited"},
+		{name: "exhausted", limited: true},
+		{name: "one", limited: true, remaining: 1},
+		{name: "two", limited: true, remaining: 2},
+	}
+	policyCases := []struct {
+		name          string
+		policy        enumspb.ScheduleOverlapPolicy
+		defaultPolicy enumspb.ScheduleOverlapPolicy
+	}{
+		{name: "explicit allow all", policy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL},
+		{name: "default allow all", defaultPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL},
+	}
+
+	comparisons := 0
+	for _, policy := range policyCases {
+		for _, catchupWindow := range catchupWindows {
+			effectiveWindow := max(catchupWindow, 5*time.Second)
+			for _, deadlineRelation := range timeRelations {
+				actualTime := now.Add(-effectiveWindow + deadlineRelation.offset)
+				deadline := actualTime.Add(effectiveWindow)
+				for _, desiredRelation := range timeRelations {
+					desiredTime := deadline.Add(-desiredRelation.offset)
+					for _, manual := range []bool{false, true} {
+						for _, paused := range []bool{false, true} {
+							for _, running := range []bool{false, true} {
+								for _, capacity := range capacityCases {
+									schedule := defaultSchedule()
+									schedule.Policies.OverlapPolicy = policy.defaultPolicy
+									schedule.Policies.CatchupWindow = durationpb.New(catchupWindow)
+									schedule.State.Paused = paused
+									schedule.State.LimitedActions = capacity.limited
+									schedule.State.RemainingActions = capacity.remaining
+									start := newBufferProcessingComparisonStart(now, "pending", policy.policy)
+									start.ActualTime = timestamppb.New(actualTime)
+									start.DesiredTime = timestamppb.New(desiredTime)
+									start.Manual = manual
+									starts := []*schedulespb.BufferedStart{start}
+									if running {
+										runningStart := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)
+										runningStart.Attempt = 1
+										runningStart.RunId = "running-run"
+										starts = append([]*schedulespb.BufferedStart{runningStart}, starts...)
+									}
+									compareBufferProcessing(t, bufferProcessingComparisonInput{
+										name: fmt.Sprintf(
+											"time capacity policy=%s catchup=%s deadline=%s desired=%s manual=%t paused=%t running=%t capacity=%s",
+											policy.name,
+											catchupWindow,
+											deadlineRelation.name,
+											desiredRelation.name,
+											manual,
+											paused,
+											running,
+											capacity.name,
+										),
+										schedule:             schedule,
+										bufferedStarts:       starts,
+										lastProcessedTime:    now,
+										initialConflictToken: 17,
+									})
+									comparisons++
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+	require.Equal(t, 1728, comparisons)
+}
+
+func TestProcessBufferTask_LegacyAndPlannerDifferentialLifecycleStateSpace(t *testing.T) {
+	now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC)
+	policies := []enumspb.ScheduleOverlapPolicy{
+		enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,
+		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,
+	}
+	defaultPolicies := policies[1:]
+	lifecycleCases := []struct {
+		name      string
+		attempt   int64
+		runID     string
+		completed bool
+	}{
+		{name: "deferred", attempt: -1},
+		{name: "unprocessed"},
+		{name: "ready", attempt: 1},
+		{name: "retry", attempt: 2},
+		{name: "running", attempt: 1, runID: "run"},
+		{name: "completed", attempt: 1, runID: "run", completed: true},
+		{name: "completion before start", attempt: 1, completed: true},
+	}
+
+	comparisons := 0
+	for _, defaultPolicy := range defaultPolicies {
+		for _, policy := range policies {
+			for _, lifecycle := range lifecycleCases {
+				schedule := defaultSchedule()
+				schedule.Policies.OverlapPolicy = defaultPolicy
+				start := newBufferProcessingComparisonStart(now, "start", policy)
+				start.Attempt = lifecycle.attempt
+				start.RunId = lifecycle.runID
+				if lifecycle.completed {
+					start.Completed = &schedulespb.CompletedResult{}
+				}
+				compareBufferProcessing(t, bufferProcessingComparisonInput{
+					name: fmt.Sprintf(
+						"lifecycle=%s policy=%s default=%s",
+						lifecycle.name,
+						policy,
+						defaultPolicy,
+					),
+					schedule:          schedule,
+					bufferedStarts:    []*schedulespb.BufferedStart{start},
+					lastProcessedTime: now,
+				})
+				comparisons++
+			}
+		}
+	}
+	require.Equal(t, 294, comparisons)
+}
+
+func forEachOverlapPolicySequence(
+	policies []enumspb.ScheduleOverlapPolicy,
+	length int,
+	visit func([]enumspb.ScheduleOverlapPolicy),
+) {
+	sequence := make([]enumspb.ScheduleOverlapPolicy, length)
+	var generate func(int)
+	generate = func(index int) {
+		if index == len(sequence) {
+			visit(sequence)
+			return
+		}
+		for _, policy := range policies {
+			sequence[index] = policy
+			generate(index + 1)
+		}
+	}
+	generate(0)
+}
+
+func newBufferProcessingComparisonStart(
+	now time.Time,
+	requestID string,
+	policy enumspb.ScheduleOverlapPolicy,
+) *schedulespb.BufferedStart {
+	return &schedulespb.BufferedStart{
+		NominalTime:   timestamppb.New(now),
+		ActualTime:    timestamppb.New(now),
+		DesiredTime:   timestamppb.New(now),
+		RequestId:     requestID,
+		WorkflowId:    "workflow-" + requestID,
+		OverlapPolicy: policy,
+	}
+}
+
+func TestApplyBufferPlan_RevalidatesRequestBeforeMutation(t *testing.T) {
+	schedule := defaultSchedule()
+	schedule.State.LimitedActions = true
+	schedule.State.RemainingActions = 1
+	env := newSchedulerTestEngine(t, schedule)
+	now := env.timeSource.Now()
+	config := defaultConfig()
+	handler := scheduler.NewInvokerProcessBufferTaskHandler(scheduler.InvokerTaskHandlerOptions{
+		Config: config, MetricsHandler: metrics.NoopMetricsHandler, BaseLogger: env.logger,
+	})
+	var apply func(chasm.MutableContext, *scheduler.Scheduler, *scheduler.Invoker) int64
+	var initialConflictToken int64
+	require.NoError(t, env.updateScheduler(func(s *scheduler.Scheduler, ctx chasm.MutableContext) error {
+		invoker := s.Invoker.Get(ctx)
+		invoker.BufferedStarts = []*schedulespb.BufferedStart{{
+			NominalTime: timestamppb.New(now), ActualTime: timestamppb.New(now), DesiredTime: timestamppb.New(now),
+			RequestId: "request", WorkflowId: "workflow", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
+		}}
+		invoker.LastProcessedTime = timestamppb.New(now)
+		apply = handler.PlanBufferProcessingForTest(invoker, s, now)
+		initialConflictToken = s.ConflictToken
+		return nil
+	}))
+
+	var invalidatedDecisions int64
+	require.NoError(t, env.updateScheduler(func(s *scheduler.Scheduler, ctx chasm.MutableContext) error {
+		invoker := s.Invoker.Get(ctx)
+		invoker.BufferedStarts[0].Attempt = 1
+		invalidatedDecisions = apply(ctx, s, invoker)
+		return nil
+	}))
+
+	require.Equal(t, int64(1), invalidatedDecisions)
+	require.NoError(t, env.readScheduler(func(s *scheduler.Scheduler, ctx chasm.Context) error {
+		require.Equal(t, int64(1), s.Invoker.Get(ctx).BufferedStarts[0].GetAttempt())
+		require.Equal(t, int64(1), s.Schedule.State.GetRemainingActions())
+		require.Equal(t, initialConflictToken, s.ConflictToken)
+		return nil
+	}))
+}
+
+func compareBufferProcessing(t *testing.T, input bufferProcessingComparisonInput) {
+	t.Helper()
+	legacy := runBufferProcessing(t, input, false)
+	planned := runBufferProcessing(t, input, true)
+	require.True(t, proto.Equal(legacy.schedulerState, planned.schedulerState), "%s: Scheduler state mismatch\nlegacy: %v\nplanned: %v", input.name, legacy.schedulerState, planned.schedulerState)
+	require.True(t, proto.Equal(legacy.invokerState, planned.invokerState), "%s: Invoker state mismatch\nlegacy: %v\nplanned: %v", input.name, legacy.invokerState, planned.invokerState)
+	require.Equal(t, legacy.tasks, planned.tasks, input.name)
+	require.Equal(t, legacy.actionRequests, planned.actionRequests, input.name)
+	require.Equal(t, legacy.metrics, planned.metrics, input.name)
+	require.Equal(t, legacy.outcomes, planned.outcomes, input.name)
+	require.Equal(t, legacy.remainingDelta, planned.remainingDelta, input.name)
+	require.Equal(t, legacy.conflictTokenDelta, planned.conflictTokenDelta, input.name)
+}
+
+func runBufferProcessing(t *testing.T, input bufferProcessingComparisonInput, planner bool) normalizedBufferProcessing {
+	t.Helper()
+	env := newTestEnv(t)
+	env.TimeSource.Update(input.lastProcessedTime)
+	ctx := env.MutableContext()
+	env.Scheduler.Schedule = proto.Clone(input.schedule).(*schedulepb.Schedule)
+	env.Scheduler.Info.CreateTime = timestamppb.New(input.lastProcessedTime)
+	initialConflictToken := input.initialConflictToken
+	if initialConflictToken == 0 {
+		initialConflictToken = env.Scheduler.ConflictToken
+	}
+	env.Scheduler.ConflictToken = initialConflictToken
+	if input.workflowMigration {
+		env.Scheduler.WorkflowMigration = &schedulerpb.WorkflowMigrationState{}
+	}
+	invoker := env.Scheduler.Invoker.Get(ctx)
+	invoker.InvokerState = &schedulerpb.InvokerState{
+		BufferedStarts:     cloneBufferedStarts(input.bufferedStarts),
+		CancelWorkflows:    cloneWorkflowExecutions(input.cancelWorkflows),
+		TerminateWorkflows: cloneWorkflowExecutions(input.terminateWorkflows),
+		LastProcessedTime:  timestamppb.New(input.lastProcessedTime),
+	}
+	env.NodeBackend.TasksByCategory = nil
+
+	config := defaultConfig()
+	recorder := metricstest.NewCaptureHandler()
+	capture := recorder.StartCapture()
+	defer recorder.StopCapture(capture)
+	handler := scheduler.NewInvokerProcessBufferTaskHandler(scheduler.InvokerTaskHandlerOptions{
+		Config: config, MetricsHandler: recorder, BaseLogger: env.Logger,
+	})
+	initialRemaining := env.Scheduler.Schedule.State.GetRemainingActions()
+	if planner {
+		require.NoError(t, handler.Execute(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerProcessBufferTask{}))
+	} else {
+		require.NoError(t, handler.ExecuteProcessBufferLegacyForTest(ctx, invoker))
+	}
+	require.NoError(t, env.CloseTransaction())
+
+	return normalizedBufferProcessing{
+		schedulerState:     proto.Clone(env.Scheduler.SchedulerState).(*schedulerpb.SchedulerState),
+		invokerState:       proto.Clone(invoker.InvokerState).(*schedulerpb.InvokerState),
+		tasks:              normalizeTasks(env.NodeBackend.TasksByCategory),
+		actionRequests:     normalizeActionRequests(invoker),
+		metrics:            normalizeMetrics(capture.Snapshot()),
+		outcomes:           normalizeBufferOutcomes(input.bufferedStarts, invoker.GetBufferedStarts()),
+		remainingDelta:     env.Scheduler.Schedule.State.GetRemainingActions() - initialRemaining,
+		conflictTokenDelta: env.Scheduler.ConflictToken - initialConflictToken,
+	}
+}
+
+func cloneBufferedStarts(starts []*schedulespb.BufferedStart) []*schedulespb.BufferedStart {
+	cloned := make([]*schedulespb.BufferedStart, 0, len(starts))
+	for _, start := range starts {
+		cloned = append(cloned, proto.Clone(start).(*schedulespb.BufferedStart))
+	}
+	return cloned
+}
+
+func cloneWorkflowExecutions(executions []*commonpb.WorkflowExecution) []*commonpb.WorkflowExecution {
+	cloned := make([]*commonpb.WorkflowExecution, 0, len(executions))
+	for _, execution := range executions {
+		cloned = append(cloned, proto.Clone(execution).(*commonpb.WorkflowExecution))
+	}
+	return cloned
+}
+
+func normalizeTasks(tasksByCategory map[tasks.Category][]tasks.Task) []string {
+	var categories []tasks.Category
+	for category := range tasksByCategory {
+		categories = append(categories, category)
+	}
+	sort.Slice(categories, func(i, j int) bool { return categories[i].ID() < categories[j].ID() })
+	var normalized []string
+	for _, category := range categories {
+		for index, task := range tasksByCategory[category] {
+			identity := ""
+			if chasmTask, ok := task.(*tasks.ChasmTask); ok {
+				identity = fmt.Sprintf("path=%s,type=%d,data=%x", strings.Join(chasmTask.Info.GetPath(), "/"), chasmTask.Info.GetTypeId(), chasmTask.Info.GetData().GetData())
+			}
+			normalized = append(normalized, fmt.Sprintf("%d:%d:%s:%s:%s", category.ID(), index, task.GetType(), task.GetVisibilityTime().UTC().Format(time.RFC3339Nano), identity))
+		}
+	}
+	return normalized
+}
+
+func normalizeActionRequests(invoker *scheduler.Invoker) []string {
+	requests := make([]string, 0, len(invoker.GetTerminateWorkflows())+len(invoker.GetCancelWorkflows()))
+	for _, execution := range invoker.GetTerminateWorkflows() {
+		requests = append(requests, "terminate:"+execution.GetWorkflowId()+":"+execution.GetRunId())
+	}
+	for _, execution := range invoker.GetCancelWorkflows() {
+		requests = append(requests, "cancel:"+execution.GetWorkflowId()+":"+execution.GetRunId())
+	}
+	return requests
+}
+
+func normalizeMetrics(snapshot metricstest.CaptureSnapshot) []string {
+	var normalized []string
+	for name, recordings := range snapshot {
+		for _, recording := range recordings {
+			var tags []string
+			for key, value := range recording.Tags {
+				tags = append(tags, key+"="+value)
+			}
+			sort.Strings(tags)
+			normalized = append(normalized, fmt.Sprintf("%s:%v:%s", name, recording.Value, strings.Join(tags, ",")))
+		}
+	}
+	sort.Strings(normalized)
+	return normalized
+}
+
+func normalizeBufferOutcomes(before, after []*schedulespb.BufferedStart) []string {
+	afterByRequestID := make(map[string][]*schedulespb.BufferedStart)
+	for _, start := range after {
+		afterByRequestID[start.GetRequestId()] = append(afterByRequestID[start.GetRequestId()], start)
+	}
+	outcomes := make([]string, 0, len(before))
+	for _, start := range before {
+		matches := afterByRequestID[start.GetRequestId()]
+		if len(matches) == 0 {
+			outcomes = append(outcomes, start.GetRequestId()+":discard")
+			continue
+		}
+		current := matches[0]
+		afterByRequestID[start.GetRequestId()] = matches[1:]
+		switch {
+		case start.GetAttempt() == 0 && current.GetAttempt() > 0:
+			outcomes = append(outcomes, start.GetRequestId()+":execute")
+		case current.GetAttempt() == -1:
+			outcomes = append(outcomes, start.GetRequestId()+":defer")
+		case current.GetCompleted() != nil:
+			outcomes = append(outcomes, start.GetRequestId()+":completed")
+		case current.GetRunId() != "":
+			outcomes = append(outcomes, start.GetRequestId()+":running")
+		case current.GetAttempt() > 1:
+			outcomes = append(outcomes, start.GetRequestId()+":retry")
+		default:
+			outcomes = append(outcomes, start.GetRequestId()+":retain")
+		}
+	}
+	return outcomes
+}
+
 // A buffered start with an overlap policy to cancel other workflows is processed.
 func TestProcessBufferTask_NeedsCancel(t *testing.T) {
 	env := newTestEnv(t)
diff --git a/chasm/lib/scheduler/invoker_tasks.go b/chasm/lib/scheduler/invoker_tasks.go
index fae04d93329..a358cf50a94 100644
--- a/chasm/lib/scheduler/invoker_tasks.go
+++ b/chasm/lib/scheduler/invoker_tasks.go
@@ -22,9 +22,7 @@ import (
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/resource"
-	"go.temporal.io/server/common/util"
 	queueerrors "go.temporal.io/server/service/history/queues/errors"
-	legacyscheduler "go.temporal.io/server/service/worker/scheduler"
 	"go.uber.org/fx"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -460,18 +458,24 @@ func (h *InvokerProcessBufferTaskHandler) Execute(
 		return queueerrors.NewUnprocessableTaskError("schedules must have an Action set")
 	}
 
-	// Compute actions to take from the current buffer.
-	result := h.processBuffer(ctx, invoker, scheduler)
+	tweakables := h.config.Tweakables(scheduler.Namespace)
+	snapshot := newBufferProcessingSnapshot(invoker, scheduler, catchupWindow(scheduler, tweakables))
+	plan := schedulerinternal.PlanBufferProcessing(snapshot, ctx.Now(invoker))
+	result := applyBufferPlan(ctx, scheduler, invoker, plan).result
+
+	h.recordBufferProcessingMetrics(scheduler, result)
+	return nil
+}
 
-	// Update Scheduler metadata.
-	var totalMissedCatchup int64
-	for _, count := range result.missedCatchupByActionRunning {
-		totalMissedCatchup += count
+func (h *InvokerProcessBufferTaskHandler) recordBufferProcessingMetrics(
+	scheduler *Scheduler,
+	result processBufferResult,
+) {
+	metricsHandler := newTaggedMetricsHandler(h.metricsHandler, scheduler)
+	for _, reason := range result.bufferedStartDropReasons {
+		metricsHandler.Counter(metrics.ScheduleBufferedStartDropped.Name()).
+			Record(1, metrics.ReasonTag(reason))
 	}
-	scheduler.recordActionResult(&schedulerActionResult{
-		overlapSkipped:      result.overlapSkipped,
-		missedCatchupWindow: totalMissedCatchup,
-	})
 	for overlapPolicy, count := range result.overlapSkippedByPolicy {
 		newTaggedMetricsHandler(h.metricsHandler, scheduler).WithTags(
 			metrics.StringTag(metrics.ScheduleOverlapPolicyTag, overlapPolicy.String()),
@@ -483,104 +487,6 @@ func (h *InvokerProcessBufferTaskHandler) Execute(
 			metrics.StringTag(metrics.ScheduleActionRunningTag, fmt.Sprintf("%t", actionRunning)),
 		).Counter(metrics.ScheduleMissedCatchupWindow.Name()).Record(count)
 	}
-
-	// Update internal state and create new tasks.
-	invoker.recordProcessBufferResult(ctx, &result)
-
-	return nil
-}
-
-// processBuffer resolves the Invoker's buffered starts that haven't yet begun
-// execution. This is where the decision is made to drive execution to
-// completion, or skip/drop a start.
-func (h *InvokerProcessBufferTaskHandler) processBuffer(
-	ctx chasm.MutableContext,
-	invoker *Invoker,
-	scheduler *Scheduler,
-) (result processBufferResult) {
-	runningWorkflows := invoker.runningWorkflowExecutions()
-	isRunning := len(runningWorkflows) > 0
-	result.missedCatchupByActionRunning = make(map[bool]int64)
-
-	// Processing ignores starts that are already executing or backing off. An existing
-	// deferred BUFFER_ONE start still participates so it can reject later starts.
-	pendingBufferedStarts := util.FilterSlice(invoker.GetBufferedStarts(), func(start *schedulespb.BufferedStart) bool {
-		return start.Attempt == 0 ||
-			(start.Attempt == -1 && scheduler.resolveOverlapPolicy(start.GetOverlapPolicy()) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE)
-	})
-
-	// Resolve overlap policies and trim BufferedStarts that are skipped by policy.
-	action := legacyscheduler.ProcessBuffer(pendingBufferedStarts, isRunning, scheduler.resolveOverlapPolicy)
-
-	// ProcessBuffer will drop starts by omitting them from NewBuffer. Start with the
-	// diff between the input and NewBuffer, and add any executing starts.
-	keepStarts := make(map[string]struct{}) // request ID -> is present
-	for _, start := range action.NewBuffer {
-		keepStarts[start.GetRequestId()] = struct{}{}
-	}
-
-	// Combine all available starts.
-	readyStarts := action.OverlappingStarts
-	if action.NonOverlappingStart != nil {
-		readyStarts = append(readyStarts, action.NonOverlappingStart)
-	}
-
-	// Update result metrics.
-	result.overlapSkipped = action.OverlapSkipped
-	result.overlapSkippedByPolicy = action.OverlapSkippedByPolicy
-
-	// Add starting workflows to result, trim others. Catchup-window expiry is
-	// checked before consumeScheduledAction so that a start past its catchup
-	// window doesn't consume a LimitedActions slot.
-	droppedCounter := newTaggedMetricsHandler(h.metricsHandler, scheduler).
-		Counter(metrics.ScheduleBufferedStartDropped.Name())
-	for _, start := range readyStarts {
-		deadline := h.startWorkflowDeadline(ctx, scheduler, start)
-		if ctx.Now(invoker).After(deadline) {
-			// Action was buffered in time but expired before execution
-			// (e.g., due to overlap deferral, retries, or system delay).
-			// Only emit the metric if the schedule would have run this
-			// start -- skip paused or action-exhausted schedules.
-			if start.Manual || scheduler.canTakeScheduledAction() {
-				// Determine if a running action contributed: either one is still
-				// running, or the previous action's CloseTime (stored in DesiredTime)
-				// was already past this start's deadline.
-				// Note: if no prior action completed, DesiredTime is zero-valued,
-				// so After(deadline) is false, correctly yielding actionRunning=false.
-				actionRunning := isRunning ||
-					start.GetDesiredTime().AsTime().After(deadline)
-				result.missedCatchupByActionRunning[actionRunning]++
-			}
-			result.discardStarts = append(result.discardStarts, start)
-			droppedCounter.Record(1, metrics.ReasonTag(bufferedStartDroppedMissedCatchup))
-			continue
-		}
-
-		// Ensure we can take more actions. Manual actions are always allowed.
-		if !start.Manual && !scheduler.consumeScheduledAction() {
-			// Drop buffered automated actions while paused or out of actions.
-			result.discardStarts = append(result.discardStarts, start)
-			droppedCounter.Record(1, metrics.ReasonTag(bufferedStartDroppedPausedOrLimited))
-			continue
-		}
-
-		keepStarts[start.GetRequestId()] = struct{}{}
-		result.startWorkflows = append(result.startWorkflows, start)
-	}
-
-	result.discardStarts = util.FilterSlice(pendingBufferedStarts, func(start *schedulespb.BufferedStart) bool {
-		_, keep := keepStarts[start.GetRequestId()]
-		return !keep
-	})
-
-	// Terminate overrides cancel if both are requested.
-	if action.NeedTerminate {
-		result.terminateWorkflows = runningWorkflows
-	} else if action.NeedCancel {
-		result.cancelWorkflows = runningWorkflows
-	}
-
-	return
 }
 
 // applyBackoff advances start's attempt and BackoffTime based on err and the retry policy.