Nexus CHASM async completion (2/2) (#9972)

## What changed? Support completion-before-start. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s)
2026-08-30 18:41:49 -07:00 · 2026-04-23 08:14:57 -07:00
parent 562d26fc83
commit ff3ed0b16f
13 changed files with 497 additions and 212 deletions
--- a/api/historyservice/v1/request_response.pb.go
+++ b/api/historyservice/v1/request_response.pb.go
@@ -8594,7 +8594,12 @@ type CompleteNexusOperationChasmRequest struct {
 	// Time when the operation was closed.
 	CloseTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=close_time,json=closeTime,proto3" json:"close_time,omitempty"`
 	// Links from the Nexus completion callback (e.g. references to the handler workflow).
-	Links         []*v14.Link `protobuf:"bytes,5,rep,name=links,proto3" json:"links,omitempty"`
+	Links []*v14.Link `protobuf:"bytes,5,rep,name=links,proto3" json:"links,omitempty"`
+	// Async operation token from the callback request, used to synthesize a started event when
+	// completion arrives before the worker's start response.
+	OperationToken string `protobuf:"bytes,6,opt,name=operation_token,json=operationToken,proto3" json:"operation_token,omitempty"`
+	// Time when the operation was started. Used when completion is received before the started response.
+	StartTime     *timestamppb.Timestamp `protobuf:"bytes,7,opt,name=start_time,json=startTime,proto3" json:"start_time,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -8675,6 +8680,20 @@ func (x *CompleteNexusOperationChasmRequest) GetLinks() []*v14.Link {
 	return nil
 }

+func (x *CompleteNexusOperationChasmRequest) GetOperationToken() string {
+	if x != nil {
+		return x.OperationToken
+	}
+	return ""
+}
+
+func (x *CompleteNexusOperationChasmRequest) GetStartTime() *timestamppb.Timestamp {
+	if x != nil {
+		return x.StartTime
+	}
+	return nil
+}
+
 type isCompleteNexusOperationChasmRequest_Outcome interface {
 	isCompleteNexusOperationChasmRequest_Outcome()
 }
@@ -11076,7 +11095,7 @@ const file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc
 	"\x10ListTasksRequest\x12V\n" +
 	"\arequest\x18\x01 \x01(\v2<.temporal.server.api.adminservice.v1.ListHistoryTasksRequestR\arequest:\x16\x92\xc4\x03\x12\x1a\x10request.shard_id\"n\n" +
 	"\x11ListTasksResponse\x12Y\n" +
-	"\bresponse\x18\x01 \x01(\v2=.temporal.server.api.adminservice.v1.ListHistoryTasksResponseR\bresponse\"\x91\x03\n" +
+	"\bresponse\x18\x01 \x01(\v2=.temporal.server.api.adminservice.v1.ListHistoryTasksResponseR\bresponse\"\xf5\x03\n" +
 	"\"CompleteNexusOperationChasmRequest\x12V\n" +
 	"\n" +
 	"completion\x18\x01 \x01(\v26.temporal.server.api.token.v1.NexusOperationCompletionR\n" +
@@ -11085,7 +11104,10 @@ const file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc
 	"\afailure\x18\x03 \x01(\v2 .temporal.api.failure.v1.FailureH\x00R\afailure\x129\n" +
 	"\n" +
 	"close_time\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tcloseTime\x122\n" +
-	"\x05links\x18\x05 \x03(\v2\x1c.temporal.api.common.v1.LinkR\x05links:\x1e\x92\xc4\x03\x1aB\x18completion.component_refB\t\n" +
+	"\x05links\x18\x05 \x03(\v2\x1c.temporal.api.common.v1.LinkR\x05links\x12'\n" +
+	"\x0foperation_token\x18\x06 \x01(\tR\x0eoperationToken\x129\n" +
+	"\n" +
+	"start_time\x18\a \x01(\v2\x1a.google.protobuf.TimestampR\tstartTime:\x1e\x92\xc4\x03\x1aB\x18completion.component_refB\t\n" +
 	"\aoutcome\"%\n" +
 	"#CompleteNexusOperationChasmResponse\"\xe0\x03\n" +
 	"\x1dCompleteNexusOperationRequest\x12V\n" +
@@ -11696,48 +11718,49 @@ var file_temporal_server_api_historyservice_v1_request_response_proto_depIdxs =
 	173, // 221: temporal.server.api.historyservice.v1.CompleteNexusOperationChasmRequest.failure:type_name -> temporal.api.failure.v1.Failure
 	171, // 222: temporal.server.api.historyservice.v1.CompleteNexusOperationChasmRequest.close_time:type_name -> google.protobuf.Timestamp
 	185, // 223: temporal.server.api.historyservice.v1.CompleteNexusOperationChasmRequest.links:type_name -> temporal.api.common.v1.Link
-	261, // 224: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.completion:type_name -> temporal.server.api.token.v1.NexusOperationCompletion
-	262, // 225: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.success:type_name -> temporal.api.common.v1.Payload
-	263, // 226: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.failure:type_name -> temporal.api.nexus.v1.Failure
-	171, // 227: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.start_time:type_name -> google.protobuf.Timestamp
-	185, // 228: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.links:type_name -> temporal.api.common.v1.Link
-	264, // 229: temporal.server.api.historyservice.v1.InvokeStateMachineMethodRequest.ref:type_name -> temporal.server.api.persistence.v1.StateMachineRef
-	265, // 230: temporal.server.api.historyservice.v1.DeepHealthCheckResponse.state:type_name -> temporal.server.api.enums.v1.HealthState
-	266, // 231: temporal.server.api.historyservice.v1.DeepHealthCheckResponse.checks:type_name -> temporal.server.api.health.v1.HealthCheck
-	186, // 232: temporal.server.api.historyservice.v1.SyncWorkflowStateRequest.execution:type_name -> temporal.api.common.v1.WorkflowExecution
-	188, // 233: temporal.server.api.historyservice.v1.SyncWorkflowStateRequest.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition
-	192, // 234: temporal.server.api.historyservice.v1.SyncWorkflowStateRequest.version_histories:type_name -> temporal.server.api.history.v1.VersionHistories
-	267, // 235: temporal.server.api.historyservice.v1.SyncWorkflowStateResponse.versioned_transition_artifact:type_name -> temporal.server.api.replication.v1.VersionedTransitionArtifact
-	268, // 236: temporal.server.api.historyservice.v1.UpdateActivityOptionsRequest.update_request:type_name -> temporal.api.workflowservice.v1.UpdateActivityOptionsRequest
-	269, // 237: temporal.server.api.historyservice.v1.UpdateActivityOptionsResponse.activity_options:type_name -> temporal.api.activity.v1.ActivityOptions
-	270, // 238: temporal.server.api.historyservice.v1.PauseActivityRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.PauseActivityRequest
-	271, // 239: temporal.server.api.historyservice.v1.UnpauseActivityRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.UnpauseActivityRequest
-	272, // 240: temporal.server.api.historyservice.v1.ResetActivityRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.ResetActivityRequest
-	273, // 241: temporal.server.api.historyservice.v1.UpdateWorkflowExecutionOptionsRequest.update_request:type_name -> temporal.api.workflowservice.v1.UpdateWorkflowExecutionOptionsRequest
-	274, // 242: temporal.server.api.historyservice.v1.UpdateWorkflowExecutionOptionsResponse.workflow_execution_options:type_name -> temporal.api.workflow.v1.WorkflowExecutionOptions
-	275, // 243: temporal.server.api.historyservice.v1.PauseWorkflowExecutionRequest.pause_request:type_name -> temporal.api.workflowservice.v1.PauseWorkflowExecutionRequest
-	276, // 244: temporal.server.api.historyservice.v1.UnpauseWorkflowExecutionRequest.unpause_request:type_name -> temporal.api.workflowservice.v1.UnpauseWorkflowExecutionRequest
-	277, // 245: temporal.server.api.historyservice.v1.StartNexusOperationRequest.request:type_name -> temporal.api.nexus.v1.StartOperationRequest
-	278, // 246: temporal.server.api.historyservice.v1.StartNexusOperationResponse.response:type_name -> temporal.api.nexus.v1.StartOperationResponse
-	279, // 247: temporal.server.api.historyservice.v1.CancelNexusOperationRequest.request:type_name -> temporal.api.nexus.v1.CancelOperationRequest
-	280, // 248: temporal.server.api.historyservice.v1.CancelNexusOperationResponse.response:type_name -> temporal.api.nexus.v1.CancelOperationResponse
-	1,   // 249: temporal.server.api.historyservice.v1.ExecuteMultiOperationRequest.Operation.start_workflow:type_name -> temporal.server.api.historyservice.v1.StartWorkflowExecutionRequest
-	105, // 250: temporal.server.api.historyservice.v1.ExecuteMultiOperationRequest.Operation.update_workflow:type_name -> temporal.server.api.historyservice.v1.UpdateWorkflowExecutionRequest
-	2,   // 251: temporal.server.api.historyservice.v1.ExecuteMultiOperationResponse.Response.start_workflow:type_name -> temporal.server.api.historyservice.v1.StartWorkflowExecutionResponse
-	106, // 252: temporal.server.api.historyservice.v1.ExecuteMultiOperationResponse.Response.update_workflow:type_name -> temporal.server.api.historyservice.v1.UpdateWorkflowExecutionResponse
-	281, // 253: temporal.server.api.historyservice.v1.RecordWorkflowTaskStartedResponse.QueriesEntry.value:type_name -> temporal.api.query.v1.WorkflowQuery
-	281, // 254: temporal.server.api.historyservice.v1.RecordWorkflowTaskStartedResponseWithRawHistory.QueriesEntry.value:type_name -> temporal.api.query.v1.WorkflowQuery
-	282, // 255: temporal.server.api.historyservice.v1.GetReplicationMessagesResponse.ShardMessagesEntry.value:type_name -> temporal.server.api.replication.v1.ReplicationMessages
-	98,  // 256: temporal.server.api.historyservice.v1.ShardReplicationStatus.RemoteClustersEntry.value:type_name -> temporal.server.api.historyservice.v1.ShardReplicationStatusPerCluster
-	97,  // 257: temporal.server.api.historyservice.v1.ShardReplicationStatus.HandoverNamespacesEntry.value:type_name -> temporal.server.api.historyservice.v1.HandoverNamespaceInfo
-	226, // 258: temporal.server.api.historyservice.v1.AddTasksRequest.Task.blob:type_name -> temporal.api.common.v1.DataBlob
-	283, // 259: temporal.server.api.historyservice.v1.routing:extendee -> google.protobuf.MessageOptions
-	0,   // 260: temporal.server.api.historyservice.v1.routing:type_name -> temporal.server.api.historyservice.v1.RoutingOptions
-	261, // [261:261] is the sub-list for method output_type
-	261, // [261:261] is the sub-list for method input_type
-	260, // [260:261] is the sub-list for extension type_name
-	259, // [259:260] is the sub-list for extension extendee
-	0,   // [0:259] is the sub-list for field type_name
+	171, // 224: temporal.server.api.historyservice.v1.CompleteNexusOperationChasmRequest.start_time:type_name -> google.protobuf.Timestamp
+	261, // 225: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.completion:type_name -> temporal.server.api.token.v1.NexusOperationCompletion
+	262, // 226: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.success:type_name -> temporal.api.common.v1.Payload
+	263, // 227: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.failure:type_name -> temporal.api.nexus.v1.Failure
+	171, // 228: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.start_time:type_name -> google.protobuf.Timestamp
+	185, // 229: temporal.server.api.historyservice.v1.CompleteNexusOperationRequest.links:type_name -> temporal.api.common.v1.Link
+	264, // 230: temporal.server.api.historyservice.v1.InvokeStateMachineMethodRequest.ref:type_name -> temporal.server.api.persistence.v1.StateMachineRef
+	265, // 231: temporal.server.api.historyservice.v1.DeepHealthCheckResponse.state:type_name -> temporal.server.api.enums.v1.HealthState
+	266, // 232: temporal.server.api.historyservice.v1.DeepHealthCheckResponse.checks:type_name -> temporal.server.api.health.v1.HealthCheck
+	186, // 233: temporal.server.api.historyservice.v1.SyncWorkflowStateRequest.execution:type_name -> temporal.api.common.v1.WorkflowExecution
+	188, // 234: temporal.server.api.historyservice.v1.SyncWorkflowStateRequest.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition
+	192, // 235: temporal.server.api.historyservice.v1.SyncWorkflowStateRequest.version_histories:type_name -> temporal.server.api.history.v1.VersionHistories
+	267, // 236: temporal.server.api.historyservice.v1.SyncWorkflowStateResponse.versioned_transition_artifact:type_name -> temporal.server.api.replication.v1.VersionedTransitionArtifact
+	268, // 237: temporal.server.api.historyservice.v1.UpdateActivityOptionsRequest.update_request:type_name -> temporal.api.workflowservice.v1.UpdateActivityOptionsRequest
+	269, // 238: temporal.server.api.historyservice.v1.UpdateActivityOptionsResponse.activity_options:type_name -> temporal.api.activity.v1.ActivityOptions
+	270, // 239: temporal.server.api.historyservice.v1.PauseActivityRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.PauseActivityRequest
+	271, // 240: temporal.server.api.historyservice.v1.UnpauseActivityRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.UnpauseActivityRequest
+	272, // 241: temporal.server.api.historyservice.v1.ResetActivityRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.ResetActivityRequest
+	273, // 242: temporal.server.api.historyservice.v1.UpdateWorkflowExecutionOptionsRequest.update_request:type_name -> temporal.api.workflowservice.v1.UpdateWorkflowExecutionOptionsRequest
+	274, // 243: temporal.server.api.historyservice.v1.UpdateWorkflowExecutionOptionsResponse.workflow_execution_options:type_name -> temporal.api.workflow.v1.WorkflowExecutionOptions
+	275, // 244: temporal.server.api.historyservice.v1.PauseWorkflowExecutionRequest.pause_request:type_name -> temporal.api.workflowservice.v1.PauseWorkflowExecutionRequest
+	276, // 245: temporal.server.api.historyservice.v1.UnpauseWorkflowExecutionRequest.unpause_request:type_name -> temporal.api.workflowservice.v1.UnpauseWorkflowExecutionRequest
+	277, // 246: temporal.server.api.historyservice.v1.StartNexusOperationRequest.request:type_name -> temporal.api.nexus.v1.StartOperationRequest
+	278, // 247: temporal.server.api.historyservice.v1.StartNexusOperationResponse.response:type_name -> temporal.api.nexus.v1.StartOperationResponse
+	279, // 248: temporal.server.api.historyservice.v1.CancelNexusOperationRequest.request:type_name -> temporal.api.nexus.v1.CancelOperationRequest
+	280, // 249: temporal.server.api.historyservice.v1.CancelNexusOperationResponse.response:type_name -> temporal.api.nexus.v1.CancelOperationResponse
+	1,   // 250: temporal.server.api.historyservice.v1.ExecuteMultiOperationRequest.Operation.start_workflow:type_name -> temporal.server.api.historyservice.v1.StartWorkflowExecutionRequest
+	105, // 251: temporal.server.api.historyservice.v1.ExecuteMultiOperationRequest.Operation.update_workflow:type_name -> temporal.server.api.historyservice.v1.UpdateWorkflowExecutionRequest
+	2,   // 252: temporal.server.api.historyservice.v1.ExecuteMultiOperationResponse.Response.start_workflow:type_name -> temporal.server.api.historyservice.v1.StartWorkflowExecutionResponse
+	106, // 253: temporal.server.api.historyservice.v1.ExecuteMultiOperationResponse.Response.update_workflow:type_name -> temporal.server.api.historyservice.v1.UpdateWorkflowExecutionResponse
+	281, // 254: temporal.server.api.historyservice.v1.RecordWorkflowTaskStartedResponse.QueriesEntry.value:type_name -> temporal.api.query.v1.WorkflowQuery
+	281, // 255: temporal.server.api.historyservice.v1.RecordWorkflowTaskStartedResponseWithRawHistory.QueriesEntry.value:type_name -> temporal.api.query.v1.WorkflowQuery
+	282, // 256: temporal.server.api.historyservice.v1.GetReplicationMessagesResponse.ShardMessagesEntry.value:type_name -> temporal.server.api.replication.v1.ReplicationMessages
+	98,  // 257: temporal.server.api.historyservice.v1.ShardReplicationStatus.RemoteClustersEntry.value:type_name -> temporal.server.api.historyservice.v1.ShardReplicationStatusPerCluster
+	97,  // 258: temporal.server.api.historyservice.v1.ShardReplicationStatus.HandoverNamespacesEntry.value:type_name -> temporal.server.api.historyservice.v1.HandoverNamespaceInfo
+	226, // 259: temporal.server.api.historyservice.v1.AddTasksRequest.Task.blob:type_name -> temporal.api.common.v1.DataBlob
+	283, // 260: temporal.server.api.historyservice.v1.routing:extendee -> google.protobuf.MessageOptions
+	0,   // 261: temporal.server.api.historyservice.v1.routing:type_name -> temporal.server.api.historyservice.v1.RoutingOptions
+	262, // [262:262] is the sub-list for method output_type
+	262, // [262:262] is the sub-list for method input_type
+	261, // [261:262] is the sub-list for extension type_name
+	260, // [260:261] is the sub-list for extension extendee
+	0,   // [0:260] is the sub-list for field type_name
 }

 func init() { file_temporal_server_api_historyservice_v1_request_response_proto_init() }
--- a/api/persistence/v1/chasm.pb.go
+++ b/api/persistence/v1/chasm.pb.go
@@ -608,7 +608,11 @@ type ChasmNexusCompletion struct {
 	// Allows completing a started operation after a workflow has been reset.
 	RequestId string `protobuf:"bytes,4,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
 	// Links from the Nexus completion callback (e.g. references to the handler workflow).
-	Links         []*v1.Link `protobuf:"bytes,5,rep,name=links,proto3" json:"links,omitempty"`
+	Links []*v1.Link `protobuf:"bytes,5,rep,name=links,proto3" json:"links,omitempty"`
+	// Async operation token from the callback request, used for completion-before-start.
+	OperationToken string `protobuf:"bytes,6,opt,name=operation_token,json=operationToken,proto3" json:"operation_token,omitempty"`
+	// Start time from the callback request, used for completion-before-start.
+	StartTime     *timestamppb.Timestamp `protobuf:"bytes,7,opt,name=start_time,json=startTime,proto3" json:"start_time,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -689,6 +693,20 @@ func (x *ChasmNexusCompletion) GetLinks() []*v1.Link {
 	return nil
 }

+func (x *ChasmNexusCompletion) GetOperationToken() string {
+	if x != nil {
+		return x.OperationToken
+	}
+	return ""
+}
+
+func (x *ChasmNexusCompletion) GetStartTime() *timestamppb.Timestamp {
+	if x != nil {
+		return x.StartTime
+	}
+	return nil
+}
+
 type isChasmNexusCompletion_Outcome interface {
 	isChasmNexusCompletion_Outcome()
 }
@@ -858,7 +876,7 @@ const file_temporal_server_api_persistence_v1_chasm_proto_rawDesc = "" +
 	"\farchetype_id\x18\x04 \x01(\rR\varchetypeId\x12}\n" +
 	"\x1eexecution_versioned_transition\x18\x05 \x01(\v27.temporal.server.api.persistence.v1.VersionedTransitionR\x1cexecutionVersionedTransition\x12%\n" +
 	"\x0ecomponent_path\x18\x06 \x03(\tR\rcomponentPath\x12\x8c\x01\n" +
-	"&component_initial_versioned_transition\x18\a \x01(\v27.temporal.server.api.persistence.v1.VersionedTransitionR#componentInitialVersionedTransition\"\xaa\x02\n" +
+	"&component_initial_versioned_transition\x18\a \x01(\v27.temporal.server.api.persistence.v1.VersionedTransitionR#componentInitialVersionedTransition\"\x8e\x03\n" +
 	"\x14ChasmNexusCompletion\x12;\n" +
 	"\asuccess\x18\x01 \x01(\v2\x1f.temporal.api.common.v1.PayloadH\x00R\asuccess\x12<\n" +
 	"\afailure\x18\x02 \x01(\v2 .temporal.api.failure.v1.FailureH\x00R\afailure\x129\n" +
@@ -866,7 +884,10 @@ const file_temporal_server_api_persistence_v1_chasm_proto_rawDesc = "" +
 	"close_time\x18\x03 \x01(\v2\x1a.google.protobuf.TimestampR\tcloseTime\x12\x1d\n" +
 	"\n" +
 	"request_id\x18\x04 \x01(\tR\trequestId\x122\n" +
-	"\x05links\x18\x05 \x03(\v2\x1c.temporal.api.common.v1.LinkR\x05linksB\t\n" +
+	"\x05links\x18\x05 \x03(\v2\x1c.temporal.api.common.v1.LinkR\x05links\x12'\n" +
+	"\x0foperation_token\x18\x06 \x01(\tR\x0eoperationToken\x129\n" +
+	"\n" +
+	"start_time\x18\a \x01(\v2\x1a.google.protobuf.TimestampR\tstartTimeB\t\n" +
 	"\aoutcomeB6Z4go.temporal.io/server/api/persistence/v1;persistenceb\x06proto3"

 var (
@@ -920,14 +941,15 @@ var file_temporal_server_api_persistence_v1_chasm_proto_depIdxs = []int32{
 	13, // 16: temporal.server.api.persistence.v1.ChasmNexusCompletion.failure:type_name -> temporal.api.failure.v1.Failure
 	14, // 17: temporal.server.api.persistence.v1.ChasmNexusCompletion.close_time:type_name -> google.protobuf.Timestamp
 	15, // 18: temporal.server.api.persistence.v1.ChasmNexusCompletion.links:type_name -> temporal.api.common.v1.Link
-	14, // 19: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.scheduled_time:type_name -> google.protobuf.Timestamp
-	10, // 20: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.data:type_name -> temporal.api.common.v1.DataBlob
-	11, // 21: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition
-	22, // [22:22] is the sub-list for method output_type
-	22, // [22:22] is the sub-list for method input_type
-	22, // [22:22] is the sub-list for extension type_name
-	22, // [22:22] is the sub-list for extension extendee
-	0,  // [0:22] is the sub-list for field type_name
+	14, // 19: temporal.server.api.persistence.v1.ChasmNexusCompletion.start_time:type_name -> google.protobuf.Timestamp
+	14, // 20: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.scheduled_time:type_name -> google.protobuf.Timestamp
+	10, // 21: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.data:type_name -> temporal.api.common.v1.DataBlob
+	11, // 22: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition
+	23, // [23:23] is the sub-list for method output_type
+	23, // [23:23] is the sub-list for method input_type
+	23, // [23:23] is the sub-list for extension type_name
+	23, // [23:23] is the sub-list for extension extendee
+	0,  // [0:23] is the sub-list for field type_name
 }

 func init() { file_temporal_server_api_persistence_v1_chasm_proto_init() }
--- a/chasm/lib/nexusoperation/operation.go
+++ b/chasm/lib/nexusoperation/operation.go
@@ -11,6 +11,7 @@ import (
 	"go.temporal.io/server/chasm"
 	nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
 	"go.temporal.io/server/common/backoff"
+	"go.temporal.io/server/common/primitives/timestamp"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -40,7 +41,7 @@ const TaskGroupName = "nexus"
 // OperationStore defines the interface that must be implemented by any parent component that wants to manage Nexus operations.
 // It's the responsibility of the parrent component to apply the appropriate state transitions to the operation.
 type OperationStore interface {
-	OnNexusOperationStarted(ctx chasm.MutableContext, operation *Operation, operationToken string, links []*commonpb.Link) error
+	OnNexusOperationStarted(ctx chasm.MutableContext, operation *Operation, operationToken string, startTime *time.Time, links []*commonpb.Link) error
 	OnNexusOperationCanceled(ctx chasm.MutableContext, operation *Operation, cause *failurepb.Failure) error
 	OnNexusOperationFailed(ctx chasm.MutableContext, operation *Operation, cause *failurepb.Failure) error
 	OnNexusOperationTimedOut(ctx chasm.MutableContext, operation *Operation, cause *failurepb.Failure) error
@@ -125,14 +126,15 @@ func (o *Operation) Cancel(ctx chasm.MutableContext, parentData *anypb.Any) erro
 }

 // onStarted applies the started transition or delegates to the store if one is present.
-func (o *Operation) onStarted(ctx chasm.MutableContext, operationToken string, links []*commonpb.Link) error {
+func (o *Operation) onStarted(ctx chasm.MutableContext, operationToken string, startTime *time.Time, links []*commonpb.Link) error {
 	store, ok := o.Store.TryGet(ctx)
 	if ok {
-		return store.OnNexusOperationStarted(ctx, o, operationToken, links)
+		return store.OnNexusOperationStarted(ctx, o, operationToken, startTime, links)
 	}
 	// TODO(stephan): for standalone, store links
 	return TransitionStarted.Apply(o, ctx, EventStarted{
 		OperationToken: operationToken,
+		StartTime:      startTime,
 	})
 }

@@ -230,7 +232,7 @@ func (o *Operation) saveInvocationResult(
 		if r.response.Pending != nil {
 			// An async operation transitions to STARTED here;
 			// HandleNexusCompletion will apply its outcome from the completion callback.
-			return nil, o.onStarted(ctx, r.response.Pending.Token, links)
+			return nil, o.onStarted(ctx, r.response.Pending.Token, nil, links)
 		}
 		return nil, o.onCompleted(ctx, r.response.Successful, links)
 	case startResultCancel:
@@ -254,16 +256,27 @@ func (o *Operation) HandleNexusCompletion(
 	ctx chasm.MutableContext,
 	completion *persistencespb.ChasmNexusCompletion,
 ) error {
-	// TODO: support completion-before-start
-
 	// Request ID lets us reject a stale or misrouted completion.
 	if completion.GetRequestId() != "" && o.GetRequestId() != completion.GetRequestId() {
 		return serviceerror.NewNotFound("operation not found")
 	}

+	links := completion.GetLinks()
+
+	// For completion-before-start, apply the started transition first.
+	if o.GetStatus() == nexusoperationpb.OPERATION_STATUS_SCHEDULED {
+		startTime := timestamp.TimeValuePtr(completion.GetStartTime())
+		if err := o.onStarted(ctx, completion.GetOperationToken(), startTime, links); err != nil {
+			return err
+		}
+
+		// Links belong only to the synthetic started event.
+		links = nil
+	}
+
 	switch outcome := completion.Outcome.(type) {
 	case *persistencespb.ChasmNexusCompletion_Success:
-		return o.onCompleted(ctx, outcome.Success, completion.GetLinks())
+		return o.onCompleted(ctx, outcome.Success, links)
 	case *persistencespb.ChasmNexusCompletion_Failure:
 		if outcome.Failure.GetCanceledFailureInfo() != nil {
 			return o.onCanceled(ctx, outcome.Failure)
--- a/chasm/lib/nexusoperation/operation_test.go
+++ b/chasm/lib/nexusoperation/operation_test.go
@@ -10,6 +10,7 @@ import (
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
 	nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )

 func newScheduledTestOperation(t *testing.T, ctx *chasm.MockMutableContext) *Operation {
@@ -35,47 +36,129 @@ func TestHandleNexusCompletion(t *testing.T) {
 	}

 	t.Run("Success", func(t *testing.T) {
-		ctx := newCtx()
-		op := newStartedOp(t, ctx)
-		err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
-			RequestId: op.GetRequestId(),
-			Outcome: &persistencespb.ChasmNexusCompletion_Success{
-				Success: mustToPayload(t, "result"),
-			},
+		t.Run("AfterStarted", func(t *testing.T) {
+			ctx := newCtx()
+			op := newStartedOp(t, ctx)
+			err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				RequestId: op.GetRequestId(),
+				Outcome: &persistencespb.ChasmNexusCompletion_Success{
+					Success: mustToPayload(t, "result"),
+				},
+			})
+			require.NoError(t, err)
+			require.Equal(t, nexusoperationpb.OPERATION_STATUS_SUCCEEDED, op.GetStatus())
+		})
+
+		t.Run("CompletionBeforeStart", func(t *testing.T) {
+			ctx := newCtx()
+			op := newScheduledTestOperation(t, ctx)
+			startTime := defaultTime.Add(-time.Second)
+			err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				StartTime:      timestamppb.New(startTime),
+				RequestId:      op.GetRequestId(),
+				OperationToken: "tok",
+				Outcome: &persistencespb.ChasmNexusCompletion_Success{
+					Success: mustToPayload(t, "result"),
+				},
+			})
+			require.NoError(t, err)
+			require.Equal(t, nexusoperationpb.OPERATION_STATUS_SUCCEEDED, op.GetStatus())
+			require.Equal(t, "tok", op.GetOperationToken())
+			require.Equal(t, startTime, op.GetStartedTime().AsTime())
+		})
+
+		t.Run("CompletionBeforeStartWithoutStartTime", func(t *testing.T) {
+			ctx := newCtx()
+			op := newScheduledTestOperation(t, ctx)
+			err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				// missing StartTime
+				RequestId:      op.GetRequestId(),
+				OperationToken: "tok",
+				Outcome: &persistencespb.ChasmNexusCompletion_Success{
+					Success: mustToPayload(t, "result"),
+				},
+			})
+			require.NoError(t, err)
+			require.Equal(t, nexusoperationpb.OPERATION_STATUS_SUCCEEDED, op.GetStatus())
+			require.Equal(t, "tok", op.GetOperationToken())
+			require.Equal(t, defaultTime, op.GetStartedTime().AsTime())
 		})
-		require.NoError(t, err)
-		require.Equal(t, nexusoperationpb.OPERATION_STATUS_SUCCEEDED, op.GetStatus())
 	})

 	t.Run("Failure", func(t *testing.T) {
-		ctx := newCtx()
-		op := newStartedOp(t, ctx)
-		err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
-			RequestId: op.GetRequestId(),
-			Outcome: &persistencespb.ChasmNexusCompletion_Failure{
-				Failure: &failurepb.Failure{Message: "oops"},
-			},
+		t.Run("AfterStarted", func(t *testing.T) {
+			ctx := newCtx()
+			op := newStartedOp(t, ctx)
+			err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				RequestId: op.GetRequestId(),
+				Outcome: &persistencespb.ChasmNexusCompletion_Failure{
+					Failure: &failurepb.Failure{Message: "oops"},
+				},
+			})
+			require.NoError(t, err)
+			require.Equal(t, nexusoperationpb.OPERATION_STATUS_FAILED, op.GetStatus())
+		})
+
+		t.Run("CompletionBeforeStart", func(t *testing.T) {
+			ctx := newCtx()
+			op := newScheduledTestOperation(t, ctx)
+			startTime := defaultTime.Add(-time.Second)
+			err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				StartTime:      timestamppb.New(startTime),
+				RequestId:      op.GetRequestId(),
+				OperationToken: "tok",
+				Outcome: &persistencespb.ChasmNexusCompletion_Failure{
+					Failure: &failurepb.Failure{Message: "oops"},
+				},
+			})
+			require.NoError(t, err)
+			require.Equal(t, nexusoperationpb.OPERATION_STATUS_FAILED, op.GetStatus())
+			require.Equal(t, "tok", op.GetOperationToken())
+			require.Equal(t, startTime, op.GetStartedTime().AsTime())
 		})
-		require.NoError(t, err)
-		require.Equal(t, nexusoperationpb.OPERATION_STATUS_FAILED, op.GetStatus())
 	})

 	t.Run("Canceled", func(t *testing.T) {
-		ctx := newCtx()
-		op := newStartedOp(t, ctx)
-		err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
-			RequestId: op.GetRequestId(),
-			Outcome: &persistencespb.ChasmNexusCompletion_Failure{
-				Failure: &failurepb.Failure{
-					Message: "canceled",
-					FailureInfo: &failurepb.Failure_CanceledFailureInfo{
-						CanceledFailureInfo: &failurepb.CanceledFailureInfo{},
+		t.Run("AfterStarted", func(t *testing.T) {
+			ctx := newCtx()
+			op := newStartedOp(t, ctx)
+			err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				RequestId: op.GetRequestId(),
+				Outcome: &persistencespb.ChasmNexusCompletion_Failure{
+					Failure: &failurepb.Failure{
+						Message: "canceled",
+						FailureInfo: &failurepb.Failure_CanceledFailureInfo{
+							CanceledFailureInfo: &failurepb.CanceledFailureInfo{},
+						},
 					},
 				},
-			},
+			})
+			require.NoError(t, err)
+			require.Equal(t, nexusoperationpb.OPERATION_STATUS_CANCELED, op.GetStatus())
+		})
+
+		t.Run("CompletionBeforeStart", func(t *testing.T) {
+			ctx := newCtx()
+			op := newScheduledTestOperation(t, ctx)
+			startTime := defaultTime.Add(-time.Second)
+			err := op.HandleNexusCompletion(ctx, &persistencespb.ChasmNexusCompletion{
+				StartTime:      timestamppb.New(startTime),
+				RequestId:      op.GetRequestId(),
+				OperationToken: "tok",
+				Outcome: &persistencespb.ChasmNexusCompletion_Failure{
+					Failure: &failurepb.Failure{
+						Message: "canceled",
+						FailureInfo: &failurepb.Failure_CanceledFailureInfo{
+							CanceledFailureInfo: &failurepb.CanceledFailureInfo{},
+						},
+					},
+				},
+			})
+			require.NoError(t, err)
+			require.Equal(t, nexusoperationpb.OPERATION_STATUS_CANCELED, op.GetStatus())
+			require.Equal(t, "tok", op.GetOperationToken())
+			require.Equal(t, startTime, op.GetStartedTime().AsTime())
 		})
-		require.NoError(t, err)
-		require.Equal(t, nexusoperationpb.OPERATION_STATUS_CANCELED, op.GetStatus())
 	})

 	t.Run("RequestIDMismatch", func(t *testing.T) {
--- a/chasm/lib/nexusoperation/tasks_test_helpers_test.go
+++ b/chasm/lib/nexusoperation/tasks_test_helpers_test.go
@@ -2,6 +2,7 @@ package nexusoperation

 import (
 	"testing"
+	"time"

 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
@@ -45,8 +46,11 @@ type mockStoreComponent struct {
 	// Data is required by CHASM for serialization - every component needs a proto.Message field.
 	Data *nexusoperationpb.OperationState

-	invocationData InvocationData
-	Op             chasm.Field[*Operation]
+	invocationData  InvocationData
+	Op              chasm.Field[*Operation]
+	startLinks      []*commonpb.Link
+	completionLinks []*commonpb.Link
+	startTime       *time.Time
 }

 func (m *mockStoreComponent) LifecycleState(_ chasm.Context) chasm.LifecycleState {
@@ -65,11 +69,14 @@ func (m *mockStoreComponent) NexusOperationInvocationData(_ chasm.Context, _ *Op
 	return m.invocationData, nil
 }

-func (m *mockStoreComponent) OnNexusOperationStarted(ctx chasm.MutableContext, op *Operation, operationToken string, _ []*commonpb.Link) error {
-	return TransitionStarted.Apply(op, ctx, EventStarted{OperationToken: operationToken})
+func (m *mockStoreComponent) OnNexusOperationStarted(ctx chasm.MutableContext, op *Operation, operationToken string, startTime *time.Time, links []*commonpb.Link) error {
+	m.startTime = startTime
+	m.startLinks = links
+	return TransitionStarted.Apply(op, ctx, EventStarted{OperationToken: operationToken, StartTime: startTime})
 }

-func (m *mockStoreComponent) OnNexusOperationCompleted(ctx chasm.MutableContext, op *Operation, _ *commonpb.Payload, _ []*commonpb.Link) error {
+func (m *mockStoreComponent) OnNexusOperationCompleted(ctx chasm.MutableContext, op *Operation, _ *commonpb.Payload, links []*commonpb.Link) error {
+	m.completionLinks = links
 	return TransitionSucceeded.Apply(op, ctx, EventSucceeded{})
 }

--- a/chasm/lib/nexusoperation/workflow/events.go
+++ b/chasm/lib/nexusoperation/workflow/events.go
@@ -182,9 +182,12 @@ func (d StartedEventDefinition) Apply(ctx chasm.MutableContext, wf *chasmworkflo

 	// TODO: Store event.Links on the Operation for standalone mode, where links won't be available via history.

-	return nexusoperation.TransitionStarted.Apply(op, ctx, nexusoperation.EventStarted{
+	startTime := event.GetEventTime().AsTime()
+	evt := nexusoperation.EventStarted{
 		OperationToken: attrs.GetOperationToken(),
-	})
+		StartTime:      &startTime,
+	}
+	return nexusoperation.TransitionStarted.Apply(op, ctx, evt)
 }

 func (d StartedEventDefinition) CherryPick(ctx chasm.MutableContext, wf *chasmworkflow.Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error {
@@ -214,7 +217,9 @@ func (d CompletedEventDefinition) Apply(ctx chasm.MutableContext, wf *chasmworkf
 	}
 	op := field.Get(ctx)

-	if err := nexusoperation.TransitionSucceeded.Apply(op, ctx, nexusoperation.EventSucceeded{}); err != nil {
+	completeTime := event.GetEventTime().AsTime()
+	evt := nexusoperation.EventSucceeded{CompleteTime: &completeTime}
+	if err := nexusoperation.TransitionSucceeded.Apply(op, ctx, evt); err != nil {
 		return err
 	}
 	wf.RemoveNexusOperation(attrs.GetScheduledEventId())
@@ -248,7 +253,9 @@ func (d FailedEventDefinition) Apply(ctx chasm.MutableContext, wf *chasmworkflow
 	}
 	op := field.Get(ctx)

-	if err := nexusoperation.TransitionFailed.Apply(op, ctx, nexusoperation.EventFailed{}); err != nil {
+	completeTime := event.GetEventTime().AsTime()
+	evt := nexusoperation.EventFailed{CompleteTime: &completeTime}
+	if err := nexusoperation.TransitionFailed.Apply(op, ctx, evt); err != nil {
 		return err
 	}
 	wf.RemoveNexusOperation(attrs.GetScheduledEventId())
@@ -282,7 +289,9 @@ func (d CanceledEventDefinition) Apply(ctx chasm.MutableContext, wf *chasmworkfl
 	}
 	op := field.Get(ctx)

-	if err := nexusoperation.TransitionCanceled.Apply(op, ctx, nexusoperation.EventCanceled{}); err != nil {
+	completeTime := event.GetEventTime().AsTime()
+	evt := nexusoperation.EventCanceled{CompleteTime: &completeTime}
+	if err := nexusoperation.TransitionCanceled.Apply(op, ctx, evt); err != nil {
 		return err
 	}
 	wf.RemoveNexusOperation(attrs.GetScheduledEventId())
--- a/chasm/lib/nexusoperation/workflow/events_test.go
+++ b/chasm/lib/nexusoperation/workflow/events_test.go
@@ -35,6 +35,47 @@ func scheduleOperation(t *testing.T, tcx testContext) (*historypb.HistoryEvent,
 	return event, event.EventId
 }

+func applyStartedEvent(t *testing.T, tcx testContext, scheduledEventID int64, eventTime time.Time) {
+	t.Helper()
+	applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, &historypb.HistoryEvent{
+		EventTime: timestamppb.New(eventTime),
+		Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
+			NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
+				ScheduledEventId: scheduledEventID,
+				OperationToken:   "token",
+			},
+		},
+	})
+}
+
+func applyEventDefinition(
+	t *testing.T,
+	tcx testContext,
+	eventType enumspb.EventType,
+	event *historypb.HistoryEvent,
+) {
+	t.Helper()
+	chReg := chasmworkflow.NewRegistry()
+	require.NoError(t, chReg.Register(newLibrary(defaultConfig, chasm.NewNexusEndpointProcessor())))
+	def, ok := chReg.EventDefinitionByEventType(eventType)
+	require.True(t, ok)
+	err := def.Apply(tcx.chasmCtx, tcx.wf, event)
+	require.NoError(t, err)
+}
+
+func assertTerminalEventApplied(
+	t *testing.T,
+	tcx testContext,
+	key int64,
+	op *nexusoperationpb.OperationState,
+	expectedStatus nexusoperationpb.OperationStatus,
+) {
+	t.Helper()
+	require.Equal(t, expectedStatus, op.GetStatus())
+	_, ok := tcx.wf.Operations[key]
+	require.False(t, ok, "operation should be removed after terminal event")
+}
+
 func TestCherryPick(t *testing.T) {
 	t.Run("should exclude nexus events", func(t *testing.T) {
 		tcx := newTestContext(t, defaultConfig)
@@ -117,93 +158,158 @@ func TestCherryPick(t *testing.T) {
 	})
 }

-func TestTerminalStatesDeletion(t *testing.T) {
-	testCases := []struct {
-		name      string
-		eventType enumspb.EventType
-		event     *historypb.HistoryEvent
-	}{
-		{
-			name:      "CompletedDeletesOperation",
-			eventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED,
-		},
-		{
-			name:      "FailedDeletesOperation",
-			eventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED,
-		},
-		{
-			name:      "CanceledDeletesOperation",
-			eventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED,
-		},
-		{
-			name:      "TimedOutDeletesOperation",
-			eventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT,
-		},
+func TestCompletedEventDefinitionApply(t *testing.T) {
+	eventTime := time.Now().UTC()
+	buildEvent := func(scheduledEventID int64) *historypb.HistoryEvent {
+		return &historypb.HistoryEvent{
+			EventTime: timestamppb.New(eventTime),
+			Attributes: &historypb.HistoryEvent_NexusOperationCompletedEventAttributes{
+				NexusOperationCompletedEventAttributes: &historypb.NexusOperationCompletedEventAttributes{
+					ScheduledEventId: scheduledEventID,
+				},
+			},
+		}
 	}

-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			tcx := newTestContext(t, defaultConfig)
-			scheduledEvent, key := scheduleOperation(t, tcx)
-			scheduledEventID := scheduledEvent.EventId
+	t.Run("without started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		// no start event
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_SUCCEEDED)
+		require.Equal(t, eventTime, op.GetClosedTime().AsTime())
+	})

-			// Verify operation exists.
-			_, ok := tcx.wf.Operations[key]
-			require.True(t, ok)
+	t.Run("with started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		applyStartedEvent(t, tcx, scheduledEvent.EventId, eventTime) // add start event firsts
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_SUCCEEDED)
+		require.Equal(t, eventTime, op.GetClosedTime().AsTime())
+	})
+}

-			// Build the terminal event.
-			event := &historypb.HistoryEvent{
-				EventTime: timestamppb.Now(),
-			}
-			switch tc.eventType {
-			case enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED:
-				event.Attributes = &historypb.HistoryEvent_NexusOperationCompletedEventAttributes{
-					NexusOperationCompletedEventAttributes: &historypb.NexusOperationCompletedEventAttributes{
-						ScheduledEventId: scheduledEventID,
-					},
-				}
-			case enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED:
-				event.Attributes = &historypb.HistoryEvent_NexusOperationFailedEventAttributes{
-					NexusOperationFailedEventAttributes: &historypb.NexusOperationFailedEventAttributes{
-						ScheduledEventId: scheduledEventID,
-					},
-				}
-			case enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED:
-				event.Attributes = &historypb.HistoryEvent_NexusOperationCanceledEventAttributes{
-					NexusOperationCanceledEventAttributes: &historypb.NexusOperationCanceledEventAttributes{
-						ScheduledEventId: scheduledEventID,
-					},
-				}
-			case enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT:
-				event.Attributes = &historypb.HistoryEvent_NexusOperationTimedOutEventAttributes{
-					NexusOperationTimedOutEventAttributes: &historypb.NexusOperationTimedOutEventAttributes{
-						ScheduledEventId: scheduledEventID,
-					},
-				}
-			default:
-				t.Fatalf("unexpected event type: %v", tc.eventType)
-			}
-
-			// Look up the event definition from the registry.
-			chReg := chasmworkflow.NewRegistry()
-			require.NoError(t, chReg.Register(newLibrary(defaultConfig, chasm.NewNexusEndpointProcessor())))
-			def, ok := chReg.EventDefinitionByEventType(tc.eventType)
-			require.True(t, ok)
-
-			err := def.Apply(tcx.chasmCtx, tcx.wf, event)
-			require.NoError(t, err)
-
-			// Verify operation was removed.
-			_, ok = tcx.wf.Operations[key]
-			require.False(t, ok, "operation should be removed after terminal event")
-		})
+func TestFailedEventDefinitionApply(t *testing.T) {
+	eventTime := time.Now().UTC()
+	buildEvent := func(scheduledEventID int64) *historypb.HistoryEvent {
+		return &historypb.HistoryEvent{
+			EventTime: timestamppb.New(eventTime),
+			Attributes: &historypb.HistoryEvent_NexusOperationFailedEventAttributes{
+				NexusOperationFailedEventAttributes: &historypb.NexusOperationFailedEventAttributes{
+					ScheduledEventId: scheduledEventID,
+				},
+			},
+		}
 	}
+
+	t.Run("without started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		// no start event
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_FAILED)
+		require.Equal(t, eventTime, op.GetClosedTime().AsTime())
+	})
+
+	t.Run("with started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		applyStartedEvent(t, tcx, scheduledEvent.EventId, eventTime) // add start event first
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_FAILED)
+		require.Equal(t, eventTime, op.GetClosedTime().AsTime())
+	})
+}
+
+func TestCanceledEventDefinitionApply(t *testing.T) {
+	eventTime := time.Now().UTC()
+	buildEvent := func(scheduledEventID int64) *historypb.HistoryEvent {
+		return &historypb.HistoryEvent{
+			EventTime: timestamppb.New(eventTime),
+			Attributes: &historypb.HistoryEvent_NexusOperationCanceledEventAttributes{
+				NexusOperationCanceledEventAttributes: &historypb.NexusOperationCanceledEventAttributes{
+					ScheduledEventId: scheduledEventID,
+				},
+			},
+		}
+	}
+
+	t.Run("without started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		// no start event
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_CANCELED)
+		require.Equal(t, eventTime, op.GetClosedTime().AsTime())
+	})
+
+	t.Run("with started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		applyStartedEvent(t, tcx, scheduledEvent.EventId, eventTime) // add start event first
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_CANCELED)
+		require.Equal(t, eventTime, op.GetClosedTime().AsTime())
+	})
+}
+
+func TestTimedOutEventDefinitionApply(t *testing.T) {
+	eventTime := time.Now().UTC()
+	buildEvent := func(scheduledEventID int64) *historypb.HistoryEvent {
+		return &historypb.HistoryEvent{
+			EventTime: timestamppb.New(eventTime),
+			Attributes: &historypb.HistoryEvent_NexusOperationTimedOutEventAttributes{
+				NexusOperationTimedOutEventAttributes: &historypb.NexusOperationTimedOutEventAttributes{
+					ScheduledEventId: scheduledEventID,
+				},
+			},
+		}
+	}
+
+	t.Run("without started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		// no start event
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_TIMED_OUT)
+	})
+
+	t.Run("with started event", func(t *testing.T) {
+		tcx := newTestContext(t, defaultConfig)
+		scheduledEvent, key := scheduleOperation(t, tcx)
+		field, ok := tcx.wf.Operations[key]
+		require.True(t, ok)
+		op := field.Get(tcx.chasmCtx)
+		applyStartedEvent(t, tcx, scheduledEvent.EventId, eventTime) // add start event first
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT, buildEvent(scheduledEvent.EventId))
+		assertTerminalEventApplied(t, tcx, key, op.OperationState, nexusoperationpb.OPERATION_STATUS_TIMED_OUT)
+	})
 }

 func TestScheduledEventDefinitionApply(t *testing.T) {
 	tcx := newTestContext(t, defaultConfig)
-
-	def := ScheduledEventDefinition{}
 	event := &historypb.HistoryEvent{
 		EventId:   int64(10),
 		EventTime: timestamppb.Now(),
@@ -220,8 +326,7 @@ func TestScheduledEventDefinitionApply(t *testing.T) {
 		},
 	}

-	err := def.Apply(tcx.chasmCtx, tcx.wf, event)
-	require.NoError(t, err)
+	applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, event)

 	field, ok := tcx.wf.Operations[event.EventId]
 	require.True(t, ok)
@@ -237,10 +342,10 @@ func TestScheduledEventDefinitionApply(t *testing.T) {
 func TestStartedEventDefinitionApply(t *testing.T) {
 	tcx := newTestContext(t, defaultConfig)
 	event, key := scheduleOperation(t, tcx)
+	startTime := time.Now().UTC()

-	def := StartedEventDefinition{}
-	err := def.Apply(tcx.chasmCtx, tcx.wf, &historypb.HistoryEvent{
-		EventTime: timestamppb.Now(),
+	applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, &historypb.HistoryEvent{
+		EventTime: timestamppb.New(startTime),
 		Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
 			NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
 				ScheduledEventId: event.EventId,
@@ -248,22 +353,20 @@ func TestStartedEventDefinitionApply(t *testing.T) {
 			},
 		},
 	})
-	require.NoError(t, err)

 	field, ok := tcx.wf.Operations[key]
 	require.True(t, ok)
 	op := field.Get(tcx.chasmCtx)
 	require.Equal(t, nexusoperationpb.OPERATION_STATUS_STARTED, op.Status)
 	require.Equal(t, "test-token", op.GetOperationToken())
+	require.Equal(t, startTime, op.GetStartedTime().AsTime())
 }

 func TestCancelRequestedEventDefinitionApply(t *testing.T) {
 	t.Run("creates cancellation child", func(t *testing.T) {
 		tcx := newTestContext(t, defaultConfig)
 		event, key := scheduleOperation(t, tcx)
-
-		def := CancelRequestedEventDefinition{}
-		err := def.Apply(tcx.chasmCtx, tcx.wf, &historypb.HistoryEvent{
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED, &historypb.HistoryEvent{
 			EventId:   int64(20),
 			EventTime: timestamppb.Now(),
 			Attributes: &historypb.HistoryEvent_NexusOperationCancelRequestedEventAttributes{
@@ -272,7 +375,6 @@ func TestCancelRequestedEventDefinitionApply(t *testing.T) {
 				},
 			},
 		})
-		require.NoError(t, err)

 		field, ok := tcx.wf.Operations[key]
 		require.True(t, ok)
@@ -283,9 +385,7 @@ func TestCancelRequestedEventDefinitionApply(t *testing.T) {

 	t.Run("tolerates missing operation", func(t *testing.T) {
 		tcx := newTestContext(t, defaultConfig)
-
-		def := CancelRequestedEventDefinition{}
-		err := def.Apply(tcx.chasmCtx, tcx.wf, &historypb.HistoryEvent{
+		applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED, &historypb.HistoryEvent{
 			EventId:   int64(20),
 			EventTime: timestamppb.Now(),
 			Attributes: &historypb.HistoryEvent_NexusOperationCancelRequestedEventAttributes{
@@ -294,7 +394,6 @@ func TestCancelRequestedEventDefinitionApply(t *testing.T) {
 				},
 			},
 		})
-		require.NoError(t, err)
 	})
 }

@@ -303,8 +402,7 @@ func TestCancelRequestCompletedEventDefinitionApply(t *testing.T) {
 	event, key := scheduleOperation(t, tcx)

 	// First, request cancellation.
-	cancelDef := CancelRequestedEventDefinition{}
-	err := cancelDef.Apply(tcx.chasmCtx, tcx.wf, &historypb.HistoryEvent{
+	applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED, &historypb.HistoryEvent{
 		EventId:   int64(20),
 		EventTime: timestamppb.Now(),
 		Attributes: &historypb.HistoryEvent_NexusOperationCancelRequestedEventAttributes{
@@ -313,11 +411,9 @@ func TestCancelRequestCompletedEventDefinitionApply(t *testing.T) {
 			},
 		},
 	})
-	require.NoError(t, err)

 	// Transition the operation to STARTED so the cancellation gets scheduled.
-	startDef := StartedEventDefinition{}
-	err = startDef.Apply(tcx.chasmCtx, tcx.wf, &historypb.HistoryEvent{
+	applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, &historypb.HistoryEvent{
 		EventTime: timestamppb.Now(),
 		Attributes: &historypb.HistoryEvent_NexusOperationStartedEventAttributes{
 			NexusOperationStartedEventAttributes: &historypb.NexusOperationStartedEventAttributes{
@@ -326,11 +422,9 @@ func TestCancelRequestCompletedEventDefinitionApply(t *testing.T) {
 			},
 		},
 	})
-	require.NoError(t, err)

 	// Now complete the cancel request.
-	completedDef := CancelRequestCompletedEventDefinition{}
-	err = completedDef.Apply(tcx.chasmCtx, tcx.wf, &historypb.HistoryEvent{
+	applyEventDefinition(t, tcx, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED, &historypb.HistoryEvent{
 		EventTime: timestamppb.Now(),
 		Attributes: &historypb.HistoryEvent_NexusOperationCancelRequestCompletedEventAttributes{
 			NexusOperationCancelRequestCompletedEventAttributes: &historypb.NexusOperationCancelRequestCompletedEventAttributes{
@@ -338,7 +432,6 @@ func TestCancelRequestCompletedEventDefinitionApply(t *testing.T) {
 			},
 		},
 	})
-	require.NoError(t, err)

 	field, ok := tcx.wf.Operations[key]
 	require.True(t, ok)
--- a/chasm/lib/workflow/workflow.go
+++ b/chasm/lib/workflow/workflow.go
@@ -2,6 +2,7 @@ package workflow

 import (
 	"fmt"
+	"time"

 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
@@ -175,6 +176,7 @@ func (w *Workflow) OnNexusOperationStarted(
 	ctx chasm.MutableContext,
 	op *nexusoperation.Operation,
 	operationToken string,
+	startTime *time.Time,
 	links []*commonpb.Link,
 ) error {
 	parentData := &workflowpb.NexusOperationParentData{}
@@ -191,6 +193,10 @@ func (w *Workflow) OnNexusOperationStarted(
 			},
 		}
 		e.Links = links
+		if startTime != nil {
+			// For completion-before-start, use the callback-provided start time for the synthetic started event.
+			e.EventTime = timestamppb.New(*startTime)
+		}
 	})
 	return err
 }
--- a/proto/internal/temporal/server/api/historyservice/v1/request_response.proto
+++ b/proto/internal/temporal/server/api/historyservice/v1/request_response.proto
@@ -1207,6 +1207,11 @@ message CompleteNexusOperationChasmRequest {
  google.protobuf.Timestamp close_time = 4;
  // Links from the Nexus completion callback (e.g. references to the handler workflow).
  repeated temporal.api.common.v1.Link links = 5;
+  // Async operation token from the callback request, used to synthesize a started event when
+  // completion arrives before the worker's start response.
+  string operation_token = 6;
+  // Time when the operation was started. Used when completion is received before the started response.
+  google.protobuf.Timestamp start_time = 7;
 }

 message CompleteNexusOperationChasmResponse {}
--- a/proto/internal/temporal/server/api/persistence/v1/chasm.proto
+++ b/proto/internal/temporal/server/api/persistence/v1/chasm.proto
@@ -130,4 +130,8 @@ message ChasmNexusCompletion {
  string request_id = 4;
  // Links from the Nexus completion callback (e.g. references to the handler workflow).
  repeated temporal.api.common.v1.Link links = 5;
+  // Async operation token from the callback request, used for completion-before-start.
+  string operation_token = 6;
+  // Start time from the callback request, used for completion-before-start.
+  google.protobuf.Timestamp start_time = 7;
 }
--- a/service/frontend/nexus_completion_http_handler.go
+++ b/service/frontend/nexus_completion_http_handler.go
@@ -328,7 +328,11 @@ func (h *nexusCompletionHandler) completeChasmOperation(
 			RequestId:    completion.GetRequestId(),
 			ComponentRef: completion.GetComponentRef(),
 		},
-		Links: links,
+		Links:          links,
+		OperationToken: req.OperationToken,
+	}
+	if !req.StartTime.IsZero() {
+		hr.StartTime = timestamppb.New(req.StartTime)
 	}
 	if !req.CloseTime.IsZero() {
 		hr.CloseTime = timestamppb.New(req.CloseTime)
--- a/service/history/handler.go
+++ b/service/history/handler.go
@@ -2257,9 +2257,11 @@ func (h *Handler) CompleteNexusOperationChasm(
 	request *historyservice.CompleteNexusOperationChasmRequest,
 ) (*historyservice.CompleteNexusOperationChasmResponse, error) {
 	completion := &persistencespb.ChasmNexusCompletion{
-		CloseTime: request.CloseTime,
-		RequestId: request.Completion.RequestId,
-		Links:     request.Links,
+		StartTime:      request.StartTime,
+		CloseTime:      request.CloseTime,
+		RequestId:      request.Completion.RequestId,
+		Links:          request.Links,
+		OperationToken: request.OperationToken,
 	}
 	switch variant := request.Outcome.(type) {
 	case *historyservice.CompleteNexusOperationChasmRequest_Failure:
--- a/tests/nexus_workflow_test.go
+++ b/tests/nexus_workflow_test.go
@@ -932,9 +932,6 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletion(chasmEnabled
 }

 func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionBeforeStart(chasmEnabled bool) {
-	if chasmEnabled {
-		s.T().Skip("Blocked on CHASM Nexus completion-before-start support")
-	}
 	env := s.newNexusWorkflowTestEnv(chasmEnabled)
 	ctx := testcore.NewContext()
 	taskQueues := []string{testcore.RandomizeStr(s.T().Name()), testcore.RandomizeStr(s.T().Name())}
@@ -1089,6 +1086,15 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionBeforeStart(ch
 		},
 	})
 	s.NoError(err)
+	completionWorkflowHistory := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{
+		WorkflowId: completionWFID,
+		RunId:      completionWfRunIDs[0],
+	})
+	completionWorkflowStartedEventIdx := slices.IndexFunc(completionWorkflowHistory, func(e *historypb.HistoryEvent) bool {
+		return e.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED
+	})
+	s.NotEqual(-1, completionWorkflowStartedEventIdx)
+	completionWorkflowStartTime := completionWorkflowHistory[completionWorkflowStartedEventIdx].GetEventTime().AsTime()

 	expectedLinks := []*commonpb.Link_WorkflowEvent{
 		{
@@ -1126,18 +1132,26 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionBeforeStart(ch
 			Identity: "test",
 		})
 		s.NoError(err)
+
 		startedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
 			return e.GetNexusOperationStartedEventAttributes() != nil
 		})
 		s.NotEqual(-1, startedEventIdx)
 		nexusOpStartedEvent := pollResp.History.Events[startedEventIdx]
 		s.Equal(completionWFID, nexusOpStartedEvent.GetNexusOperationStartedEventAttributes().OperationToken)
+		s.Equal(
+			completionWorkflowStartTime.Truncate(time.Second),
+			nexusOpStartedEvent.GetEventTime().AsTime().Truncate(time.Second),
+		)
 		s.Len(nexusOpStartedEvent.Links, 1)
 		s.ProtoEqual(expectedLinks[i], nexusOpStartedEvent.Links[0].GetWorkflowEvent())
+
 		completedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
 			return e.GetNexusOperationCompletedEventAttributes() != nil
 		})
 		s.Positive(completedEventIdx)
+		s.Less(startedEventIdx, completedEventIdx)
+		s.Empty(pollResp.History.Events[completedEventIdx].Links)

 		// Complete start request to verify response is ignored.
 		_, err = env.FrontendClient().RespondNexusTaskCompleted(ctx, &workflowservice.RespondNexusTaskCompletedRequest{
@@ -1169,7 +1183,7 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionBeforeStart(ch
 						CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
 							Result: &commonpb.Payloads{
 								Payloads: []*commonpb.Payload{
-									pollResp.History.Events[completedEventIdx].GetNexusOperationCompletedEventAttributes().Result,
+									testcore.MustToPayload(s.T(), "result"),
 								},
 							},
 						},