Add CallbackRequestID to workflow execution for use by CHASM Schedules (#9479)

## What changed?
When a workflow is reset, `ApplyWorkflowExecutionStartedEvent`
re-registers the start-event callbacks using the reset operation's
request ID. `HandleNexusCompletion` cannot find a matching
`BufferedStart` and discards the completion. The `resetRequestID` param
is removed from `WorkflowResetter.ResetWorkflow` and the original
request ID is used. `findStartRequestID` reads the original request ID
back from `WorkflowExecutionInfo.RequestIds` by finding the
`EVENT_TYPE_WORKFLOW_EXECUTION_STARTED` entry.

## Why?
CHASM scheduler relies on callback `request_id` to match WF completions
to originating `BufferedStart` entries. When it cannot be found the
scheduler is permanently stuck with the workflow marked as still
running.

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

-
`TestScheduledWorkflowDoubleReset_SchedulerSeesCompletion_{HSM,CHASM}Callbacks`:
create a schedule, trigger immediately, reset the workflow twice, signal
the completion to complete, poll `ListSchedules` until scheduler shows
`COMPLETED`.

## Potential risks
CHASM scheduler has not been enabled in production yet, the blast radius
should be minimal.

api/persistence/v1/executions.pb.go

@@ -1283,7 +1283,7 @@ type WorkflowExecutionState struct {
 	Status                        v11.WorkflowExecutionStatus `protobuf:"varint,4,opt,name=status,proto3,enum=temporal.api.enums.v1.WorkflowExecutionStatus" json:"status,omitempty"`
 	LastUpdateVersionedTransition *VersionedTransition        `protobuf:"bytes,5,opt,name=last_update_versioned_transition,json=lastUpdateVersionedTransition,proto3" json:"last_update_versioned_transition,omitempty"`
 	StartTime                     *timestamppb.Timestamp      `protobuf:"bytes,6,opt,name=start_time,json=startTime,proto3" json:"start_time,omitempty"`
-	// Request IDs that are attached to the workflow execution. It can the request ID that started
+	// Request IDs that are attached to the workflow execution. It can be the request ID that started
 	// the workflow execution or request IDs that were attached to an existing running workflow
 	// execution via StartWorkflowExecutionRequest.OnConflictOptions.
 	RequestIds    map[string]*RequestIDInfo `protobuf:"bytes,7,rep,name=request_ids,json=requestIds,proto3" json:"request_ids,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`

proto/internal/temporal/server/api/persistence/v1/executions.proto

@@ -342,7 +342,7 @@ message WorkflowExecutionState {
     temporal.api.enums.v1.WorkflowExecutionStatus status = 4;
     VersionedTransition last_update_versioned_transition = 5;
     google.protobuf.Timestamp start_time = 6;
-    // Request IDs that are attached to the workflow execution. It can the request ID that started
+    // Request IDs that are attached to the workflow execution. It can be the request ID that started
     // the workflow execution or request IDs that were attached to an existing running workflow
     // execution via StartWorkflowExecutionRequest.OnConflictOptions.
     map<string, RequestIDInfo> request_ids = 7;

service/history/api/reapplyevents/api.go

@@ -131,7 +131,6 @@ func Invoke(
 					baseRebuildLastEventVersion,
 					baseNextEventID,
 					resetRunID.String(),
-					uuid.New().String(),
 					baseWorkflow,
 					baseWorkflow,
 					ndc.EventsReapplicationResetWorkflowReason,

service/history/api/resetworkflow/api.go

@@ -163,7 +163,6 @@ func Invoke(
 		baseRebuildLastEventVersion,
 		baseNextEventID,
 		resetRunID,
-		request.GetRequestId(),
 		baseWorkflow,
 		ndc.NewWorkflow(
 			shardContext.GetClusterMetadata(),

service/history/history_engine_test.go

@@ -5409,7 +5409,7 @@ func (s *engineSuite) TestReapplyEvents_ResetWorkflow() {
 	s.mockEventsReapplier.EXPECT().ReapplyEvents(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
 	s.mockWorkflowResetter.EXPECT().ResetWorkflow(
 		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
-		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
+		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(),
 	).Return(nil)

service/history/ndc/conflict_resolver.go

@@ -5,7 +5,6 @@ package ndc
 import (
 	"context"
 
-	"github.com/google/uuid"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/log"
@@ -107,7 +106,7 @@ func (r *ConflictResolverImpl) getOrRebuildMutableStateByIndex(
 	// task.getVersion() > currentLastItem
 	// incoming replication task, after application, will become the current branch
 	// (because higher version wins), we need to Rebuild the mutable state for that
-	rebuiltMutableState, err := r.rebuild(ctx, branchIndex, uuid.NewString())
+	rebuiltMutableState, err := r.rebuild(ctx, branchIndex)
 	if err != nil {
 		return nil, false, err
 	}
@@ -117,7 +116,6 @@ func (r *ConflictResolverImpl) getOrRebuildMutableStateByIndex(
 func (r *ConflictResolverImpl) rebuild(
 	ctx context.Context,
 	branchIndex int32,
-	requestID string,
 ) (historyi.MutableState, error) {
 
 	versionHistories := r.mutableState.GetExecutionInfo().GetVersionHistories()
@@ -150,7 +148,7 @@ func (r *ConflictResolverImpl) rebuild(
 		util.Ptr(lastItem.GetVersion()),
 		workflowKey,
 		replayVersionHistory.GetBranchToken(),
-		requestID,
+		findStartRequestID(executionState),
 	)
 	if err != nil {
 		return nil, err

service/history/ndc/conflict_resolver_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	enumspb "go.temporal.io/api/enums/v1"
 	historyspb "go.temporal.io/server/api/history/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/definition"
@@ -116,6 +117,9 @@ func (s *conflictResolverSuite) TestRebuild() {
 	}).AnyTimes()
 	s.mockMutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{
 		RunId: s.runID,
+		RequestIds: map[string]*persistencespb.RequestIDInfo{
+			requestID: {EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED},
+		},
 	}).AnyTimes()
 	s.mockMutableState.EXPECT().GetHistorySize().Return(historySize).AnyTimes()
 	s.mockMutableState.EXPECT().GetExternalPayloadSize().Return(externalPayloadSize).AnyTimes()
@@ -159,7 +163,7 @@ func (s *conflictResolverSuite) TestRebuild() {
 	}, nil)
 
 	s.mockContext.EXPECT().Clear()
-	rebuiltMutableState, err := s.nDCConflictResolver.rebuild(ctx, 1, requestID)
+	rebuiltMutableState, err := s.nDCConflictResolver.rebuild(ctx, 1)
 	s.NoError(err)
 	s.NotNil(rebuiltMutableState)
 	s.Equal(int32(1), versionHistories.GetCurrentVersionHistoryIndex())

service/history/ndc/state_rebuilder.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
@@ -48,7 +49,6 @@ type (
 			baseLastEventVersion *int64,
 			targetWorkflowIdentifier definition.WorkflowKey,
 			targetBranchToken []byte,
-			requestID string,
 			currentMutableState *persistencespb.WorkflowMutableState,
 		) (historyi.MutableState, RebuildStats, error)
 	}
@@ -158,9 +158,9 @@ func (r *StateRebuilderImpl) RebuildWithCurrentMutableState(
 	baseLastEventVersion *int64,
 	targetWorkflowIdentifier definition.WorkflowKey,
 	targetBranchToken []byte,
-	requestID string,
 	currentMutableState *persistencespb.WorkflowMutableState,
 ) (historyi.MutableState, RebuildStats, error) {
+	// Use the original start request ID handlers can still correlate rebuilt callbacks to the correct BufferedStart entry.
 	rebuiltMutableState, lastTxnId, err := r.buildMutableStateFromEvent(
 		ctx,
 		now,
@@ -170,7 +170,7 @@ func (r *StateRebuilderImpl) RebuildWithCurrentMutableState(
 		baseLastEventVersion,
 		targetWorkflowIdentifier,
 		targetBranchToken,
-		requestID,
+		findStartRequestID(currentMutableState.GetExecutionState()),
 	)
 	if err != nil {
 		return nil, RebuildStats{}, err
@@ -398,3 +398,14 @@ func (r *StateRebuilderImpl) getPaginationFn(
 		return paginateItems, resp.NextPageToken, nil
 	}
 }
+
+// findStartRequestID returns the request ID associated with the WorkflowExecutionStarted
+// event from the RequestIds map, or the create request ID if not found.
+func findStartRequestID(executionState *persistencespb.WorkflowExecutionState) string {
+	for reqID, info := range executionState.GetRequestIds() {
+		if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
+			return reqID
+		}
+	}
+	return executionState.GetCreateRequestId()
+}

service/history/ndc/state_rebuilder_mock.go

@@ -61,9 +61,9 @@ func (mr *MockStateRebuilderMockRecorder) Rebuild(ctx, now, baseWorkflowIdentifi
 }
 
 // RebuildWithCurrentMutableState mocks base method.
-func (m *MockStateRebuilder) RebuildWithCurrentMutableState(ctx context.Context, now time.Time, baseWorkflowIdentifier definition.WorkflowKey, baseBranchToken []byte, baseLastEventID int64, baseLastEventVersion *int64, targetWorkflowIdentifier definition.WorkflowKey, targetBranchToken []byte, requestID string, currentMutableState *persistence.WorkflowMutableState) (interfaces.MutableState, RebuildStats, error) {
+func (m *MockStateRebuilder) RebuildWithCurrentMutableState(ctx context.Context, now time.Time, baseWorkflowIdentifier definition.WorkflowKey, baseBranchToken []byte, baseLastEventID int64, baseLastEventVersion *int64, targetWorkflowIdentifier definition.WorkflowKey, targetBranchToken []byte, currentMutableState *persistence.WorkflowMutableState) (interfaces.MutableState, RebuildStats, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "RebuildWithCurrentMutableState", ctx, now, baseWorkflowIdentifier, baseBranchToken, baseLastEventID, baseLastEventVersion, targetWorkflowIdentifier, targetBranchToken, requestID, currentMutableState)
+	ret := m.ctrl.Call(m, "RebuildWithCurrentMutableState", ctx, now, baseWorkflowIdentifier, baseBranchToken, baseLastEventID, baseLastEventVersion, targetWorkflowIdentifier, targetBranchToken, currentMutableState)
 	ret0, _ := ret[0].(interfaces.MutableState)
 	ret1, _ := ret[1].(RebuildStats)
 	ret2, _ := ret[2].(error)
@@ -71,7 +71,7 @@ func (m *MockStateRebuilder) RebuildWithCurrentMutableState(ctx context.Context,
 }
 
 // RebuildWithCurrentMutableState indicates an expected call of RebuildWithCurrentMutableState.
-func (mr *MockStateRebuilderMockRecorder) RebuildWithCurrentMutableState(ctx, now, baseWorkflowIdentifier, baseBranchToken, baseLastEventID, baseLastEventVersion, targetWorkflowIdentifier, targetBranchToken, requestID, currentMutableState any) *gomock.Call {
+func (mr *MockStateRebuilderMockRecorder) RebuildWithCurrentMutableState(ctx, now, baseWorkflowIdentifier, baseBranchToken, baseLastEventID, baseLastEventVersion, targetWorkflowIdentifier, targetBranchToken, currentMutableState any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RebuildWithCurrentMutableState", reflect.TypeOf((*MockStateRebuilder)(nil).RebuildWithCurrentMutableState), ctx, now, baseWorkflowIdentifier, baseBranchToken, baseLastEventID, baseLastEventVersion, targetWorkflowIdentifier, targetBranchToken, requestID, currentMutableState)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RebuildWithCurrentMutableState", reflect.TypeOf((*MockStateRebuilder)(nil).RebuildWithCurrentMutableState), ctx, now, baseWorkflowIdentifier, baseBranchToken, baseLastEventID, baseLastEventVersion, targetWorkflowIdentifier, targetBranchToken, currentMutableState)
 }

service/history/ndc/state_rebuilder_test.go

@@ -373,7 +373,7 @@ func (s *stateRebuilderSuite) TestRebuild() {
 }
 
 func (s *stateRebuilderSuite) TestRebuildWithCurrentMutableState() {
-	requestID := uuid.NewString()
+	startRequestID := uuid.NewString()
 	version := int64(12)
 	lastEventID := int64(2)
 	branchToken := []byte("other random branch token")
@@ -460,6 +460,11 @@ func (s *stateRebuilderSuite) TestRebuildWithCurrentMutableState() {
 
 	s.mockTaskRefresher.EXPECT().Refresh(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
 	currentMutableState := &persistencespb.WorkflowMutableState{
+		ExecutionState: &persistencespb.WorkflowExecutionState{
+			RequestIds: map[string]*persistencespb.RequestIDInfo{
+				startRequestID: {EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED},
+			},
+		},
 		ExecutionInfo: &persistencespb.WorkflowExecutionInfo{
 			TransitionHistory: []*persistencespb.VersionedTransition{
 				{
@@ -479,7 +484,6 @@ func (s *stateRebuilderSuite) TestRebuildWithCurrentMutableState() {
 		util.Ptr(version),
 		definition.NewWorkflowKey(targetNamespaceID.String(), targetWorkflowID, targetRunID),
 		targetBranchToken,
-		requestID,
 		currentMutableState,
 	)
 	s.NoError(err)
@@ -498,4 +502,5 @@ func (s *stateRebuilderSuite) TestRebuildWithCurrentMutableState() {
 	s.Equal(timestamp.TimeValue(rebuildMutableState.GetExecutionState().StartTime), s.now)
 	s.Equal(expectedLastFirstTransactionID, rebuildExecutionInfo.LastFirstEventTxnId)
 	s.Equal(int64(11), rebuildExecutionInfo.TransitionHistory[0].TransitionCount)
+	s.Equal(startRequestID, rebuildMutableState.GetExecutionState().CreateRequestId)
 }

service/history/ndc/transaction_manager.go

@@ -329,7 +329,6 @@ func (r *transactionMgrImpl) backfillWorkflowEventsReapply(
 			baseRebuildLastEventVersion,
 			baseNextEventID,
 			resetRunID,
-			uuid.NewString(),
 			targetWorkflow,
 			targetWorkflow,
 			EventsReapplicationResetWorkflowReason,

service/history/ndc/transaction_manager_test.go

@@ -225,7 +225,6 @@ func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Active_Closed
 		lastWorkflowTaskStartedVersion,
 		nextEventID,
 		gomock.Any(),
-		gomock.Any(),
 		targetWorkflow,
 		targetWorkflow,
 		EventsReapplicationResetWorkflowReason,
@@ -309,7 +308,6 @@ func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_ResetF
 		lastWorkflowTaskStartedVersion,
 		nextEventID,
 		gomock.Any(),
-		gomock.Any(),
 		targetWorkflow,
 		targetWorkflow,
 		EventsReapplicationResetWorkflowReason,

service/history/ndc/workflow_resetter.go

@@ -59,7 +59,6 @@ type (
 			baseRebuildLastEventVersion int64,
 			baseNextEventID int64,
 			resetRunID string,
-			resetRequestID string,
 			baseWorkflow Workflow,
 			currentWorkflow Workflow,
 			resetReason string,
@@ -115,7 +114,6 @@ func (r *workflowResetterImpl) ResetWorkflow(
 	baseRebuildLastEventVersion int64,
 	baseNextEventID int64,
 	resetRunID string,
-	resetRequestID string,
 	baseWorkflow Workflow,
 	currentWorkflow Workflow,
 	resetReason string,
@@ -203,6 +201,14 @@ func (r *workflowResetterImpl) ResetWorkflow(
 		}
 	}
 
+	// Use the original start request ID from the base run so callbacks on the
+	// reset workflow are associated with the original start request.
+	// The run ID provides uniqueness per execution, so the start request ID can
+	// be used consistently across resets.
+	//
+	// Read from the base run's RequestIds map; fall back to CreateRequestId otherwise.
+	startRequestID := findStartRequestID(baseWorkflow.GetMutableState().GetExecutionState())
+
 	resetWorkflow, err := r.prepareResetWorkflow(
 		ctx,
 		namespaceID,
@@ -212,7 +218,7 @@ func (r *workflowResetterImpl) ResetWorkflow(
 		baseRebuildLastEventID,
 		baseRebuildLastEventVersion,
 		resetRunID,
-		resetRequestID,
+		startRequestID,
 		resetWorkflowVersion,
 		resetReason,
 		allowResetWithPendingChildren,
@@ -263,7 +269,7 @@ func (r *workflowResetterImpl) prepareResetWorkflow(
 	baseRebuildLastEventID int64,
 	baseRebuildLastEventVersion int64,
 	resetRunID string,
-	resetRequestID string,
+	requestID string,
 	resetWorkflowVersion int64,
 	resetReason string,
 	allowResetWithPendingChildren bool,
@@ -278,7 +284,7 @@ func (r *workflowResetterImpl) prepareResetWorkflow(
 		baseRebuildLastEventID,
 		baseRebuildLastEventVersion,
 		resetRunID,
-		resetRequestID,
+		requestID,
 	)
 	if err != nil {
 		return nil, err

service/history/ndc/workflow_resetter_mock.go

@@ -45,15 +45,15 @@ func (m *MockWorkflowResetter) EXPECT() *MockWorkflowResetterMockRecorder {
 }
 
 // ResetWorkflow mocks base method.
-func (m *MockWorkflowResetter) ResetWorkflow(ctx context.Context, namespaceID namespace.ID, workflowID, baseRunID string, baseBranchToken []byte, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID int64, resetRunID, resetRequestID string, baseWorkflow, currentWorkflow Workflow, resetReason string, additionalReapplyEvents []*history.HistoryEvent, resetReapplyExcludeTypes map[enums.ResetReapplyExcludeType]struct{}, allowResetWithPendingChildren bool, postResetOperations []*workflow.PostResetOperation) error {
+func (m *MockWorkflowResetter) ResetWorkflow(ctx context.Context, namespaceID namespace.ID, workflowID, baseRunID string, baseBranchToken []byte, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID int64, resetRunID string, baseWorkflow, currentWorkflow Workflow, resetReason string, additionalReapplyEvents []*history.HistoryEvent, resetReapplyExcludeTypes map[enums.ResetReapplyExcludeType]struct{}, allowResetWithPendingChildren bool, postResetOperations []*workflow.PostResetOperation) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ResetWorkflow", ctx, namespaceID, workflowID, baseRunID, baseBranchToken, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID, resetRunID, resetRequestID, baseWorkflow, currentWorkflow, resetReason, additionalReapplyEvents, resetReapplyExcludeTypes, allowResetWithPendingChildren, postResetOperations)
+	ret := m.ctrl.Call(m, "ResetWorkflow", ctx, namespaceID, workflowID, baseRunID, baseBranchToken, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID, resetRunID, baseWorkflow, currentWorkflow, resetReason, additionalReapplyEvents, resetReapplyExcludeTypes, allowResetWithPendingChildren, postResetOperations)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // ResetWorkflow indicates an expected call of ResetWorkflow.
-func (mr *MockWorkflowResetterMockRecorder) ResetWorkflow(ctx, namespaceID, workflowID, baseRunID, baseBranchToken, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID, resetRunID, resetRequestID, baseWorkflow, currentWorkflow, resetReason, additionalReapplyEvents, resetReapplyExcludeTypes, allowResetWithPendingChildren, postResetOperations any) *gomock.Call {
+func (mr *MockWorkflowResetterMockRecorder) ResetWorkflow(ctx, namespaceID, workflowID, baseRunID, baseBranchToken, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID, resetRunID, baseWorkflow, currentWorkflow, resetReason, additionalReapplyEvents, resetReapplyExcludeTypes, allowResetWithPendingChildren, postResetOperations any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResetWorkflow", reflect.TypeOf((*MockWorkflowResetter)(nil).ResetWorkflow), ctx, namespaceID, workflowID, baseRunID, baseBranchToken, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID, resetRunID, resetRequestID, baseWorkflow, currentWorkflow, resetReason, additionalReapplyEvents, resetReapplyExcludeTypes, allowResetWithPendingChildren, postResetOperations)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResetWorkflow", reflect.TypeOf((*MockWorkflowResetter)(nil).ResetWorkflow), ctx, namespaceID, workflowID, baseRunID, baseBranchToken, baseRebuildLastEventID, baseRebuildLastEventVersion, baseNextEventID, resetRunID, baseWorkflow, currentWorkflow, resetReason, additionalReapplyEvents, resetReapplyExcludeTypes, allowResetWithPendingChildren, postResetOperations)
 }

service/history/transfer_queue_active_task_executor.go

@@ -1739,7 +1739,6 @@ func (t *transferQueueActiveTaskExecutor) resetWorkflow(
 		baseRebuildLastEventVersion,
 		baseNextEventID,
 		resetRunID,
-		uuid.NewString(),
 		baseWorkflow,
 		ndc.NewWorkflow(
 			t.shardContext.GetClusterMetadata(),

service/history/workflow_rebuilder.go

@@ -28,7 +28,6 @@ type (
 		branchToken          []byte
 		stateTransitionCount int64
 		dbRecordVersion      int64
-		requestID            string
 		mutableState         *persistencespb.WorkflowMutableState
 	}
 	workflowRebuilder interface {
@@ -96,7 +95,6 @@ func (r *workflowRebuilderImpl) rebuild(
 		rebuildSpec.branchToken,
 		rebuildSpec.stateTransitionCount,
 		rebuildSpec.dbRecordVersion,
-		rebuildSpec.requestID,
 		rebuildSpec.mutableState,
 	)
 	if err != nil {
@@ -196,7 +194,6 @@ func (r *workflowRebuilderImpl) getRebuildSpecFromMutableState(
 		branchToken:          currentVersionHistory.BranchToken,
 		stateTransitionCount: mutableState.ExecutionInfo.StateTransitionCount,
 		dbRecordVersion:      resp.DBRecordVersion,
-		requestID:            mutableState.ExecutionState.CreateRequestId,
 		mutableState:         resp.State,
 	}, nil
 }
@@ -207,7 +204,6 @@ func (r *workflowRebuilderImpl) replayResetWorkflow(
 	branchToken []byte,
 	stateTransitionCount int64,
 	dbRecordVersion int64,
-	requestID string,
 	mutableState *persistencespb.WorkflowMutableState,
 ) (historyi.MutableState, error) {
 	rebuildMutableState, rebuildStats, err := ndc.NewStateRebuilder(r.shard, r.logger).RebuildWithCurrentMutableState(
@@ -219,7 +215,6 @@ func (r *workflowRebuilderImpl) replayResetWorkflow(
 		nil,             // skip event ID & version check
 		workflowKey,
 		branchToken,
-		requestID,
 		mutableState,
 	)
 	if err != nil {

tests/schedule_test.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http/httptest"
 	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/nexus-rpc/sdk-go/nexus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
@@ -26,11 +28,13 @@ import (
 	"go.temporal.io/sdk/workflow"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/headers"
+	"go.temporal.io/server/common/nexus/nexusrpc"
 	"go.temporal.io/server/common/payload"
 	"go.temporal.io/server/common/payloads"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/searchattribute/sadefs"
 	"go.temporal.io/server/common/testing/protorequire"
+	"go.temporal.io/server/components/callbacks"
 	"go.temporal.io/server/service/worker/scheduler"
 	"go.temporal.io/server/tests/testcore"
 	"google.golang.org/grpc/metadata"
@@ -1387,6 +1391,466 @@ func (s *scheduleFunctionalSuiteBase) cleanup(sid string) {
 	})
 }
 
+// TestScheduledWorkflowDoubleReset_SchedulerSeesCompletion_HSMCallbacks verifies that
+// the CHASM scheduler correctly processes a completion after the workflow is reset twice,
+// using the HSM callback implementation.
+func (s *ScheduleCHASMFunctionalSuite) TestScheduledWorkflowDoubleReset_SchedulerSeesCompletion_HSMCallbacks() {
+	s.OverrideDynamicConfig(dynamicconfig.EnableCHASMCallbacks, false)
+	s.runScheduledWorkflowDoubleResetSchedulerSeesCompletion(
+		"sched-test-double-reset-hsm-cb",
+		"sched-test-double-reset-hsm-cb-wf",
+		"sched-test-double-reset-hsm-cb-wt",
+	)
+}
+
+// TestScheduledWorkflowDoubleReset_SchedulerSeesCompletion_ChasmCallbacks verifies the
+// same double-reset fix using the CHASM callback implementation (EnableCHASMCallbacks = true).
+func (s *ScheduleCHASMFunctionalSuite) TestScheduledWorkflowDoubleReset_SchedulerSeesCompletion_ChasmCallbacks() {
+	s.OverrideDynamicConfig(dynamicconfig.EnableCHASMCallbacks, true)
+	s.runScheduledWorkflowDoubleResetSchedulerSeesCompletion(
+		"sched-test-double-reset-chasm-cb",
+		"sched-test-double-reset-chasm-cb-wf",
+		"sched-test-double-reset-chasm-cb-wt",
+	)
+}
+
+// runScheduledWorkflowDoubleResetSchedulerSeesCompletion verifies that the callback
+// request_id is preserved across two chained resets. After a double reset, the
+// workflow is completed via signal; the scheduler must still see the completion.
+//
+// Without the fix, the second reset would use the first reset run's CreateRequestId as
+// the callback request_id, which doesn't match the original BufferedStart.RequestId.
+// The fix passes the original start request ID as the effective requestID to
+// applyEvents during rebuild, so AttachRequestID stores it in RequestIds with
+// EventType=STARTED; findStartRequestID then retrieves it correctly across chained resets.
+func (s *ScheduleCHASMFunctionalSuite) runScheduledWorkflowDoubleResetSchedulerSeesCompletion(sid, wid, wt string) {
+	// A workflow that blocks until it receives a "complete" signal. This lets us reset
+	// it multiple times before allowing it to finish.
+	s.worker.RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
+		ch := workflow.GetSignalChannel(ctx, "complete")
+		var signal any
+		ch.Receive(ctx, &signal)
+		return nil
+	}, workflow.RegisterOptions{Name: wt})
+
+	ctx := s.newContext()
+
+	_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
+		Namespace:  s.Namespace().String(),
+		ScheduleId: sid,
+		Schedule: &schedulepb.Schedule{
+			Spec: &schedulepb.ScheduleSpec{
+				Interval: []*schedulepb.IntervalSpec{
+					{Interval: durationpb.New(24 * time.Hour)},
+				},
+			},
+			Action: &schedulepb.ScheduleAction{
+				Action: &schedulepb.ScheduleAction_StartWorkflow{
+					StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
+						WorkflowId:   wid,
+						WorkflowType: &commonpb.WorkflowType{Name: wt},
+						TaskQueue:    &taskqueuepb.TaskQueue{Name: s.taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
+					},
+				},
+			},
+		},
+		InitialPatch: &schedulepb.SchedulePatch{
+			TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{},
+		},
+		RequestId: uuid.NewString(),
+	})
+	s.NoError(err)
+	s.cleanup(sid)
+
+	// Wait for scheduler to start the workflow and show it as RUNNING.
+	listEntry := s.getScheduleEntryFomVisibility(sid, func(ent *schedulepb.ScheduleListEntry) bool {
+		return len(ent.Info.RecentActions) >= 1 &&
+			ent.Info.RecentActions[0].GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING
+	})
+	a1 := listEntry.Info.RecentActions[0]
+	wfExec := &commonpb.WorkflowExecution{
+		WorkflowId: a1.StartWorkflowResult.WorkflowId,
+		RunId:      a1.StartWorkflowResult.RunId,
+	}
+
+	// Wait for the base run to complete a workflow task (valid reset point).
+	s.WaitForHistoryEvents(`
+		1 WorkflowExecutionStarted
+		2 WorkflowTaskScheduled
+		3 WorkflowTaskStarted
+		4 WorkflowTaskCompleted`,
+		s.GetHistoryFunc(s.Namespace().String(), wfExec),
+		5*time.Second,
+		10*time.Millisecond,
+	)
+
+	// Capture the original run's start request ID before any reset.
+	// We assert it is preserved through both resets.
+	origDesc, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+		Namespace: s.Namespace().String(),
+		Execution: wfExec,
+	})
+	s.NoError(err)
+	var originalStartReqID string
+	for reqID, info := range origDesc.GetWorkflowExtendedInfo().GetRequestIdInfos() {
+		if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
+			originalStartReqID = reqID
+			break
+		}
+	}
+	s.NotEmpty(originalStartReqID, "original run must have a request ID for WorkflowExecutionStarted")
+
+	// First reset: base run → reset run 1.
+	// We reset with WorkflowTaskFinishEventId: 3 (WorkflowTaskStarted). resetRun1 is
+	// created by replaying history up to and including event 3 from the base run.
+	resp1, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
+		Namespace:                 s.Namespace().String(),
+		WorkflowExecution:         wfExec,
+		Reason:                    "double-reset-test-first",
+		WorkflowTaskFinishEventId: 3,
+		RequestId:                 uuid.NewString(),
+	})
+	s.NoError(err)
+	resetRun1 := &commonpb.WorkflowExecution{
+		WorkflowId: wfExec.WorkflowId,
+		RunId:      resp1.RunId,
+	}
+
+	// Verify the original start request ID is preserved on reset run 1.
+	s.EventuallyWithT(func(col *assert.CollectT) {
+		resetDesc, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: s.Namespace().String(),
+			Execution: resetRun1,
+		})
+		require.NoError(col, err)
+		var resetStartReqID string
+		for reqID, info := range resetDesc.GetWorkflowExtendedInfo().GetRequestIdInfos() {
+			if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
+				resetStartReqID = reqID
+				break
+			}
+		}
+		require.Equal(col, originalStartReqID, resetStartReqID,
+			"start request ID must be preserved across first reset")
+	}, 10*time.Second, 100*time.Millisecond)
+
+	// Second reset: reset run 1 → reset run 2.
+	// resetRun1 already has event 3 (WorkflowTaskStarted) from the replay performed by
+	// the first reset, so WorkflowTaskFinishEventId: 3 is valid immediately. We do not
+	// need to wait for resetRun1 to complete a new workflow task before resetting it.
+	resp2, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
+		Namespace:                 s.Namespace().String(),
+		WorkflowExecution:         resetRun1,
+		Reason:                    "double-reset-test-second",
+		WorkflowTaskFinishEventId: 3,
+		RequestId:                 uuid.NewString(),
+	})
+	s.NoError(err)
+	resetRun2 := &commonpb.WorkflowExecution{
+		WorkflowId: wfExec.WorkflowId,
+		RunId:      resp2.RunId,
+	}
+
+	// Verify the original start request ID is preserved on reset run 2.
+	// Without the fix, the second reset would use resetRun1's CreateRequestId,
+	// breaking the callback → BufferedStart matching in the CHASM scheduler.
+	s.EventuallyWithT(func(col *assert.CollectT) {
+		resetDesc, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: s.Namespace().String(),
+			Execution: resetRun2,
+		})
+		require.NoError(col, err)
+		var resetStartReqID string
+		for reqID, info := range resetDesc.GetWorkflowExtendedInfo().GetRequestIdInfos() {
+			if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
+				resetStartReqID = reqID
+				break
+			}
+		}
+		require.Equal(col, originalStartReqID, resetStartReqID,
+			"start request ID must be preserved across double reset")
+	}, 10*time.Second, 100*time.Millisecond)
+
+	// Signal the latest run (reset run 2) to complete.
+	// Sending without a RunId targets the current/latest run.
+	_, err = s.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
+		Namespace: s.Namespace().String(),
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: wfExec.WorkflowId,
+		},
+		SignalName: "complete",
+	})
+	s.NoError(err)
+
+	// Poll until the scheduler shows the original action as COMPLETED.
+	// Without the chained-reset fix, the second reset run's callback would carry
+	// reset1RequestId instead of originalStartRequestId, so HandleNexusCompletion
+	// would fail to match and silently drop the completion.
+	s.getScheduleEntryFomVisibility(sid, func(ent *schedulepb.ScheduleListEntry) bool {
+		for _, action := range ent.Info.RecentActions {
+			if action.GetStartWorkflowResult().GetRunId() == wfExec.RunId {
+				return action.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+			}
+		}
+		return false
+	})
+}
+
+// TestScheduledWorkflow_ResetWithAdditionalCallback_{HSM,CHASM}Callbacks verifies that after:
+//  1. A workflow is started via a schedule trigger (the schedule attaches a callback with
+//     the original start request ID),
+//  2. A second callback is manually attached to the running workflow (different request ID),
+//  3. The workflow is reset,
+//
+// ...both callbacks are preserved in the reset run with their original/distinct request IDs,
+// the schedule correctly records the completion (proving the original request ID survived the
+// reset), and the second Nexus callback is also delivered to its HTTP endpoint.
+func (s *ScheduleCHASMFunctionalSuite) TestScheduledWorkflow_ResetWithAdditionalCallback_HSMCallbacks() {
+	s.OverrideDynamicConfig(dynamicconfig.EnableCHASMCallbacks, false)
+	s.runScheduledWorkflowResetWithAdditionalCallback(
+		"sched-test-reset-extra-cb-hsm",
+		"sched-test-reset-extra-cb-hsm-wf",
+		"sched-test-reset-extra-cb-hsm-wt",
+	)
+}
+
+// TestScheduledWorkflow_ResetWithAdditionalCallback_ChasmCallbacks verifies the
+// same reset-with-additional-callback fix using the CHASM callback implementation (EnableCHASMCallbacks = true).
+func (s *ScheduleCHASMFunctionalSuite) TestScheduledWorkflow_ResetWithAdditionalCallback_ChasmCallbacks() {
+	s.OverrideDynamicConfig(dynamicconfig.EnableCHASMCallbacks, true)
+	s.runScheduledWorkflowResetWithAdditionalCallback(
+		"sched-test-reset-extra-cb-chasm",
+		"sched-test-reset-extra-cb-chasm-wf",
+		"sched-test-reset-extra-cb-chasm-wt",
+	)
+}
+
+//  1. A schedule fires and starts a workflow, attaching the schedule's callback with
+//     request_id = original_start_request_id.
+//  2. A second Nexus callback is attached to the running workflow via a separate
+//     StartWorkflowExecution request (request_id = attachRequestId ≠ original_start_request_id).
+//  3. The workflow is reset. The WorkflowExecutionOptionsUpdated event (carrying the second
+//     callback) is reapplied to the reset run.
+//  4. After reset, DescribeWorkflowExecution shows 2 callbacks. WorkflowExtendedInfo confirms
+//     their request IDs are distinct: one tied to the start event, one to the options-update.
+//  5. The workflow is completed via signal.
+//  6. The schedule records the completion (original_start_request_id still matches the
+//     scheduler's BufferedStart), and the second callback is delivered to the HTTP endpoint.
+func (s *ScheduleCHASMFunctionalSuite) runScheduledWorkflowResetWithAdditionalCallback(sid, wid, wt string) {
+	// Allow the manually-attached callback URL (httptest uses plain HTTP on 127.0.0.1).
+	s.OverrideDynamicConfig(
+		callbacks.AllowedAddresses,
+		[]any{map[string]any{"Pattern": "*", "AllowInsecure": true}},
+	)
+
+	// HTTP completion server for the second (manually-attached) callback.
+	ch := &completionHandler{
+		requestCh:         make(chan *nexusrpc.CompletionRequest, 1),
+		requestCompleteCh: make(chan error, 1),
+	}
+	defer func() {
+		close(ch.requestCh)
+		close(ch.requestCompleteCh)
+	}()
+	secondCallbackURL := func() string {
+		hh := nexusrpc.NewCompletionHTTPHandler(nexusrpc.CompletionHandlerOptions{Handler: ch})
+		srv := httptest.NewServer(hh)
+		s.T().Cleanup(func() { srv.Close() })
+		return srv.URL + "/callback"
+	}()
+
+	// Workflow blocks on a "complete" signal so we can reset it before it finishes.
+	s.worker.RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
+		sigCh := workflow.GetSignalChannel(ctx, "complete")
+		var signal any
+		sigCh.Receive(ctx, &signal)
+		return nil
+	}, workflow.RegisterOptions{Name: wt})
+
+	ctx := s.newContext()
+
+	// create the schedule and trigger it immediately.
+	_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
+		Namespace:  s.Namespace().String(),
+		ScheduleId: sid,
+		Schedule: &schedulepb.Schedule{
+			Spec: &schedulepb.ScheduleSpec{
+				Interval: []*schedulepb.IntervalSpec{
+					{Interval: durationpb.New(24 * time.Hour)},
+				},
+			},
+			Action: &schedulepb.ScheduleAction{
+				Action: &schedulepb.ScheduleAction_StartWorkflow{
+					StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
+						WorkflowId:   wid,
+						WorkflowType: &commonpb.WorkflowType{Name: wt},
+						TaskQueue:    &taskqueuepb.TaskQueue{Name: s.taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
+					},
+				},
+			},
+		},
+		InitialPatch: &schedulepb.SchedulePatch{
+			TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{},
+		},
+		RequestId: uuid.NewString(),
+	})
+	s.NoError(err)
+	s.cleanup(sid)
+
+	// Wait for the workflow to appear as RUNNING in the schedule's visibility.
+	listEntry := s.getScheduleEntryFomVisibility(sid, func(ent *schedulepb.ScheduleListEntry) bool {
+		return len(ent.Info.RecentActions) >= 1 &&
+			ent.Info.RecentActions[0].GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING
+	})
+	a1 := listEntry.Info.RecentActions[0]
+	wfExec := &commonpb.WorkflowExecution{
+		WorkflowId: a1.StartWorkflowResult.WorkflowId,
+		RunId:      a1.StartWorkflowResult.RunId,
+	}
+
+	// Wait for the base run to complete a workflow task — this is our reset point.
+	s.WaitForHistoryEvents(`
+		1 WorkflowExecutionStarted
+		2 WorkflowTaskScheduled
+		3 WorkflowTaskStarted
+		4 WorkflowTaskCompleted`,
+		s.GetHistoryFunc(s.Namespace().String(), wfExec),
+		5*time.Second,
+		10*time.Millisecond,
+	)
+
+	// attach a second Nexus callback to the running workflow.
+	// The USE_EXISTING conflict policy + AttachCompletionCallbacks makes the server
+	// reuse the existing run and append the callback, generating a
+	// WorkflowExecutionOptionsUpdated event with request_id = attachRequestId.
+	attachRequestID := uuid.NewString()
+	attachResp, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:                attachRequestID,
+		Namespace:                s.Namespace().String(),
+		WorkflowId:               wfExec.WorkflowId, // The scheduler appends a time suffix to wid
+		WorkflowType:             &commonpb.WorkflowType{Name: wt},
+		TaskQueue:                &taskqueuepb.TaskQueue{Name: s.taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
+		WorkflowIdConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
+		OnConflictOptions: &workflowpb.OnConflictOptions{
+			AttachRequestId:           true,
+			AttachCompletionCallbacks: true,
+		},
+		CompletionCallbacks: []*commonpb.Callback{
+			{
+				Variant: &commonpb.Callback_Nexus_{
+					Nexus: &commonpb.Callback_Nexus{
+						Url: secondCallbackURL,
+					},
+				},
+			},
+		},
+	})
+	s.NoError(err)
+	s.False(attachResp.Started, "expected to attach to existing run, not start a new one")
+
+	// Wait for the WorkflowExecutionOptionsUpdated event to be persisted so that it
+	// will be reapplied by the reset.
+	s.WaitForHistoryEvents(`
+		1 WorkflowExecutionStarted
+		2 WorkflowTaskScheduled
+		3 WorkflowTaskStarted
+		4 WorkflowTaskCompleted
+		5 WorkflowExecutionOptionsUpdated`,
+		s.GetHistoryFunc(s.Namespace().String(), wfExec),
+		5*time.Second,
+		10*time.Millisecond,
+	)
+
+	// reset the workflow to event 3 (WorkflowTaskStarted). The resetter
+	// reapplies events after the reset point, so WorkflowExecutionOptionsUpdated
+	// (event 5 in the base run) is reapplied to the reset run.
+	resetResp, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
+		Namespace:                 s.Namespace().String(),
+		WorkflowExecution:         wfExec,
+		Reason:                    "reset-with-additional-callback-test",
+		WorkflowTaskFinishEventId: 3,
+		RequestId:                 uuid.NewString(),
+	})
+	s.NoError(err)
+	resetRun := &commonpb.WorkflowExecution{
+		WorkflowId: wfExec.WorkflowId,
+		RunId:      resetResp.RunId,
+	}
+
+	// verify the reset run has 2 callbacks with distinct request IDs.
+	//
+	// The schedule's callback has request_id = original_start_request_id (stored in
+	// RequestIds under that key with EventType=STARTED, propagated by Rebuild via the
+	// effectiveRequestID mechanism). It appears in WorkflowExtendedInfo.RequestIdInfos
+	// as the key whose EventType == WORKFLOW_EXECUTION_STARTED.
+	//
+	// The manually-attached callback has request_id = attachRequestId, which appears
+	// in RequestIdInfos as the key whose EventType ==
+	// WORKFLOW_EXECUTION_OPTIONS_UPDATED.
+	var startRequestID string
+	s.EventuallyWithT(func(col *assert.CollectT) {
+		descResp, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+			Namespace: s.Namespace().String(),
+			Execution: resetRun,
+		})
+		require.NoError(col, err)
+
+		// Both callbacks must be present.
+		require.Len(col, descResp.Callbacks, 2)
+
+		// Find request IDs from WorkflowExtendedInfo.
+		reqIDs := descResp.GetWorkflowExtendedInfo().GetRequestIdInfos()
+
+		// attachRequestID must map to WORKFLOW_EXECUTION_OPTIONS_UPDATED.
+		attachInfo, ok := reqIDs[attachRequestID]
+		require.True(col, ok, "attachRequestId not found in RequestIdInfos")
+		require.Equal(col, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED, attachInfo.GetEventType())
+
+		// There must also be a request ID for the original start event.
+		for reqID, info := range reqIDs {
+			if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
+				startRequestID = reqID
+				break
+			}
+		}
+		require.NotEmpty(col, startRequestID, "no request ID found for WorkflowExecutionStarted")
+		require.NotEqual(col, startRequestID, attachRequestID,
+			"schedule callback and manually-attached callback must have different request IDs")
+	}, 10*time.Second, 100*time.Millisecond)
+
+	// signal the reset run to complete.
+	// Sending without a RunId targets the latest (reset) run.
+	_, err = s.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
+		Namespace: s.Namespace().String(),
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: wfExec.WorkflowId,
+		},
+		SignalName: "complete",
+	})
+	s.NoError(err)
+
+	// the schedule must record the original action as COMPLETED, proving that
+	// the schedule's callback (with original_start_request_id) survived the reset and
+	// correctly matched the scheduler's BufferedStart.
+	s.getScheduleEntryFomVisibility(sid, func(ent *schedulepb.ScheduleListEntry) bool {
+		for _, action := range ent.Info.RecentActions {
+			if action.GetStartWorkflowResult().GetRunId() == wfExec.RunId {
+				return action.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+			}
+		}
+		return false
+	})
+
+	// the second (manually-attached) callback must be delivered to the HTTP server.
+	select {
+	case completion := <-ch.requestCh:
+		s.Equal(nexus.OperationStateSucceeded, completion.State)
+		ch.requestCompleteCh <- nil // acknowledge so the server handler can return
+	case <-time.After(10 * time.Second):
+		s.Fail("timeout waiting for second callback to be delivered")
+	}
+}
+
 // TestCreateScheduleAlreadyExists verifies that creating a schedule with the same ID
 // returns an AlreadyExists serviceerror.
 func (s *ScheduleCHASMFunctionalSuite) TestCreateScheduleAlreadyExists() {