diff --git a/backend/pkg/app/backend.go b/backend/pkg/app/backend.go
index 6d6d1d10107..09bf92852b9 100644
--- a/backend/pkg/app/backend.go
+++ b/backend/pkg/app/backend.go
@@ -47,6 +47,7 @@ import (
 	credentialrevocationdeletion "github.com/Azure/ARO-HCP/backend/pkg/controllers/cluster/credentialrevocation/deletion"
 	credentialrevocationoperations "github.com/Azure/ARO-HCP/backend/pkg/controllers/cluster/credentialrevocation/operations"
 	clusterdeletion "github.com/Azure/ARO-HCP/backend/pkg/controllers/cluster/deletion"
+	clusteridentity "github.com/Azure/ARO-HCP/backend/pkg/controllers/cluster/identity"
 	"github.com/Azure/ARO-HCP/backend/pkg/controllers/cluster/legacycredentialrequest"
 	clusteroperations "github.com/Azure/ARO-HCP/backend/pkg/controllers/cluster/operations"
 	clusterplacement "github.com/Azure/ARO-HCP/backend/pkg/controllers/cluster/placement"
@@ -695,9 +696,8 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 		unionKubeApplierInformers,
 		unionReadDesireLister,
 	)
-	identityMigrationController := clusterproperties.NewIdentityMigrationController(
+	clusterIdentitySyncController := clusteridentity.NewClusterIdentitySyncController(
 		b.options.ResourcesDBClient,
-		b.options.ClustersServiceClient,
 		backendInformers,
 		unionKubeApplierInformers,
 	)
@@ -1006,6 +1006,13 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 		backendInformers,
 	)
 
+	fetchMSIIdentitiesInfoController := clusteridentity.NewFetchMSIIdentitiesInfoController(
+		b.clock,
+		b.options.ResourcesDBClient,
+		backendInformers,
+		b.options.FPAMIDataplaneClientBuilder,
+	)
+
 	leaderElectionConfig := leaderelection.LeaderElectionConfig{
 		Lock:          b.options.LeaderElectionLock,
 		LeaseDuration: sharedleaderelection.RecommendedLeaseDuration,
@@ -1069,7 +1076,7 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 				go triggerControlPlaneUpgradeController.Run(ctx, 20)
 				go clusterBaseDomainPrefixSyncController.Run(ctx, 20)
 				go clusterPropertiesSyncController.Run(ctx, 20)
-				go identityMigrationController.Run(ctx, 20)
+				go clusterIdentitySyncController.Run(ctx, 20)
 				go clusterDegradedAggregatorController.Run(ctx, 20)
 				go clusterRequirementsValidAggregatorController.Run(ctx, 20)
 				go nodePoolDegradedAggregatorController.Run(ctx, 20)
@@ -1117,6 +1124,7 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 				go cosmosMigrationController.Run(ctx, 5)
 				go virtualMachineResourceSKUsCachedReaderController.Run(ctx, 20)
 				go backupScheduleController.Run(ctx, 20)
+				go fetchMSIIdentitiesInfoController.Run(ctx, 20)
 			},
 			OnStoppedLeading: func() {
 				// This needs to be defined even though it does nothing.
diff --git a/backend/pkg/controllers/cluster/identity/cluster_identity_sync.go b/backend/pkg/controllers/cluster/identity/cluster_identity_sync.go
new file mode 100644
index 00000000000..d087d7c15e3
--- /dev/null
+++ b/backend/pkg/controllers/cluster/identity/cluster_identity_sync.go
@@ -0,0 +1,204 @@
+// Copyright 2026 Microsoft Corporation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package identity
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/equality"
+
+	"github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils"
+	"github.com/Azure/ARO-HCP/internal/api/coreapi"
+	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/corecosmosstorage"
+	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils"
+	"github.com/Azure/ARO-HCP/internal/database/informers/coreinformers"
+	"github.com/Azure/ARO-HCP/internal/database/listers/corelisters"
+	unionkubeapplierinformers "github.com/Azure/ARO-HCP/internal/database/unioninformers/kubeapplier"
+	"github.com/Azure/ARO-HCP/internal/utils"
+)
+
+const ClusterIdentitySyncControllerName = "ClusterIdentitySync"
+
+// clusterIdentitySyncer keeps ClientID/PrincipalID on
+// HCPOpenShiftCluster.Identity.UserAssignedIdentities in sync with
+// ServiceProviderCluster.Status.MSIManagedIdentities. It iterates the existing
+// Identity map keys (preserving casing) and looks up each one in SPC by
+// lowercased resource ID.
+type clusterIdentitySyncer struct {
+	clusterLister                corelisters.ClusterLister
+	serviceProviderClusterLister corelisters.ServiceProviderClusterLister
+	resourcesDBClient            corecosmosstorage.ResourcesDBClient
+}
+
+var _ controllerutils.ClusterSyncer = (*clusterIdentitySyncer)(nil)
+
+// NewClusterIdentitySyncController creates a new controller that continuously
+// syncs Identity.UserAssignedIdentities ClientID/PrincipalID from
+// ServiceProviderCluster.Status.MSIManagedIdentities.
+//
+// It compares Cluster.Identity against SPC and updates when ClientID/PrincipalID
+// would change (including nil values returned when an identity does not exist).
+// Map keys in Identity keep the casing from CustomerProperties; SPC lookups use
+// lowercased resource IDs. Keys remain even when SPC does not yet have a matching
+// identity entry. Deleting clusters are skipped.
+func NewClusterIdentitySyncController(
+	resourcesDBClient corecosmosstorage.ResourcesDBClient,
+	informers coreinformers.BackendInformers,
+	kubeApplierInformers *unionkubeapplierinformers.UnionKubeApplierInformers,
+) controllerutils.Controller {
+	_, clusterLister := informers.Clusters()
+	_, serviceProviderClusterLister := informers.ServiceProviderClusters()
+
+	syncer := &clusterIdentitySyncer{
+		clusterLister:                clusterLister,
+		serviceProviderClusterLister: serviceProviderClusterLister,
+		resourcesDBClient:            resourcesDBClient,
+	}
+
+	controller := controllerutils.NewClusterWatchingController(
+		ClusterIdentitySyncControllerName,
+		resourcesDBClient,
+		informers,
+		kubeApplierInformers,
+		60*time.Minute, // Check every 60 minutes
+		syncer,
+	)
+
+	return controller
+}
+
+func (c *clusterIdentitySyncer) NeedsWork(ctx context.Context, existingCluster *coreapi.HCPOpenShiftCluster) bool {
+	if existingCluster.ServiceProviderProperties.DeletionTimestamp != nil {
+		return false
+	}
+
+	if existingCluster.Identity == nil || len(existingCluster.Identity.UserAssignedIdentities) == 0 {
+		return false
+	}
+
+	return true
+}
+
+// SyncOnce performs a single reconciliation of cluster identity information.
+// It iterates Identity.UserAssignedIdentities, looks up each key (lowercased)
+// in ServiceProviderCluster.Status.MSIManagedIdentities, and updates
+// ClientID/PrincipalID when SPC has a match. Keys that are absent from SPC
+// remain unchanged.
+func (c *clusterIdentitySyncer) SyncOnce(ctx context.Context, key controllerutils.HCPClusterKey) error {
+	logger := utils.LoggerFromContext(ctx)
+
+	// do the super cheap cache check first
+	cachedCluster, err := c.clusterLister.Get(ctx, key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName)
+	if cosmosstorageutils.IsNotFoundError(err) {
+		// we'll be re-fired if it is created again
+		return nil
+	}
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to get cluster from cache: %w", err))
+	}
+	if !c.NeedsWork(ctx, cachedCluster) {
+		// if the cache doesn't need work, then we'll be retriggered if those values change when the cache updates.
+		// if the values don't change, then we still have no work to do.
+		return nil
+	}
+
+	// Get the cluster from Cosmos
+	clusterCRUD := c.resourcesDBClient.HCPClusters(key.SubscriptionID, key.ResourceGroupName)
+	existingCluster, err := clusterCRUD.Get(ctx, key.HCPClusterName)
+	if cosmosstorageutils.IsNotFoundError(err) {
+		return nil // cluster doesn't exist, no work to do
+	}
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to get Cluster: %w", err))
+	}
+	// check if we need to do work again. Sometimes the live data is ahead of the cache and obviates the need to do any work
+	if !c.NeedsWork(ctx, existingCluster) {
+		return nil
+	}
+
+	existingSPC, err := c.serviceProviderClusterLister.Get(ctx, key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName)
+	if cosmosstorageutils.IsNotFoundError(err) {
+		// SPC may not exist yet; nothing to copy into Identity.
+		return nil
+	}
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to get ServiceProviderCluster from cache: %w", err))
+	}
+
+	replacement := existingCluster.DeepCopy()
+	c.updateIdentityUserAssignedIdentitiesFromSPC(
+		replacement.Identity.UserAssignedIdentities,
+		existingSPC.Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities,
+		existingSPC.Status.MSIManagedIdentities.ServiceManagedIdentity,
+	)
+
+	if equality.Semantic.DeepEqual(existingCluster.Identity, replacement.Identity) {
+		return nil
+	}
+
+	// Write the updated cluster back to Cosmos
+	_, err = clusterCRUD.Replace(ctx, replacement, nil)
+	if cosmosstorageutils.IsPreconditionFailedError(err) {
+		// if we have a conflict error, then we're guaranteed that our informer will eventually see an update and trigger us again.
+		return nil
+	}
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to replace Cluster: %w", err))
+	}
+
+	logger.Info("synced identity information from ServiceProviderCluster")
+	return nil
+}
+
+// updateIdentityUserAssignedIdentitiesFromSPC walks the existing Identity map and, for
+// each key, looks up the lowercased resource ID in SPC control-plane operator
+// identities or the service managed identity. When found, ClientID and
+// PrincipalID are updated in place. Keys missing from SPC are left as-is.
+func (c *clusterIdentitySyncer) updateIdentityUserAssignedIdentitiesFromSPC(
+	identityUserAssignedIdentities map[string]*coreapi.UserAssignedIdentity,
+	spcControlPlaneOperatorsIdentities map[string]*coreapi.ServiceProviderClusterControlPlaneOperatorIdentity,
+	spcServiceManagedIdentity *coreapi.ServiceProviderClusterServiceManagedIdentity,
+) {
+	for identityResourceIDStr := range identityUserAssignedIdentities {
+		lowerResourceIDStr := strings.ToLower(identityResourceIDStr)
+
+		var clientID, principalID *string
+
+		// If we found the identity in the SPC control plane operators identities, we use the ClientID and PrincipalID from the SPC.
+		if spcIdentity, ok := spcControlPlaneOperatorsIdentities[lowerResourceIDStr]; ok && spcIdentity != nil {
+			clientID = spcIdentity.ClientID
+			principalID = spcIdentity.PrincipalID
+			// If we found the identity in the SPC service managed identity, we use the ClientID and PrincipalID from the SPC.
+		} else if spcServiceManagedIdentity != nil && spcServiceManagedIdentity.ResourceID != nil &&
+			strings.ToLower(spcServiceManagedIdentity.ResourceID.String()) == lowerResourceIDStr {
+			clientID = spcServiceManagedIdentity.ClientID
+			principalID = spcServiceManagedIdentity.PrincipalID
+		} else { // otherwise, we leave the identity as-is
+			continue
+		}
+
+		// We initialize the identity if it does not exist, to avoid nil dereferencing
+		if identityUserAssignedIdentities[identityResourceIDStr] == nil {
+			identityUserAssignedIdentities[identityResourceIDStr] = &coreapi.UserAssignedIdentity{}
+		}
+		// We update the identity with the ClientID and PrincipalID values the SPC. Those values themselves
+		// maybe be nil or empty strings.
+		identityUserAssignedIdentities[identityResourceIDStr].ClientID = clientID
+		identityUserAssignedIdentities[identityResourceIDStr].PrincipalID = principalID
+	}
+}
diff --git a/backend/pkg/controllers/cluster/identity/cluster_identity_sync_test.go b/backend/pkg/controllers/cluster/identity/cluster_identity_sync_test.go
new file mode 100644
index 00000000000..34ad117d794
--- /dev/null
+++ b/backend/pkg/controllers/cluster/identity/cluster_identity_sync_test.go
@@ -0,0 +1,442 @@
+// Copyright 2026 Microsoft Corporation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package identity
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+
+	azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
+
+	"github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils"
+	"github.com/Azure/ARO-HCP/internal/api/coreapi"
+	"github.com/Azure/ARO-HCP/internal/api/metadataapi"
+	"github.com/Azure/ARO-HCP/internal/database/cosmosstoragetesting/corecosmosstoragetesting"
+	"github.com/Azure/ARO-HCP/internal/database/listertesting/corelistertesting"
+)
+
+const (
+	testSubscriptionID     = "00000000-0000-0000-0000-000000000000"
+	testResourceGroupName  = "test-rg"
+	testClusterName        = "test-cluster"
+	testIdentityResourceID = "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/test-rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/test-identity"
+	testClientID           = "client-id-123"
+	testPrincipalID        = "principal-id-456"
+	testLocation           = "test-location"
+)
+
+func TestClusterIdentitySyncer_SyncOnce(t *testing.T) {
+	mixedCaseIdentityResourceID := "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/Test-RG/providers/Microsoft.ManagedIdentity/userAssignedIdentities/Test-Identity"
+
+	testCases := []struct {
+		name                        string
+		cachedCluster               *coreapi.HCPOpenShiftCluster // cluster in cache, nil means use same as existingCluster
+		existingCluster             *coreapi.HCPOpenShiftCluster // cluster in cosmos
+		existingSPC                 *coreapi.ServiceProviderCluster
+		expectError                 bool
+		expectedHasIdentity         bool
+		expectedIdentityCount       int
+		expectedIdentityResourceIDs []string
+		expectedClientID            *string
+		expectedPrincipalID         *string
+	}{
+		{
+			name: "cache indicates no work needed - identity already matches SPC",
+			cachedCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {
+							ClientID:    ptr.To(testClientID),
+							PrincipalID: ptr.To(testPrincipalID),
+						},
+					},
+				}
+			}),
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {
+							ClientID:    ptr.To(testClientID),
+							PrincipalID: ptr.To(testPrincipalID),
+						},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            ptr.To(testClientID),
+			expectedPrincipalID:         ptr.To(testPrincipalID),
+		},
+		{
+			name: "cache differs from SPC but live data already matches",
+			cachedCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {},
+					},
+				}
+			}),
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				// cosmos has identity filled (cache is stale)
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {
+							ClientID:    ptr.To(testClientID),
+							PrincipalID: ptr.To(testPrincipalID),
+						},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            ptr.To(testClientID),
+			expectedPrincipalID:         ptr.To(testPrincipalID),
+		},
+		{
+			name: "no work to do - identity already matches SPC",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {
+							ClientID:    ptr.To(testClientID),
+							PrincipalID: ptr.To(testPrincipalID),
+						},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            ptr.To(testClientID),
+			expectedPrincipalID:         ptr.To(testPrincipalID),
+		},
+		{
+			name: "no work to do - nil ClientID/PrincipalID already match SPC not-found values",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentityPtrs(testIdentityResourceID, nil, nil),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            nil,
+			expectedPrincipalID:         nil,
+		},
+		{
+			name: "success - update ClientID/PrincipalID when SPC values change",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {
+							ClientID:    ptr.To("old-client-id"),
+							PrincipalID: ptr.To("old-principal-id"),
+						},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            ptr.To(testClientID),
+			expectedPrincipalID:         ptr.To(testPrincipalID),
+		},
+		{
+			name: "no work to do - deleting cluster",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.ServiceProviderProperties.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            nil,
+			expectedPrincipalID:         nil,
+		},
+		{
+			name:                  "no work to do - identity is nil",
+			existingCluster:       newTestClusterForClusterIdentitySync(),
+			existingSPC:           newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:           false,
+			expectedHasIdentity:   false,
+			expectedIdentityCount: 0,
+		},
+		{
+			name: "success - fill ClientID/PrincipalID from SPC",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            ptr.To(testClientID),
+			expectedPrincipalID:         ptr.To(testPrincipalID),
+		},
+		{
+			name: "success - fill nil identity map entry from SPC",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: nil,
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(testIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            ptr.To(testClientID),
+			expectedPrincipalID:         ptr.To(testPrincipalID),
+		},
+		{
+			name: "preserves mixed-case identity keys and looks up SPC by lowercase",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						mixedCaseIdentityResourceID: {},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPCWithMSIIdentity(mixedCaseIdentityResourceID, testClientID, testPrincipalID),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{mixedCaseIdentityResourceID},
+			expectedClientID:            ptr.To(testClientID),
+			expectedPrincipalID:         ptr.To(testPrincipalID),
+		},
+		{
+			name: "keeps identity keys unchanged when SPC has no matching entry",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {},
+					},
+				}
+			}),
+			existingSPC:                 newTestSPC(),
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            nil,
+			expectedPrincipalID:         nil,
+		},
+		{
+			name: "keeps identity keys unchanged when SPC is missing",
+			existingCluster: newTestClusterForClusterIdentitySync(func(c *coreapi.HCPOpenShiftCluster) {
+				c.Identity = &coreapi.ManagedServiceIdentity{
+					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
+						testIdentityResourceID: {},
+					},
+				}
+			}),
+			existingSPC:                 nil,
+			expectError:                 false,
+			expectedHasIdentity:         true,
+			expectedIdentityCount:       1,
+			expectedIdentityResourceIDs: []string{testIdentityResourceID},
+			expectedClientID:            nil,
+			expectedPrincipalID:         nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+
+			// Setup mock DB
+			mockResourcesDBClient := corecosmosstoragetesting.NewMockResourcesDBClient()
+
+			// Create the cluster in the mock DB (cosmos)
+			clusterCRUD := mockResourcesDBClient.HCPClusters(testSubscriptionID, testResourceGroupName)
+			_, err := clusterCRUD.Create(ctx, tc.existingCluster, nil)
+			require.NoError(t, err)
+
+			// Setup slice cluster lister (cache)
+			// If cachedCluster is nil, use the same as existingCluster
+			cachedCluster := tc.cachedCluster
+			if cachedCluster == nil {
+				cachedCluster = tc.existingCluster
+			}
+			sliceClusterLister := &corelistertesting.SliceClusterLister{
+				Clusters: []*coreapi.HCPOpenShiftCluster{cachedCluster},
+			}
+
+			var spcList []*coreapi.ServiceProviderCluster
+			if tc.existingSPC != nil {
+				spcList = []*coreapi.ServiceProviderCluster{tc.existingSPC}
+			}
+			sliceSPCLister := &corelistertesting.SliceServiceProviderClusterLister{
+				ServiceProviderClusters: spcList,
+			}
+
+			// Create syncer
+			syncer := &clusterIdentitySyncer{
+				clusterLister:                sliceClusterLister,
+				serviceProviderClusterLister: sliceSPCLister,
+				resourcesDBClient:            mockResourcesDBClient,
+			}
+
+			// Execute
+			key := controllerutils.HCPClusterKey{
+				SubscriptionID:    testSubscriptionID,
+				ResourceGroupName: testResourceGroupName,
+				HCPClusterName:    testClusterName,
+			}
+			err = syncer.SyncOnce(ctx, key)
+
+			if tc.expectError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+
+			// Verify the cluster state in Cosmos
+			updatedCluster, err := clusterCRUD.Get(ctx, testClusterName)
+			require.NoError(t, err)
+
+			if tc.expectedHasIdentity {
+				require.NotNil(t, updatedCluster.Identity)
+				assert.Len(t, updatedCluster.Identity.UserAssignedIdentities, tc.expectedIdentityCount)
+				for _, expectedID := range tc.expectedIdentityResourceIDs {
+					identity, exists := updatedCluster.Identity.UserAssignedIdentities[expectedID]
+					assert.True(t, exists, "expected identity %s to exist", expectedID)
+					if !exists {
+						continue
+					}
+					require.NotNil(t, identity)
+					if tc.expectedClientID == nil {
+						assert.True(t, identity.ClientID == nil || len(*identity.ClientID) == 0)
+					} else {
+						require.NotNil(t, identity.ClientID)
+						assert.Equal(t, *tc.expectedClientID, *identity.ClientID)
+					}
+					if tc.expectedPrincipalID == nil {
+						assert.True(t, identity.PrincipalID == nil || len(*identity.PrincipalID) == 0)
+					} else {
+						require.NotNil(t, identity.PrincipalID)
+						assert.Equal(t, *tc.expectedPrincipalID, *identity.PrincipalID)
+					}
+				}
+			} else {
+				if updatedCluster.Identity != nil {
+					assert.Len(t, updatedCluster.Identity.UserAssignedIdentities, tc.expectedIdentityCount)
+				}
+			}
+		})
+	}
+}
+
+// newTestClusterForClusterIdentitySync creates a test HCPOpenShiftCluster with default values
+// for cluster identity sync testing.
+func newTestClusterForClusterIdentitySync(opts ...func(*coreapi.HCPOpenShiftCluster)) *coreapi.HCPOpenShiftCluster {
+	resourceID := metadataapi.Must(azcorearm.ParseResourceID(
+		"/subscriptions/" + testSubscriptionID +
+			"/resourceGroups/" + testResourceGroupName +
+			"/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/" + testClusterName,
+	))
+
+	cluster := &coreapi.HCPOpenShiftCluster{
+		CosmosMetadata: coreapi.CosmosMetadata{
+			ResourceID:   resourceID,
+			PartitionKey: strings.ToLower(resourceID.SubscriptionID),
+		},
+		TrackedResource: coreapi.TrackedResource{
+			Resource: coreapi.Resource{
+				ID:   resourceID,
+				Name: testClusterName,
+				Type: resourceID.ResourceType.String(),
+			},
+		},
+	}
+	cluster.Location = testLocation
+
+	for _, opt := range opts {
+		opt(cluster)
+	}
+
+	return cluster
+}
+
+func newTestSPC() *coreapi.ServiceProviderCluster {
+	clusterResourceID := metadataapi.Must(azcorearm.ParseResourceID(
+		"/subscriptions/" + testSubscriptionID +
+			"/resourceGroups/" + testResourceGroupName +
+			"/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/" + testClusterName,
+	))
+	spcResourceID := metadataapi.Must(azcorearm.ParseResourceID(
+		coreapi.ToServiceProviderClusterResourceIDString(testSubscriptionID, testResourceGroupName, testClusterName),
+	))
+	return &coreapi.ServiceProviderCluster{
+		CosmosMetadata: coreapi.CosmosMetadata{
+			ResourceID:   spcResourceID,
+			PartitionKey: strings.ToLower(clusterResourceID.SubscriptionID),
+		},
+	}
+}
+
+func newTestSPCWithMSIIdentity(identityResourceID, clientID, principalID string) *coreapi.ServiceProviderCluster {
+	return newTestSPCWithMSIIdentityPtrs(identityResourceID, ptr.To(clientID), ptr.To(principalID))
+}
+
+func newTestSPCWithMSIIdentityPtrs(identityResourceID string, clientID, principalID *string) *coreapi.ServiceProviderCluster {
+	spc := newTestSPC()
+	lowerResourceIDStr := strings.ToLower(identityResourceID)
+	spc.Status.MSIManagedIdentities = coreapi.ServiceProviderClusterMSIManagedIdentities{
+		ControlPlaneOperatorsIdentities: map[string]*coreapi.ServiceProviderClusterControlPlaneOperatorIdentity{
+			lowerResourceIDStr: {
+				ResourceID:  metadataapi.Must(azcorearm.ParseResourceID(lowerResourceIDStr)),
+				ClientID:    clientID,
+				PrincipalID: principalID,
+			},
+		},
+	}
+	return spc
+}
diff --git a/backend/pkg/controllers/cluster/identity/fetch_msi_identities_info.go b/backend/pkg/controllers/cluster/identity/fetch_msi_identities_info.go
new file mode 100644
index 00000000000..d3e47af97b1
--- /dev/null
+++ b/backend/pkg/controllers/cluster/identity/fetch_msi_identities_info.go
@@ -0,0 +1,393 @@
+// Copyright 2026 Microsoft Corporation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package identity
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/equality"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	utilsclock "k8s.io/utils/clock"
+	"k8s.io/utils/ptr"
+
+	azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
+	"github.com/Azure/msi-dataplane/pkg/dataplane"
+
+	azureclient "github.com/Azure/ARO-HCP/backend/pkg/azure/client"
+	"github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils"
+	"github.com/Azure/ARO-HCP/internal/api/coreapi"
+	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/corecosmosstorage"
+	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils"
+	"github.com/Azure/ARO-HCP/internal/database/informers/coreinformers"
+	"github.com/Azure/ARO-HCP/internal/utils"
+)
+
+const (
+	FetchMSIIdentitiesInfoControllerName = "FetchMSIIdentitiesInfo"
+
+	// msiIdentitiesRecheckInterval is the base interval before re-querying the
+	// Managed Identities Data Plane when ClientID/PrincipalID are already
+	// resolved. Combined with msiIdentitiesRecheckJitter via wait.Jitter.
+	msiIdentitiesRecheckInterval = 12 * time.Hour
+	msiIdentitiesRecheckJitter   = 0.5
+)
+
+// controlPlaneOperatorIdentityToFetch describes one control-plane operator MSI based
+// identity for which extra information should be fetched via the Managed Identities Data Plane.
+type controlPlaneOperatorIdentityToFetch struct {
+	// resourceID is the ARM resource ID of the user-assigned managed identity.
+	// SPC map key / credential lookups must ToLower the string form because
+	// ResourceID.String() is not a stable fully-lowercased key.
+	resourceID *azcorearm.ResourceID
+}
+
+// msiBasedIdentitiesToFetch holds the MSI-based identities for which extra information
+// should be fetched via the Managed Identities Data Plane.
+type msiBasedIdentitiesToFetch struct {
+	// controlPlaneOperators are the control-plane operator identities to fetch extra information for.
+	controlPlaneOperators []*controlPlaneOperatorIdentityToFetch
+	// serviceManagedIdentity is the ARM resource ID of the cluster's service
+	// managed identity for which extra information should be fetched.
+	serviceManagedIdentity *azcorearm.ResourceID
+}
+
+// resourceIDStrings returns the ARM resource ID strings for every identity to
+// resolve, suitable for a Managed Identities Data Plane credentials request.
+func (i msiBasedIdentitiesToFetch) resourceIDStrings() []string {
+	resourceIDs := make([]string, 0, len(i.controlPlaneOperators)+1)
+	for _, identity := range i.controlPlaneOperators {
+		resourceIDs = append(resourceIDs, identity.resourceID.String())
+	}
+
+	resourceIDs = append(resourceIDs, i.serviceManagedIdentity.String())
+
+	return resourceIDs
+}
+
+// fetchMSIIdentitiesInfoSyncer fetches ClientID and PrincipalID for the
+// cluster's MSI-based user-assigned managed identities and writes them onto
+// ServiceProviderCluster.Status.MSIManagedIdentities in Cosmos.
+type fetchMSIIdentitiesInfoSyncer struct {
+	clock                       utilsclock.PassiveClock
+	resourcesDBClient           corecosmosstorage.ResourcesDBClient
+	fpaMIdataplaneClientBuilder azureclient.FPAMIDataplaneClientBuilder
+}
+
+var _ controllerutils.ClusterSyncer = (*fetchMSIIdentitiesInfoSyncer)(nil)
+
+// NewFetchMSIIdentitiesInfoController creates a cluster-watching controller
+// that resolves ClientID and PrincipalID for every MSI-based identity of
+// the cluster and persists them on ServiceProviderCluster.Status.MSIManagedIdentities.
+//
+// These MSI-based identities are the cluster's control plane operator
+// managed identities and the cluster's service managed identity. Their
+// resource IDs come from
+// CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.
+// This controller fills in ClientID and PrincipalID for each one of them.
+// To do so, it calls the Managed Identities Data Plane service. In environments
+// where the real Managed Identities Data Plane service is not available, a fake
+// implementation of the Managed Identities Data Plane client is used, which
+// always returns the same information and same set of credentials for all
+// requests, independently on what identity is requested. The returned information
+// in those environments is the information associated to the "MI Mock" identity.
+//
+// On each SyncOnce the controller:
+//  1. Returns immediately when the cluster is deleting.
+//  2. Collects every identity resource ID from
+//     CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities
+//     (control plane operators and service managed identity).
+//  3. Via needsWork, skips Managed Identities Data Plane calls when
+//     ServiceProviderCluster.Status.MSIManagedIdentities.EarliestRecheckTime is
+//     still in the future AND the identities stored on SPC still match the
+//     collected desired set. If the desired identities have changed,
+//     EarliestRecheckTime is ignored so the dataplane is queried immediately.
+//     The recheck time is shared across every entry in
+//     MSIManagedIdentities.ControlPlaneOperatorsIdentities and
+//     MSIManagedIdentities.ServiceManagedIdentity.
+//  4. Calls the Managed Identities Data Plane (or the fake client implementation
+//     in environments where the real Managed Identities Data Plane service
+//     is not available) once with the set of identities.
+//  5. Matches each returned credential by ResourceID (case-insensitive.
+//     ARM IDs are case-insensitive and response order is not assumed)
+//     and sets ClientID and PrincipalID when the dataplane returns
+//     non-empty values. Resource IDs are stored lowercased in SPC.
+//  6. On a fully successful dataplane fetch, it sets EarliestRecheckTime on the
+//     in-memory replacement to now plus a long jittered interval.
+//  7. Replaces the ServiceProviderCluster document only when the identities map
+//     or recheck time changed. needsWork observes EarliestRecheckTime and the
+//     desired-vs-stored identity match from Cosmos, so a wait is introduced
+//     only after a successful Replace persists a matching set with a future
+//     EarliestRecheckTime. If Replace fails (or hits a precondition failure),
+//     the new EarliestRecheckTime is not stored. The workqueue requeues and the
+//     next needsWork still sees the previously persisted value (typically nil
+//     or already past, or a mismatched identity set), so the controller does
+//     not wait out the long recheck interval after write failures either.
+func NewFetchMSIIdentitiesInfoController(
+	clock utilsclock.PassiveClock,
+	resourcesDBClient corecosmosstorage.ResourcesDBClient,
+	backendInformers coreinformers.BackendInformers,
+	fpaMIdataplaneClientBuilder azureclient.FPAMIDataplaneClientBuilder,
+) controllerutils.Controller {
+	if clock == nil {
+		clock = utilsclock.RealClock{}
+	}
+
+	syncer := &fetchMSIIdentitiesInfoSyncer{
+		clock:                       clock,
+		resourcesDBClient:           resourcesDBClient,
+		fpaMIdataplaneClientBuilder: fpaMIdataplaneClientBuilder,
+	}
+
+	controller := controllerutils.NewClusterWatchingController(
+		FetchMSIIdentitiesInfoControllerName,
+		resourcesDBClient,
+		backendInformers,
+		nil,
+		1*time.Minute,
+		syncer,
+	)
+
+	return controller
+}
+
+// needsWork reports whether the Managed Identities Data Plane should be
+// queried for MSI identity metadata. EarliestRecheckTime is honored only when
+// the identities stored on SPC still match desiredIdentitiesToFetch; on
+// mismatch (or if future skip-prerequisites fail), it returns true immediately.
+// When identities match, it returns false while EarliestRecheckTime is in the
+// future, and true when EarliestRecheckTime is nil or already past. Callers
+// must skip needsWork entirely when the cluster is deleting.
+func (c *fetchMSIIdentitiesInfoSyncer) needsWork(existingSPC *coreapi.ServiceProviderCluster, desiredIdentitiesToFetch *msiBasedIdentitiesToFetch) bool {
+	// Only honor EarliestRecheckTime when the desired identity set still matches
+	// SPC. Any mismatch (or future "must work now" conditions added alongside
+	// this check) should fall through to return true and query the dataplane.
+	if c.desiredMSIResourceIDsMatchSPC(desiredIdentitiesToFetch, existingSPC) {
+		// Desired identity set still matches SPC. Honor EarliestRecheckTime so we
+		// do not repeatedly query the Managed Identities Data Plane for the same
+		// identities. Nil means recheck immediately; a future time means skip work.
+		earliestRecheckTime := existingSPC.Status.MSIManagedIdentities.EarliestRecheckTime
+		if earliestRecheckTime != nil && c.clock.Now().Before(earliestRecheckTime.Time) {
+			return false
+		}
+	}
+
+	return true
+}
+
+// desiredMSIResourceIDsMatchSPC reports whether the MSI resource IDs stored on
+// SPC match desiredIdentitiesToFetch. Comparison is by lowercased resource ID
+// presence/equality; ClientID/PrincipalID and operator names are ignored.
+func (c *fetchMSIIdentitiesInfoSyncer) desiredMSIResourceIDsMatchSPC(desiredIdentitiesToFetch *msiBasedIdentitiesToFetch, spc *coreapi.ServiceProviderCluster) bool {
+	spcMSIManagedIdentities := spc.Status.MSIManagedIdentities
+
+	spcServiceManagedIdentity := spcMSIManagedIdentities.ServiceManagedIdentity
+
+	// If the SPC service managed identity is nil, the identities do not match, because the cluster should always have a service managed identity.
+	if spcServiceManagedIdentity == nil || spcServiceManagedIdentity.ResourceID == nil {
+		return false
+	}
+	// If the SPC service managed identity resource ID does not match the cluster one then the identities do not match.
+	if !strings.EqualFold(desiredIdentitiesToFetch.serviceManagedIdentity.String(), spcServiceManagedIdentity.ResourceID.String()) {
+		return false
+	}
+
+	// If the number of control plane operators is different, the identities do not match.
+	if len(desiredIdentitiesToFetch.controlPlaneOperators) != len(spcMSIManagedIdentities.ControlPlaneOperatorsIdentities) {
+		return false
+	}
+
+	for _, identity := range desiredIdentitiesToFetch.controlPlaneOperators {
+		// SPC map keys are lowercased strings. ResourceID.String() may re-canonicalize casing so we lowercase.
+		resourceIDStr := strings.ToLower(identity.resourceID.String())
+		_, ok := spcMSIManagedIdentities.ControlPlaneOperatorsIdentities[resourceIDStr]
+		if !ok {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (c *fetchMSIIdentitiesInfoSyncer) SyncOnce(ctx context.Context, key controllerutils.HCPClusterKey) error {
+	existingCluster, err := c.resourcesDBClient.HCPClusters(key.SubscriptionID, key.ResourceGroupName).Get(ctx, key.HCPClusterName)
+	if cosmosstorageutils.IsNotFoundError(err) {
+		return nil // cluster doesn't exist, no work to do
+	}
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to get Cluster: %w", err))
+	}
+
+	if existingCluster.ServiceProviderProperties.DeletionTimestamp != nil {
+		return nil
+	}
+
+	spcCRUD := c.resourcesDBClient.ServiceProviderClusters(key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName)
+	existingSPC, err := spcCRUD.Get(ctx, coreapi.ServiceProviderClusterResourceName)
+	if cosmosstorageutils.IsNotFoundError(err) {
+		return nil // SPC doesn't exist yet, no work to do
+	}
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to get ServiceProviderCluster: %w", err))
+	}
+
+	msiBasedIdentitiesToFetch, err := c.collectMSIBasedIdentitiesToFetch(existingCluster)
+	if err != nil {
+		return err
+	}
+
+	if !c.needsWork(existingSPC, msiBasedIdentitiesToFetch) {
+		return nil
+	}
+
+	identitiesToSyncResourceIDStrs := msiBasedIdentitiesToFetch.resourceIDStrings()
+
+	// On environments where the real Managed Identities Data Plane service is not available, a
+	// fake implementation of the Managed Identities Data Plane client is used, which always returns the same information and
+	// same set of credentials for all requests, independently on what identity is request. The returned information is
+	// the information associated to the "MI Mock" identity.
+	fpaMIDataplaneClient, err := c.fpaMIdataplaneClientBuilder.ManagedIdentitiesDataplane(existingCluster.ServiceProviderProperties.ManagedIdentitiesDataPlaneIdentityURL)
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to get Managed Identities Data Plane Client: %w", err))
+	}
+
+	// We get all the Managed Identities information in a single Managed Identities Data Plane Credentials request to minimize
+	// calls to the Managed Identities Data Plane Service.
+	fpaMIDataplaneCredentialsRequest := dataplane.UserAssignedIdentitiesRequest{IdentityIDs: identitiesToSyncResourceIDStrs}
+	fpaMIDataplaneCredentials, err := fpaMIDataplaneClient.GetUserAssignedIdentitiesCredentials(ctx, fpaMIDataplaneCredentialsRequest)
+	if err != nil {
+		return utils.TrackError(fmt.Errorf("failed to get Managed Identities Data Plane Credentials: %w", err))
+	}
+
+	if len(fpaMIDataplaneCredentials.ExplicitIdentities) != len(identitiesToSyncResourceIDStrs) {
+		return utils.TrackError(fmt.Errorf("unexpected number of Managed Identities Data Plane Credentials. Expected: %d, Received: %d", len(identitiesToSyncResourceIDStrs), len(fpaMIDataplaneCredentials.ExplicitIdentities)))
+	}
+
+	// Index returned credentials by lowercased Resource ID so later lookups are
+	// case-insensitive. ARM resource IDs are case-insensitive and the MI dataplane may return a different casing than Cosmos, as well as a
+	// different order than how it's been requested.
+	returnedCredentialsByLowerResourceID := make(map[string]dataplane.UserAssignedIdentityCredentials, len(fpaMIDataplaneCredentials.ExplicitIdentities))
+	for idx, fpaMIDataplaneCredential := range fpaMIDataplaneCredentials.ExplicitIdentities {
+		if fpaMIDataplaneCredential.ResourceID == nil || len(*fpaMIDataplaneCredential.ResourceID) == 0 {
+			// The MIDataplane service should not return a nil or empty Resource ID. This is the case even when the identity does not exist in Azure.
+			// If this occurs, we return an error instead of accumulating it as this is unexpected and should not happen..
+			return utils.TrackError(fmt.Errorf("unexpected Managed Identities Data Plane Credential Resource ID is nil or empty in MI Dataplane service response at index %d (Resource ID %q, Client ID %q, Principal ID %q)",
+				idx,
+				ptr.Deref(fpaMIDataplaneCredential.ResourceID, ""),
+				ptr.Deref(fpaMIDataplaneCredential.ClientID, ""),
+				ptr.Deref(fpaMIDataplaneCredential.ObjectID, ""),
+			))
+		}
+		returnedCredentialsByLowerResourceID[strings.ToLower(*fpaMIDataplaneCredential.ResourceID)] = fpaMIDataplaneCredential
+	}
+
+	// For ClientID and PrincipalID of each identity, we set the value returned from the MIDataplane service. This includes
+	// the cases where the value is nil or empty. At the moment of writing this (2026-08-11), when the actual identity does
+	// not exist in Azure, the MIDataplane service returns null for ClientID and PrincipalID.
+	// SPC map keys are lowercased; ResourceID.String() may re-canonicalize casing, so always ToLower for keys.
+	replacementControlPlaneOperatorsIdentities := make(map[string]*coreapi.ServiceProviderClusterControlPlaneOperatorIdentity, len(msiBasedIdentitiesToFetch.controlPlaneOperators))
+	for _, identity := range msiBasedIdentitiesToFetch.controlPlaneOperators {
+		resourceIDStr := strings.ToLower(identity.resourceID.String())
+		credential, ok := returnedCredentialsByLowerResourceID[resourceIDStr]
+		if !ok {
+			// The MIDataplane service should return a Resource ID that matches one of the identities requested. That is even if the identity actually does not exist anymore in Azure.
+			// If it does not, we return an error instead of accumulating it.
+			return utils.TrackError(fmt.Errorf("unexpected Managed Identities Data Plane Credential %s Resource ID is not found in the cluster's identities", resourceIDStr))
+		}
+		replacementControlPlaneOperatorsIdentities[resourceIDStr] = &coreapi.ServiceProviderClusterControlPlaneOperatorIdentity{
+			ResourceID:  coreapi.DeepCopyResourceID(identity.resourceID),
+			ClientID:    credential.ClientID,
+			PrincipalID: credential.ObjectID,
+		}
+	}
+
+	serviceManagedIdentityResourceIDStr := strings.ToLower(msiBasedIdentitiesToFetch.serviceManagedIdentity.String())
+	serviceManagedIdentityCredential, ok := returnedCredentialsByLowerResourceID[serviceManagedIdentityResourceIDStr]
+	if !ok {
+		// The MIDataplane service should return a Resource ID that matches one of the identities requested. That is even if the identity actually does not exist anymore in Azure.
+		// If it does not, we return an error instead of accumulating it.
+		return utils.TrackError(fmt.Errorf("unexpected Managed Identities Data Plane Credential %s Resource ID is not found in the cluster's identities", serviceManagedIdentityResourceIDStr))
+	}
+	replacementServiceManagedIdentity := &coreapi.ServiceProviderClusterServiceManagedIdentity{
+		ResourceID:  coreapi.DeepCopyResourceID(msiBasedIdentitiesToFetch.serviceManagedIdentity),
+		ClientID:    serviceManagedIdentityCredential.ClientID,
+		PrincipalID: serviceManagedIdentityCredential.ObjectID,
+	}
+
+	replacement := existingSPC.DeepCopy()
+	replacement.Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities = replacementControlPlaneOperatorsIdentities
+	replacement.Status.MSIManagedIdentities.ServiceManagedIdentity = replacementServiceManagedIdentity
+
+	// Set an earliest recheck time for the controller so we do not hit the Managed Identities Data Plane service too often
+	// when the desired identity set is unchanged. needsWork ignores this wait when OperatorsAuthentication diverges from
+	// the identities stored on SPC. The value below is only honored once Replace persists it. A Replace failure leaves
+	// Cosmos unchanged, so needsWork will still see the previously persisted value (if any).
+	earliestRecheckAt := metav1.NewTime(c.clock.Now().Add(wait.Jitter(
+		msiIdentitiesRecheckInterval,
+		msiIdentitiesRecheckJitter,
+	)))
+	replacement.Status.MSIManagedIdentities.EarliestRecheckTime = &earliestRecheckAt
+
+	controlPlaneOperatorsUnchanged := equality.Semantic.DeepEqual(replacement.Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities, existingSPC.Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities)
+	serviceManagedIdentityUnchanged := equality.Semantic.DeepEqual(replacement.Status.MSIManagedIdentities.ServiceManagedIdentity, existingSPC.Status.MSIManagedIdentities.ServiceManagedIdentity)
+	recheckUnchanged := equality.Semantic.DeepEqual(replacement.Status.MSIManagedIdentities.EarliestRecheckTime, existingSPC.Status.MSIManagedIdentities.EarliestRecheckTime)
+	if controlPlaneOperatorsUnchanged && serviceManagedIdentityUnchanged && recheckUnchanged {
+		return nil
+	}
+
+	_, err = spcCRUD.Replace(ctx, replacement, nil)
+	if cosmosstorageutils.IsPreconditionFailedError(err) {
+		// Status (including any new EarliestRecheckTime) was not written.
+		// needsWork will still see the previously persisted value.
+		return nil
+	}
+	if err != nil {
+		// Same as precondition failure: EarliestRecheckTime was not
+		// persisted, so needsWork will still see the previously persisted value.
+		return utils.TrackError(fmt.Errorf("failed to replace ServiceProviderCluster: %w", err))
+	}
+
+	return nil
+}
+
+// collectMSIBasedIdentitiesToFetch returns the control-plane operator identities and
+// the service managed identity that should be resolved via the Managed
+// Identities Data Plane.
+func (c *fetchMSIIdentitiesInfoSyncer) collectMSIBasedIdentitiesToFetch(cluster *coreapi.HCPOpenShiftCluster) (*msiBasedIdentitiesToFetch, error) {
+	identities := &msiBasedIdentitiesToFetch{}
+
+	for operatorName, operatorIdentityResourceID := range cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators {
+		if len(operatorName) == 0 {
+			return nil, utils.TrackError(fmt.Errorf("unexpected empty operator name for control plane operator"))
+		}
+		if operatorIdentityResourceID == nil {
+			return nil, utils.TrackError(fmt.Errorf("unexpected nil identity Resource ID string for control plane operator %q", operatorName))
+		}
+
+		identities.controlPlaneOperators = append(identities.controlPlaneOperators, &controlPlaneOperatorIdentityToFetch{
+			resourceID: coreapi.DeepCopyResourceID(operatorIdentityResourceID),
+		})
+	}
+
+	serviceManagedIdentity := cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity
+	if serviceManagedIdentity == nil {
+		return nil, utils.TrackError(fmt.Errorf("unexpected nil identity Resource ID for service managed identity"))
+	}
+	identities.serviceManagedIdentity = coreapi.DeepCopyResourceID(serviceManagedIdentity)
+
+	return identities, nil
+}
diff --git a/backend/pkg/controllers/cluster/identity/fetch_msi_identities_info_test.go b/backend/pkg/controllers/cluster/identity/fetch_msi_identities_info_test.go
new file mode 100644
index 00000000000..c8de28fb086
--- /dev/null
+++ b/backend/pkg/controllers/cluster/identity/fetch_msi_identities_info_test.go
@@ -0,0 +1,250 @@
+// Copyright 2026 Microsoft Corporation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package identity
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clocktesting "k8s.io/utils/clock/testing"
+	"k8s.io/utils/ptr"
+
+	azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
+
+	"github.com/Azure/ARO-HCP/internal/api/coreapi"
+	"github.com/Azure/ARO-HCP/internal/api/metadataapi"
+)
+
+const (
+	testOperatorName               = "cloud-controller-manager"
+	testOperatorIdentityResourceID = "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/test-rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/ccm"
+	testServiceManagedIdentityID   = "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/test-rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/smi"
+	testOtherOperatorIdentityID    = "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/test-rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/other"
+)
+
+func TestDesiredMSIResourceIDsMatchSPC(t *testing.T) {
+	t.Parallel()
+
+	syncer := &fetchMSIIdentitiesInfoSyncer{}
+	matchingCluster, matchingSPC := newMatchingClusterAndSPC()
+	matchingToFetch, err := syncer.collectMSIBasedIdentitiesToFetch(matchingCluster)
+	require.NoError(t, err, "collect matching identities")
+
+	testCases := []struct {
+		name    string
+		toFetch *msiBasedIdentitiesToFetch
+		spc     *coreapi.ServiceProviderCluster
+		want    bool
+	}{
+		{
+			name:    "matching control plane and service managed identity",
+			toFetch: matchingToFetch,
+			spc:     matchingSPC,
+			want:    true,
+		},
+		{
+			name: "resource ID casing differences still match",
+			toFetch: func() *msiBasedIdentitiesToFetch {
+				cluster, _ := newMatchingClusterAndSPC()
+				cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators[testOperatorName] =
+					metadataapi.Must(azcorearm.ParseResourceID(strings.ToUpper(testOperatorIdentityResourceID)))
+				cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity =
+					metadataapi.Must(azcorearm.ParseResourceID(strings.ToUpper(testServiceManagedIdentityID)))
+				toFetch, err := syncer.collectMSIBasedIdentitiesToFetch(cluster)
+				require.NoError(t, err, "collect casing-variant identities")
+				return toFetch
+			}(),
+			spc:  matchingSPC,
+			want: true,
+		},
+		{
+			name: "service managed identity resource ID changed",
+			toFetch: func() *msiBasedIdentitiesToFetch {
+				cluster, _ := newMatchingClusterAndSPC()
+				cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity =
+					metadataapi.Must(azcorearm.ParseResourceID(testOtherOperatorIdentityID))
+				toFetch, err := syncer.collectMSIBasedIdentitiesToFetch(cluster)
+				require.NoError(t, err, "collect diverged SMI identities")
+				return toFetch
+			}(),
+			spc:  matchingSPC,
+			want: false,
+		},
+		{
+			name: "control plane operator resource ID changed",
+			toFetch: func() *msiBasedIdentitiesToFetch {
+				cluster, _ := newMatchingClusterAndSPC()
+				cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators[testOperatorName] =
+					metadataapi.Must(azcorearm.ParseResourceID(testOtherOperatorIdentityID))
+				toFetch, err := syncer.collectMSIBasedIdentitiesToFetch(cluster)
+				require.NoError(t, err, "collect diverged control-plane identities")
+				return toFetch
+			}(),
+			spc:  matchingSPC,
+			want: false,
+		},
+		{
+			name: "control plane operator name rebound to same resource ID still matches",
+			toFetch: func() *msiBasedIdentitiesToFetch {
+				cluster, _ := newMatchingClusterAndSPC()
+				delete(cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators, testOperatorName)
+				cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators["ingress"] =
+					metadataapi.Must(azcorearm.ParseResourceID(testOperatorIdentityResourceID))
+				toFetch, err := syncer.collectMSIBasedIdentitiesToFetch(cluster)
+				require.NoError(t, err, "collect rebound operator identities")
+				return toFetch
+			}(),
+			spc:  matchingSPC,
+			want: true,
+		},
+		{
+			name:    "extra stored control plane identity",
+			toFetch: matchingToFetch,
+			spc: func() *coreapi.ServiceProviderCluster {
+				_, spc := newMatchingClusterAndSPC()
+				otherLower := strings.ToLower(testOtherOperatorIdentityID)
+				spc.Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities[otherLower] = &coreapi.ServiceProviderClusterControlPlaneOperatorIdentity{
+					ResourceID: metadataapi.Must(azcorearm.ParseResourceID(otherLower)),
+				}
+				return spc
+			}(),
+			want: false,
+		},
+		{
+			name:    "missing stored service managed identity",
+			toFetch: matchingToFetch,
+			spc: func() *coreapi.ServiceProviderCluster {
+				_, spc := newMatchingClusterAndSPC()
+				spc.Status.MSIManagedIdentities.ServiceManagedIdentity = nil
+				return spc
+			}(),
+			want: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, syncer.desiredMSIResourceIDsMatchSPC(tc.toFetch, tc.spc))
+		})
+	}
+}
+
+func TestNeedsWorkIgnoresEarliestRecheckWhenIdentitiesDiverge(t *testing.T) {
+	t.Parallel()
+
+	now := time.Date(2026, 8, 14, 12, 0, 0, 0, time.UTC)
+	future := metav1.NewTime(now.Add(6 * time.Hour))
+	past := metav1.NewTime(now.Add(-time.Hour))
+
+	syncer := &fetchMSIIdentitiesInfoSyncer{
+		clock: clocktesting.NewFakePassiveClock(now),
+	}
+
+	matchingCluster, matchingSPC := newMatchingClusterAndSPC()
+	matchingSPC.Status.MSIManagedIdentities.EarliestRecheckTime = &future
+	matchingToFetch, err := syncer.collectMSIBasedIdentitiesToFetch(matchingCluster)
+	require.NoError(t, err, "collect matching identities")
+
+	divergedCluster, _ := newMatchingClusterAndSPC()
+	divergedCluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity =
+		metadataapi.Must(azcorearm.ParseResourceID(testOtherOperatorIdentityID))
+	divergedToFetch, err := syncer.collectMSIBasedIdentitiesToFetch(divergedCluster)
+	require.NoError(t, err, "collect diverged identities")
+
+	testCases := []struct {
+		name    string
+		toFetch *msiBasedIdentitiesToFetch
+		spc     *coreapi.ServiceProviderCluster
+		want    bool
+	}{
+		{
+			name:    "matching identities with future recheck skips work",
+			toFetch: matchingToFetch,
+			spc:     matchingSPC,
+			want:    false,
+		},
+		{
+			name:    "matching identities with past recheck needs work",
+			toFetch: matchingToFetch,
+			spc: func() *coreapi.ServiceProviderCluster {
+				_, spc := newMatchingClusterAndSPC()
+				spc.Status.MSIManagedIdentities.EarliestRecheckTime = &past
+				return spc
+			}(),
+			want: true,
+		},
+		{
+			name:    "diverged identities ignore future recheck",
+			toFetch: divergedToFetch,
+			spc:     matchingSPC,
+			want:    true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, syncer.needsWork(tc.spc, tc.toFetch))
+		})
+	}
+}
+
+func newMatchingClusterAndSPC() (*coreapi.HCPOpenShiftCluster, *coreapi.ServiceProviderCluster) {
+	operatorResourceID := metadataapi.Must(azcorearm.ParseResourceID(testOperatorIdentityResourceID))
+	serviceManagedIdentity := metadataapi.Must(azcorearm.ParseResourceID(testServiceManagedIdentityID))
+	lowerOperatorResourceIDStr := strings.ToLower(testOperatorIdentityResourceID)
+	lowerServiceManagedIdentityStr := strings.ToLower(testServiceManagedIdentityID)
+
+	cluster := &coreapi.HCPOpenShiftCluster{
+		CustomerProperties: coreapi.HCPOpenShiftClusterCustomerProperties{
+			Platform: coreapi.CustomerPlatformProfile{
+				OperatorsAuthentication: coreapi.OperatorsAuthenticationProfile{
+					UserAssignedIdentities: coreapi.UserAssignedIdentitiesProfile{
+						ControlPlaneOperators: map[string]*azcorearm.ResourceID{
+							testOperatorName: operatorResourceID,
+						},
+						ServiceManagedIdentity: serviceManagedIdentity,
+					},
+				},
+			},
+		},
+	}
+
+	spc := &coreapi.ServiceProviderCluster{
+		Status: coreapi.ServiceProviderClusterStatus{
+			MSIManagedIdentities: coreapi.ServiceProviderClusterMSIManagedIdentities{
+				ControlPlaneOperatorsIdentities: map[string]*coreapi.ServiceProviderClusterControlPlaneOperatorIdentity{
+					lowerOperatorResourceIDStr: {
+						ResourceID:  metadataapi.Must(azcorearm.ParseResourceID(lowerOperatorResourceIDStr)),
+						ClientID:    ptr.To("client-id"),
+						PrincipalID: ptr.To("principal-id"),
+					},
+				},
+				ServiceManagedIdentity: &coreapi.ServiceProviderClusterServiceManagedIdentity{
+					ResourceID:  metadataapi.Must(azcorearm.ParseResourceID(lowerServiceManagedIdentityStr)),
+					ClientID:    ptr.To("smi-client-id"),
+					PrincipalID: ptr.To("smi-principal-id"),
+				},
+			},
+		},
+	}
+
+	return cluster, spc
+}
diff --git a/backend/pkg/controllers/cluster/properties/identity_migration.go b/backend/pkg/controllers/cluster/properties/identity_migration.go
deleted file mode 100644
index fd5d479fce9..00000000000
--- a/backend/pkg/controllers/cluster/properties/identity_migration.go
+++ /dev/null
@@ -1,213 +0,0 @@
-// Copyright 2026 Microsoft Corporation
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package properties
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	"k8s.io/utils/ptr"
-
-	"github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils"
-	"github.com/Azure/ARO-HCP/internal/api/coreapi"
-	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/corecosmosstorage"
-	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils"
-	"github.com/Azure/ARO-HCP/internal/database/informers/coreinformers"
-	"github.com/Azure/ARO-HCP/internal/database/listers/corelisters"
-	unionkubeapplierinformers "github.com/Azure/ARO-HCP/internal/database/unioninformers/kubeapplier"
-	"github.com/Azure/ARO-HCP/internal/ocm"
-	"github.com/Azure/ARO-HCP/internal/utils"
-)
-
-// identityMigrationSyncer is a Cluster syncer that migrates cluster identity information
-// from Cluster Service to Cosmos DB. It ensures that the Identity.UserAssignedIdentities
-// field is populated for clusters that were created before all identity state was held in Cosmos.
-type identityMigrationSyncer struct {
-	clusterLister        corelisters.ClusterLister
-	resourcesDBClient    corecosmosstorage.ResourcesDBClient
-	clusterServiceClient ocm.ClusterServiceClientSpec
-}
-
-var _ controllerutils.ClusterSyncer = (*identityMigrationSyncer)(nil)
-
-// NewIdentityMigrationController creates a new controller that migrates identity information
-// from Cluster Service to Cosmos DB.
-// It periodically checks each cluster and populates the Identity.UserAssignedIdentities
-// field if it is not set, using GetClusterServiceUserAssignedIdentities to extract the identity data.
-func NewIdentityMigrationController(
-	resourcesDBClient corecosmosstorage.ResourcesDBClient,
-	clusterServiceClient ocm.ClusterServiceClientSpec,
-	informers coreinformers.BackendInformers,
-	kubeApplierInformers *unionkubeapplierinformers.UnionKubeApplierInformers,
-) controllerutils.Controller {
-	_, clusterLister := informers.Clusters()
-
-	syncer := &identityMigrationSyncer{
-		clusterLister:        clusterLister,
-		resourcesDBClient:    resourcesDBClient,
-		clusterServiceClient: clusterServiceClient,
-	}
-
-	controller := controllerutils.NewClusterWatchingController(
-		"IdentityMigration",
-		resourcesDBClient,
-		informers,
-		kubeApplierInformers,
-		60*time.Minute, // Check every 60 minutes
-		syncer,
-	)
-
-	return controller
-}
-
-func (c *identityMigrationSyncer) NeedsWork(ctx context.Context, existingCluster *coreapi.HCPOpenShiftCluster) bool {
-	// Check if we have a cluster service ID to query
-	if existingCluster.ServiceProviderProperties.ClusterServiceID == nil || len(existingCluster.ServiceProviderProperties.ClusterServiceID.String()) == 0 {
-		return false
-	}
-
-	// Check if identity information needs to be migrated
-	// Records that have UserAssignedIdentities already have all the identity info stored in cosmos
-	// Records that don't have this information need to be migrated
-	if existingCluster.Identity == nil {
-		return true
-	}
-	if len(existingCluster.Identity.UserAssignedIdentities) == 0 {
-		return true
-	}
-
-	expectedIdentityResourceIDs := map[string]struct{}{}
-	for _, resourceID := range existingCluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators {
-		if resourceID != nil {
-			expectedIdentityResourceIDs[resourceID.String()] = struct{}{}
-		}
-	}
-	if serviceManagedIdentity := existingCluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity; serviceManagedIdentity != nil {
-		expectedIdentityResourceIDs[serviceManagedIdentity.String()] = struct{}{}
-	}
-
-	for operatorIdentityResourceIDString, userAssignedIdentity := range existingCluster.Identity.UserAssignedIdentities {
-		if userAssignedIdentity == nil || len(ptr.Deref(userAssignedIdentity.ClientID, "")) == 0 || len(ptr.Deref(userAssignedIdentity.PrincipalID, "")) == 0 {
-			// try to fill in the information.
-			return true
-		}
-
-		if _, ok := expectedIdentityResourceIDs[operatorIdentityResourceIDString]; !ok {
-			// need to prune
-			return true
-		}
-	}
-
-	for _, operatorIdentityResourceID := range existingCluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators {
-		if operatorIdentityResourceID == nil {
-			return true
-		}
-		if needsWorkForIdentityKey(existingCluster.Identity.UserAssignedIdentities, operatorIdentityResourceID.String()) {
-			return true
-		}
-	}
-	if serviceManagedIdentity := existingCluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity; serviceManagedIdentity != nil {
-		if needsWorkForIdentityKey(existingCluster.Identity.UserAssignedIdentities, serviceManagedIdentity.String()) {
-			return true
-		}
-	}
-
-	return false
-}
-
-// needsWorkForIdentityKey returns true when the identity at key is missing or has empty
-// client/principal IDs, signalling that the migration controller should fill it in.
-func needsWorkForIdentityKey(userAssignedIdentities map[string]*coreapi.UserAssignedIdentity, key string) bool {
-	identity, ok := userAssignedIdentities[key]
-	if !ok || identity == nil {
-		return true
-	}
-	if len(ptr.Deref(identity.ClientID, "")) == 0 || len(ptr.Deref(identity.PrincipalID, "")) == 0 {
-		return true
-	}
-	return false
-}
-
-// SyncOnce performs a single reconciliation of cluster identity information.
-// It checks if the Identity.UserAssignedIdentities field is unset,
-// and if so, fetches the values from Cluster Service using
-// GetClusterServiceUserAssignedIdentities and updates Cosmos with
-// the Identity.UserAssignedIdentities only.
-func (c *identityMigrationSyncer) SyncOnce(ctx context.Context, key controllerutils.HCPClusterKey) error {
-	logger := utils.LoggerFromContext(ctx)
-
-	// do the super cheap cache check first
-	cachedCluster, err := c.clusterLister.Get(ctx, key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName)
-	if cosmosstorageutils.IsNotFoundError(err) {
-		// we'll be re-fired if it is created again
-		return nil
-	}
-	if err != nil {
-		return utils.TrackError(fmt.Errorf("failed to get cluster from cache: %w", err))
-	}
-	if !c.NeedsWork(ctx, cachedCluster) {
-		// if the cache doesn't need work, then we'll be retriggered if those values change when the cache updates.
-		// if the values don't change, then we still have no work to do.
-		return nil
-	}
-
-	// Get the cluster from Cosmos
-	clusterCRUD := c.resourcesDBClient.HCPClusters(key.SubscriptionID, key.ResourceGroupName)
-	existingCluster, err := clusterCRUD.Get(ctx, key.HCPClusterName)
-	if cosmosstorageutils.IsNotFoundError(err) {
-		return nil // cluster doesn't exist, no work to do
-	}
-	if err != nil {
-		return utils.TrackError(fmt.Errorf("failed to get Cluster: %w", err))
-	}
-	// check if we need to do work again. Sometimes the live data is ahead of the cache and obviates the need to do any work
-	if !c.NeedsWork(ctx, existingCluster) {
-		return nil
-	}
-
-	// Fetch the cluster from Cluster Service
-	csCluster, err := c.clusterServiceClient.GetCluster(ctx, *existingCluster.ServiceProviderProperties.ClusterServiceID)
-	if err != nil {
-		return utils.TrackError(fmt.Errorf("failed to get cluster from Cluster Service: %w", err))
-	}
-
-	// Use GetClusterServiceUserAssignedIdentities on a deep copy to extract identity data
-	userAssignedIdentities := ocm.GetClusterServiceUserAssignedIdentities(csCluster)
-	if len(userAssignedIdentities) == 0 {
-		// nothing to set
-		return nil
-	}
-
-	// Only assign the Identity.UserAssignedIdentities from the converted cluster
-	replacement := existingCluster.DeepCopy()
-	if replacement.Identity == nil {
-		replacement.Identity = &coreapi.ManagedServiceIdentity{}
-	}
-	replacement.Identity.UserAssignedIdentities = userAssignedIdentities
-
-	// Write the updated cluster back to Cosmos
-	_, err = clusterCRUD.Replace(ctx, replacement, nil)
-	if cosmosstorageutils.IsPreconditionFailedError(err) {
-		// if we have a conflict error, then we're guaranteed that our informer will eventually see an update and trigger us again.
-		return nil
-	}
-	if err != nil {
-		return utils.TrackError(fmt.Errorf("failed to replace Cluster: %w", err))
-	}
-
-	logger.Info("migrated identity information from Cluster Service")
-	return nil
-}
diff --git a/backend/pkg/controllers/cluster/properties/identity_migration_test.go b/backend/pkg/controllers/cluster/properties/identity_migration_test.go
deleted file mode 100644
index b4c45d074e9..00000000000
--- a/backend/pkg/controllers/cluster/properties/identity_migration_test.go
+++ /dev/null
@@ -1,309 +0,0 @@
-// Copyright 2026 Microsoft Corporation
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package properties
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"go.uber.org/mock/gomock"
-
-	azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
-
-	arohcpv1alpha1 "github.com/openshift-online/ocm-sdk-go/arohcp/v1alpha1"
-
-	"github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils"
-	"github.com/Azure/ARO-HCP/internal/api/coreapi"
-	"github.com/Azure/ARO-HCP/internal/api/metadataapi"
-	"github.com/Azure/ARO-HCP/internal/database/cosmosstoragetesting/corecosmosstoragetesting"
-	"github.com/Azure/ARO-HCP/internal/database/listertesting/corelistertesting"
-	"github.com/Azure/ARO-HCP/internal/ocm"
-)
-
-const (
-	testIdentityResourceID = "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/test-rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/test-identity"
-	testClientID           = "client-id-123"
-	testPrincipalID        = "principal-id-456"
-	testLocation           = "test-location"
-)
-
-func TestIdentityMigrationSyncer_SyncOnce(t *testing.T) {
-	testCases := []struct {
-		name                        string
-		cachedCluster               *coreapi.HCPOpenShiftCluster // cluster in cache, nil means use same as existingCluster
-		existingCluster             *coreapi.HCPOpenShiftCluster // cluster in cosmos
-		csCluster                   *arohcpv1alpha1.Cluster
-		csError                     error
-		expectCosmosGet             bool
-		expectCSCall                bool
-		expectCosmosUpdate          bool
-		expectError                 bool
-		expectedHasIdentity         bool
-		expectedIdentityCount       int
-		expectedIdentityResourceIDs []string
-	}{
-		{
-			name: "cache indicates no work needed - identity already set",
-			cachedCluster: newTestClusterForIdentityMigration(func(c *coreapi.HCPOpenShiftCluster) {
-				c.Identity = &coreapi.ManagedServiceIdentity{
-					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
-						testIdentityResourceID: {
-							ClientID:    stringPtr(testClientID),
-							PrincipalID: stringPtr(testPrincipalID),
-						},
-					},
-				}
-				c.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators = map[string]*azcorearm.ResourceID{
-					"test-operator": metadataapi.Must(azcorearm.ParseResourceID(testIdentityResourceID)),
-				}
-			}),
-			existingCluster: newTestClusterForIdentityMigration(func(c *coreapi.HCPOpenShiftCluster) {
-				c.Identity = &coreapi.ManagedServiceIdentity{
-					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
-						testIdentityResourceID: {
-							ClientID:    stringPtr(testClientID),
-							PrincipalID: stringPtr(testPrincipalID),
-						},
-					},
-				}
-				c.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators = map[string]*azcorearm.ResourceID{
-					"test-operator": metadataapi.Must(azcorearm.ParseResourceID(testIdentityResourceID)),
-				}
-			}),
-			expectCosmosGet:             false,
-			expectCSCall:                false,
-			expectCosmosUpdate:          false,
-			expectError:                 false,
-			expectedHasIdentity:         true,
-			expectedIdentityCount:       1,
-			expectedIdentityResourceIDs: []string{testIdentityResourceID},
-		},
-		{
-			name:          "cache says work needed but live data has identity",
-			cachedCluster: newTestClusterForIdentityMigration(), // cache has no identity
-			existingCluster: newTestClusterForIdentityMigration(func(c *coreapi.HCPOpenShiftCluster) {
-				// cosmos has identity (cache is stale)
-				c.Identity = &coreapi.ManagedServiceIdentity{
-					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
-						testIdentityResourceID: {
-							ClientID:    stringPtr(testClientID),
-							PrincipalID: stringPtr(testPrincipalID),
-						},
-					},
-				}
-				c.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators = map[string]*azcorearm.ResourceID{
-					"test-operator": metadataapi.Must(azcorearm.ParseResourceID(testIdentityResourceID)),
-				}
-			}),
-			expectCosmosGet:             true,
-			expectCSCall:                false,
-			expectCosmosUpdate:          false,
-			expectError:                 false,
-			expectedHasIdentity:         true,
-			expectedIdentityCount:       1,
-			expectedIdentityResourceIDs: []string{testIdentityResourceID},
-		},
-		{
-			name: "no work to do - identity already populated",
-			existingCluster: newTestClusterForIdentityMigration(func(c *coreapi.HCPOpenShiftCluster) {
-				c.Identity = &coreapi.ManagedServiceIdentity{
-					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{
-						testIdentityResourceID: {
-							ClientID:    stringPtr(testClientID),
-							PrincipalID: stringPtr(testPrincipalID),
-						},
-					},
-				}
-				c.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators = map[string]*azcorearm.ResourceID{
-					"test-operator": metadataapi.Must(azcorearm.ParseResourceID(testIdentityResourceID)),
-				}
-			}),
-			expectCosmosGet:             false,
-			expectCSCall:                false,
-			expectCosmosUpdate:          false,
-			expectError:                 false,
-			expectedHasIdentity:         true,
-			expectedIdentityCount:       1,
-			expectedIdentityResourceIDs: []string{testIdentityResourceID},
-		},
-		{
-			name:                  "error reading from cluster-service",
-			existingCluster:       newTestClusterForIdentityMigration(),
-			csError:               fmt.Errorf("connection refused"),
-			expectCosmosGet:       true,
-			expectCSCall:          true,
-			expectCosmosUpdate:    false,
-			expectError:           true,
-			expectedHasIdentity:   false,
-			expectedIdentityCount: 0,
-		},
-		{
-			name:                        "success - migrate identity when nil",
-			existingCluster:             newTestClusterForIdentityMigration(),
-			csCluster:                   buildCSClusterWithIdentity(testIdentityResourceID, testClientID, testPrincipalID),
-			expectCosmosGet:             true,
-			expectCSCall:                true,
-			expectCosmosUpdate:          true,
-			expectError:                 false,
-			expectedHasIdentity:         true,
-			expectedIdentityCount:       1,
-			expectedIdentityResourceIDs: []string{testIdentityResourceID},
-		},
-		{
-			name: "success - migrate identity when empty map",
-			existingCluster: newTestClusterForIdentityMigration(func(c *coreapi.HCPOpenShiftCluster) {
-				c.Identity = &coreapi.ManagedServiceIdentity{
-					UserAssignedIdentities: map[string]*coreapi.UserAssignedIdentity{},
-				}
-			}),
-			csCluster:                   buildCSClusterWithIdentity(testIdentityResourceID, testClientID, testPrincipalID),
-			expectCosmosGet:             true,
-			expectCSCall:                true,
-			expectCosmosUpdate:          true,
-			expectError:                 false,
-			expectedHasIdentity:         true,
-			expectedIdentityCount:       1,
-			expectedIdentityResourceIDs: []string{testIdentityResourceID},
-		},
-		{
-			name: "success - migrate identity when Identity is set but UserAssignedIdentities is nil",
-			existingCluster: newTestClusterForIdentityMigration(func(c *coreapi.HCPOpenShiftCluster) {
-				c.Identity = &coreapi.ManagedServiceIdentity{
-					Type: coreapi.ManagedServiceIdentityTypeUserAssigned,
-				}
-			}),
-			csCluster:                   buildCSClusterWithIdentity(testIdentityResourceID, testClientID, testPrincipalID),
-			expectCosmosGet:             true,
-			expectCSCall:                true,
-			expectCosmosUpdate:          true,
-			expectError:                 false,
-			expectedHasIdentity:         true,
-			expectedIdentityCount:       1,
-			expectedIdentityResourceIDs: []string{testIdentityResourceID},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			ctx := context.Background()
-			ctrl := gomock.NewController(t)
-			defer ctrl.Finish()
-
-			// Setup mock DB
-			mockResourcesDBClient := corecosmosstoragetesting.NewMockResourcesDBClient()
-
-			// Create the cluster in the mock DB (cosmos)
-			clusterCRUD := mockResourcesDBClient.HCPClusters(testSubscriptionID, testResourceGroupName)
-			_, err := clusterCRUD.Create(ctx, tc.existingCluster, nil)
-			require.NoError(t, err)
-
-			// Setup slice cluster lister (cache)
-			// If cachedCluster is nil, use the same as existingCluster
-			cachedCluster := tc.cachedCluster
-			if cachedCluster == nil {
-				cachedCluster = tc.existingCluster
-			}
-			sliceClusterLister := &corelistertesting.SliceClusterLister{
-				Clusters: []*coreapi.HCPOpenShiftCluster{cachedCluster},
-			}
-
-			// Setup mock CS client
-			mockCSClient := ocm.NewMockClusterServiceClientSpec(ctrl)
-
-			if tc.expectCSCall {
-				mockCSClient.EXPECT().
-					GetCluster(gomock.Any(), metadataapi.Must(metadataapi.NewInternalID(testClusterServiceIDStr))).
-					Return(tc.csCluster, tc.csError)
-			}
-
-			// Create syncer
-			syncer := &identityMigrationSyncer{
-				clusterLister:        sliceClusterLister,
-				resourcesDBClient:    mockResourcesDBClient,
-				clusterServiceClient: mockCSClient,
-			}
-
-			// Execute
-			key := controllerutils.HCPClusterKey{
-				SubscriptionID:    testSubscriptionID,
-				ResourceGroupName: testResourceGroupName,
-				HCPClusterName:    testClusterName,
-			}
-			err = syncer.SyncOnce(ctx, key)
-
-			if tc.expectError {
-				require.Error(t, err)
-			} else {
-				require.NoError(t, err)
-			}
-
-			// Verify the cluster state in Cosmos
-			updatedCluster, err := clusterCRUD.Get(ctx, testClusterName)
-			require.NoError(t, err)
-
-			if tc.expectedHasIdentity {
-				require.NotNil(t, updatedCluster.Identity)
-				assert.Len(t, updatedCluster.Identity.UserAssignedIdentities, tc.expectedIdentityCount)
-				for _, expectedID := range tc.expectedIdentityResourceIDs {
-					_, exists := updatedCluster.Identity.UserAssignedIdentities[expectedID]
-					assert.True(t, exists, "expected identity %s to exist", expectedID)
-				}
-			} else {
-				if updatedCluster.Identity != nil {
-					assert.Len(t, updatedCluster.Identity.UserAssignedIdentities, tc.expectedIdentityCount)
-				}
-			}
-		})
-	}
-}
-
-// newTestClusterForIdentityMigration creates a test HCPOpenShiftCluster with default values
-// for identity migration testing.
-func newTestClusterForIdentityMigration(opts ...func(*coreapi.HCPOpenShiftCluster)) *coreapi.HCPOpenShiftCluster {
-	cluster := newTestCluster(testClusterName, opts...)
-	cluster.Location = testLocation
-	return cluster
-}
-
-// buildCSClusterWithIdentity creates a mock Cluster Service cluster with managed identity information.
-func buildCSClusterWithIdentity(identityResourceID, clientID, principalID string) *arohcpv1alpha1.Cluster {
-	cluster, err := arohcpv1alpha1.NewCluster().
-		Azure(arohcpv1alpha1.NewAzure().
-			OperatorsAuthentication(arohcpv1alpha1.NewAzureOperatorsAuthentication().
-				ManagedIdentities(arohcpv1alpha1.NewAzureOperatorsAuthenticationManagedIdentities().
-					ControlPlaneOperatorsManagedIdentities(map[string]*arohcpv1alpha1.AzureControlPlaneManagedIdentityBuilder{
-						"test-operator": arohcpv1alpha1.NewAzureControlPlaneManagedIdentity().
-							ResourceID(identityResourceID).
-							ClientID(clientID).
-							PrincipalID(principalID),
-					}).
-					DataPlaneOperatorsManagedIdentities(make(map[string]*arohcpv1alpha1.AzureDataPlaneManagedIdentityBuilder)).
-					ManagedIdentitiesDataPlaneIdentityUrl("")))).
-		Console(arohcpv1alpha1.NewClusterConsole().URL(testConsoleURL)).
-		DNS(arohcpv1alpha1.NewDNS().BaseDomain(testBaseDomain)).
-		DomainPrefix(testBaseDomainPrefix).
-		Build()
-	if err != nil {
-		panic(err)
-	}
-	return cluster
-}
-
-func stringPtr(s string) *string {
-	return &s
-}
diff --git a/backend/pkg/utils/statusutils/inertia_test.go b/backend/pkg/utils/statusutils/inertia_test.go
index 97f6858184e..4302859c395 100644
--- a/backend/pkg/utils/statusutils/inertia_test.go
+++ b/backend/pkg/utils/statusutils/inertia_test.go
@@ -74,7 +74,7 @@ func TestInertiaConfig_Inertia(t *testing.T) {
 			overrides: []InertiaController{
 				{ControllerNameMatcher: regexp.MustCompile(`Migration$`), Duration: 90 * time.Second},
 			},
-			controllerName:   "IdentityMigration",
+			controllerName:   "FooControllerName",
 			expectedDuration: 90 * time.Second,
 		},
 		{
diff --git a/docs/cosmos-data-flow.md b/docs/cosmos-data-flow.md
index e441b0cdc13..2138918ac38 100644
--- a/docs/cosmos-data-flow.md
+++ b/docs/cosmos-data-flow.md
@@ -933,19 +933,35 @@ No Cosmos writes. Posts `NodePoolUpgradePolicy` to Cluster Service.
 | Read | ReadDesire (HostedCluster) | <ul><li>`Namespace` (HostedCluster namespace)</li><li>`Name` (HostedCluster name)</li></ul> |
 | **Write** | **`ServiceProviderCluster`** | <ul><li>**`Status.HostedClusterNamespace`** = HostedCluster namespace</li><li>**`Status.ControlPlaneNamespace`** = `<hostedClusterNamespace>-<hostedClusterName>` (dots replaced by dashes)</li></ul> |
 
-#### IdentityMigration
+#### ClusterIdentitySync
 
-**File:** [identity_migration.go](../backend/pkg/controllers/cluster/properties/identity_migration.go)
+**File:** [cluster_identity_sync.go](../backend/pkg/controllers/cluster/identity/cluster_identity_sync.go)
 **Trigger:** Cluster informer, 60-minute resync
 **Gate (NeedsWork on Cluster):**
-- `Cluster.ServiceProviderProperties.ClusterServiceID` != nil and non-empty
-- `Cluster.Identity` == nil, OR `len(Cluster.Identity.UserAssignedIdentities)` == 0, OR any entry has empty ClientID/PrincipalID, OR entries don't match `CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities`
+- `Cluster.ServiceProviderProperties.DeletionTimestamp` == nil, AND
+- `Cluster.Identity` != nil and `len(Cluster.Identity.UserAssignedIdentities)` > 0
 
 | | Object | Fields |
 |---|--------|--------|
-| Read | `HCPOpenShiftCluster` | <ul><li>`ServiceProviderProperties.ClusterServiceID` (NeedsWork: must not be nil and non-empty)</li><li>`Identity` (NeedsWork: must be nil, or `Identity.UserAssignedIdentities` empty, or entries incomplete)</li><li>`Identity.UserAssignedIdentities` (NeedsWork: each must have non-empty ClientID/PrincipalID)</li><li>`CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities` (NeedsWork: entries must match Identity map)</li></ul> |
-| Read | Cluster Service | <ul><li>`GetCluster` -> `GetClusterServiceUserAssignedIdentities`</li></ul> |
-| **Write** | **`HCPOpenShiftCluster`** | <ul><li>**`Identity.UserAssignedIdentities`** = migrated map from CS</li></ul> |
+| Read | `HCPOpenShiftCluster` | <ul><li>`ServiceProviderProperties.DeletionTimestamp` (NeedsWork)</li><li>`Identity.UserAssignedIdentities` (iterated; keys preserved with original casing)</li></ul> |
+| Read | `ServiceProviderCluster` | <ul><li>`Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities` (lookup by lowercased Identity map key)</li><li>`Status.MSIManagedIdentities.ServiceManagedIdentity` (lookup by lowercased ResourceID)</li></ul> |
+| **Write** | **`HCPOpenShiftCluster`** | <ul><li>**`Identity.UserAssignedIdentities[key].ClientID` / `PrincipalID`** = from SPC when a lowercased match exists; keys absent from SPC are left unchanged</li></ul> |
+
+#### FetchMSIIdentitiesInfo
+
+**File:** [fetch_msi_identities_info.go](../backend/pkg/controllers/cluster/identity/fetch_msi_identities_info.go)
+**Trigger:** Cluster informer, 1-minute resync
+**Gate (needsWork):**
+- `Cluster.ServiceProviderProperties.DeletionTimestamp` == nil
+- `ServiceProviderCluster.Status.MSIManagedIdentities.EarliestRecheckTime` is nil or in the past, OR
+- stored MSI identities on SPC no longer match `OperatorsAuthentication` (control-plane operator bindings / service managed identity resource ID), in which case EarliestRecheckTime is ignored
+
+| | Object | Fields |
+|---|--------|--------|
+| Read | `HCPOpenShiftCluster` | <ul><li>`ServiceProviderProperties.DeletionTimestamp` (NeedsWork)</li><li>`ServiceProviderProperties.ManagedIdentitiesDataPlaneIdentityURL`</li><li>`CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators`</li><li>`CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity`</li></ul> |
+| Read | `ServiceProviderCluster` | <ul><li>`Status.MSIManagedIdentities.EarliestRecheckTime` (NeedsWork)</li><li>`Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities` (deep-equal before write)</li><li>`Status.MSIManagedIdentities.ServiceManagedIdentity` (deep-equal before write)</li></ul> |
+| Read | Managed Identities Data Plane | <ul><li>`GetUserAssignedIdentitiesCredentials`</li></ul> |
+| **Write** | **`ServiceProviderCluster`** | <ul><li>**`Status.MSIManagedIdentities.ControlPlaneOperatorsIdentities`** = map keyed by lowercased resource ID (`ResourceID`, `ClientID`, `PrincipalID`)</li><li>**`Status.MSIManagedIdentities.ServiceManagedIdentity`** = `ResourceID`, `ClientID`, `PrincipalID`</li><li>**`Status.MSIManagedIdentities.EarliestRecheckTime`** = now + jittered 12h interval</li></ul> |
 
 ---
 
@@ -1327,7 +1343,7 @@ Each entry links to every actor that writes the field.
 |-------|------|
 | [Frontend: PUT Cluster (Create)](#put-cluster-create) | Rebuilt via `completeClusterIdentity` |
 | [Frontend: PUT/PATCH Cluster (Update)](#put-cluster-update) | Rebuilt via `completeClusterIdentity` with old data |
-| [IdentityMigration](#identitymigration) | Migrated from CS for clusters with incomplete identity |
+| [ClusterIdentitySync](#clusteridentitysync) | Keeps ClientID/PrincipalID on existing Identity keys in sync with ServiceProviderCluster.Status.MSIManagedIdentities |
 
 ### `HCPOpenShiftCluster.ServiceProviderProperties.DeletionTimestamp`
 
@@ -1465,6 +1481,14 @@ Single writer, but tracks the namespace containing the HostedCluster CR and user
 
 Single writer, but tracks the namespace containing control plane pods (etcd, kube-apiserver, etc.) on the management cluster.
 
+### `ServiceProviderCluster.Status.MSIManagedIdentities`
+
+| Actor | When |
+|-------|------|
+| [FetchMSIIdentitiesInfo](#fetchmsiidentitiesinfo) | Sets ControlPlaneOperatorsIdentities (lowercased resource ID keys), ServiceManagedIdentity, and EarliestRecheckTime from Managed Identities Data Plane |
+
+Single writer. Read by [ClusterIdentitySync](#clusteridentitysync) to populate `HCPOpenShiftCluster.Identity.UserAssignedIdentities`.
+
 ### `ServiceProviderCluster.Status.Validations`
 
 | Actor | When |
diff --git a/docs/resource-creation.mm b/docs/resource-creation.mm
index cec723a7969..255004729ba 100644
--- a/docs/resource-creation.mm
+++ b/docs/resource-creation.mm
@@ -144,7 +144,7 @@
 <node CREATED="1742324517000" ID="ID_1000117" MODIFIED="1742324517000" TEXT="other controllers">
 <node CREATED="1742324518000" ID="ID_1000118" MODIFIED="1742324518000" TEXT="mismatch controllers - reconcile Cosmos vs Cluster Service state"/>
 <node CREATED="1742324519000" ID="ID_1000119" MODIFIED="1742324519000" TEXT="upgrade controllers - trigger control plane upgrades"/>
-<node CREATED="1742324520000" ID="ID_1000120" MODIFIED="1742324520000" TEXT="cluster properties sync - identity migration, customer properties"/>
+<node CREATED="1742324520000" ID="ID_1000120" MODIFIED="1742324520000" TEXT="cluster identity sync - MSI identities, cluster identity from SPC"/>
 <node CREATED="1742324521000" ID="ID_1000121" MODIFIED="1742324521000" TEXT="Maestro bundle controllers - create/delete readonly resource bundles"/>
 </node>
 </node>
diff --git a/internal/api/coreapi/types_cluster.go b/internal/api/coreapi/types_cluster.go
index bc6f91b9f86..866e2c03c23 100644
--- a/internal/api/coreapi/types_cluster.go
+++ b/internal/api/coreapi/types_cluster.go
@@ -36,7 +36,7 @@ type HCPOpenShiftCluster struct {
 	CustomerProperties HCPOpenShiftClusterCustomerProperties `json:"customerProperties,omitempty"`
 	// Written by: Frontend PUT/PATCH/DELETE Cluster, all Operation*Cluster controllers, ClusterPropertiesSync, ClusterClusterServiceCreate, ClusterDeletion* controllers, CreateBillingDoc
 	ServiceProviderProperties HCPOpenShiftClusterServiceProviderProperties `json:"serviceProviderProperties,omitempty"`
-	// Written by: Frontend PUT/PATCH Cluster (Create/Update), IdentityMigration
+	// Written by: Frontend PUT/PATCH Cluster (Create/Update), ClusterIdentitySync
 	Identity *ManagedServiceIdentity `json:"identity,omitempty"`
 	// Written by: ClusterDegradedAggregator, ClusterRequirementsValidAggregator
 	Status HCPOpenShiftClusterStatus `json:"status"`
diff --git a/internal/api/coreapi/types_serviceprovider_cluster.go b/internal/api/coreapi/types_serviceprovider_cluster.go
index cf32c5ea39e..f526524952f 100644
--- a/internal/api/coreapi/types_serviceprovider_cluster.go
+++ b/internal/api/coreapi/types_serviceprovider_cluster.go
@@ -232,6 +232,102 @@ type ServiceProviderClusterStatus struct {
 	// AzureResources tracks the lifecycle of Azure resources associated with
 	// the cluster, including deny assignments and the managed resource group.
 	AzureResources AzureResources `json:"azureResources,omitempty"`
+
+	// MSIManagedIdentities tracks resolved ClientID/PrincipalID for
+	// the Managed Service Identity (MSI) based Azure User-Assigned Managed Identities
+	// associated to the cluster. Those are the cluster's control plane operators and
+	// the cluster's service managed identity.
+	// A cluster's control plane operator is a kubernetes operator associated to
+	// the cluster that runs in the cluster's control plane. For example,
+	// the Cluster's CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators
+	// map contains (and is not limited to) the set of required control plane operators associated to a Cluster.
+	// The cluster's service managed identity is used to read and modify
+	// customer-provided Azure resources (for example the cluster subnet),
+	// subject to the permissions granted to that identity.
+	// MSI-based user-assigned managed identities are the identities defined in
+	// the Cluster's `identity` section. Credentials for those identities can be
+	// obtained from Microsoft's Managed Identities Data Plane service.
+	// In ARO-HCP environments where Microsoft's Managed Identities Data Plane
+	// service is unavailable, a fake Managed Identities Data Plane client is
+	// used. That client always returns the same identity metadata and
+	// credentials, regardless of which identity is requested. The returned
+	// values belong to the "MI Mock" identity, so the ClientID and PrincipalID
+	// stored for each entry here will not match that entry's ResourceID key,
+	// nor the real ClientID/PrincipalID of the corresponding identity in the
+	// Cluster's `identity` section.
+	// Additionally, this also tracks when Azure should next be re-queried for that info.
+	// Written by: FetchMSIIdentitiesInfo
+	MSIManagedIdentities ServiceProviderClusterMSIManagedIdentities `json:"msiManagedIdentities,omitempty"`
+}
+
+// ServiceProviderClusterMSIManagedIdentities holds Managed Service Identity (MSI)
+// based identity metadata resolved by FetchMSIIdentitiesInfo and consumed by ClusterIdentitySync to
+// populate HCPOpenShiftCluster.Identity.UserAssignedIdentities.
+type ServiceProviderClusterMSIManagedIdentities struct {
+	// EarliestRecheckTime is the earliest time at which the controller
+	// should re-query Azure for ClientID/PrincipalID of ControlPlaneOperatorsIdentities
+	// and ServiceManagedIdentity.
+	// Nil means recheck immediately.
+	// The same recheck time applies across all entries in ControlPlaneOperatorsIdentities
+	// and ServiceManagedIdentity.
+	// This allows the controller to avoid repeatedly hitting an Azure API to
+	// recheck that the desired state is true.
+	// Controllers should set this field with substantial jitter: without another
+	// concern, jitter of 50% is considered normal so that any storms are quickly
+	// dissipated. Additionally, long recheck times are recommended for resources
+	// outside of their active phases. Order of at least six hours is, with
+	// durations up to 24 hours considered normal.
+	// Written by: FetchMSIIdentitiesInfo
+	EarliestRecheckTime *metav1.Time `json:"earliestRecheckTime,omitempty"`
+	// ControlPlaneOperatorsIdentities is a map containing resolved ClientID/PrincipalID
+	// for Managed Service Identity (MSI) based Azure User-Assigned Managed Identities
+	// used by the cluster's control plane operators. The key is the fully lowercased
+	// Azure Resource ID of the identity. Which operators reference each identity is
+	// tracked on Cluster.CustomerProperties, not here. Multiple operators may share
+	// one identity entry.
+	// Written by: FetchMSIIdentitiesInfo
+	ControlPlaneOperatorsIdentities map[string]*ServiceProviderClusterControlPlaneOperatorIdentity `json:"controlPlaneOperatorsIdentities,omitempty"`
+	// ServiceManagedIdentity holds resolved ClientID/PrincipalID for the cluster's
+	// service managed identity.
+	// Written by: FetchMSIIdentitiesInfo
+	ServiceManagedIdentity *ServiceProviderClusterServiceManagedIdentity `json:"serviceManagedIdentity,omitempty"`
+}
+
+// ServiceProviderClusterControlPlaneOperatorIdentity is the resolved metadata for a
+// single Managed Service Identity (MSI) based Azure User-Assigned Managed Identity
+// used by one or more control plane operators.
+// Which operators reference this identity is tracked on
+// Cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators.
+type ServiceProviderClusterControlPlaneOperatorIdentity struct {
+	// ResourceID is the Azure Resource ID of the Azure User Assigned Managed Identity.
+	// Its value comes from the Cluster's CustomerProperties.
+	// The ControlPlaneOperatorsIdentities map key is the fully lowercased form
+	// of this ID used for lookups.
+	ResourceID *azcorearm.ResourceID `json:"resourceId,omitempty"`
+	// ClientID is Client ID of the Azure User Assigned Managed Identity represented by ResourceID.
+	// Fetched from Azure and written here by the FetchMSIIdentitiesInfo.
+	// It may be be nil or empty.
+	ClientID *string `json:"clientId,omitempty"`
+	// PrincipalID Principal ID of the Azure User Assigned Managed Identity represented by ResourceID.
+	// Fetched from Azure and written here by the FetchMSIIdentitiesInfo.
+	// It may be be nil or empty.
+	PrincipalID *string `json:"principalId,omitempty"`
+}
+
+// ServiceProviderClusterServiceManagedIdentity is the resolved metadata for the
+// cluster's service managed identity.
+type ServiceProviderClusterServiceManagedIdentity struct {
+	// ResourceID is the Azure Resource ID of the Azure User Assigned Managed Identity that is associated to the cluster's Service Managed Identity.
+	// Its value comes from the Cluster's CustomerProperties. Its value comes from the Cluster's CustomerProperties.
+	ResourceID *azcorearm.ResourceID `json:"resourceId,omitempty"`
+	// ClientID is Client ID of the Azure User Assigned Managed Identity represented by ResourceID.
+	// Fetched from Azure and written here by the FetchMSIIdentitiesInfo.
+	// It may be be nil or empty.
+	ClientID *string `json:"clientId,omitempty"`
+	// PrincipalID Principal ID of the Azure User Assigned Managed Identity represented by ResourceID.
+	// Fetched from Azure and written here by the FetchMSIIdentitiesInfo.
+	// It may be be nil or empty.
+	PrincipalID *string `json:"principalId,omitempty"`
 }
 
 // AzureResources groups the Azure resource references associated with a cluster.
diff --git a/internal/api/coreapi/zz_generated.deepcopy.go b/internal/api/coreapi/zz_generated.deepcopy.go
index 335fa228d8f..9607d501546 100644
--- a/internal/api/coreapi/zz_generated.deepcopy.go
+++ b/internal/api/coreapi/zz_generated.deepcopy.go
@@ -1991,6 +1991,36 @@ func (in *ServiceProviderCluster) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ServiceProviderClusterControlPlaneOperatorIdentity) DeepCopyInto(out *ServiceProviderClusterControlPlaneOperatorIdentity) {
+	*out = *in
+	if in.ResourceID != nil {
+		in, out := &in.ResourceID, &out.ResourceID
+		*out = DeepCopyResourceID(*in)
+	}
+	if in.ClientID != nil {
+		in, out := &in.ClientID, &out.ClientID
+		*out = new(string)
+		**out = **in
+	}
+	if in.PrincipalID != nil {
+		in, out := &in.PrincipalID, &out.PrincipalID
+		*out = new(string)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceProviderClusterControlPlaneOperatorIdentity.
+func (in *ServiceProviderClusterControlPlaneOperatorIdentity) DeepCopy() *ServiceProviderClusterControlPlaneOperatorIdentity {
+	if in == nil {
+		return nil
+	}
+	out := new(ServiceProviderClusterControlPlaneOperatorIdentity)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ServiceProviderClusterList) DeepCopyInto(out *ServiceProviderClusterList) {
 	*out = *in
@@ -2024,6 +2054,76 @@ func (in *ServiceProviderClusterList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ServiceProviderClusterMSIManagedIdentities) DeepCopyInto(out *ServiceProviderClusterMSIManagedIdentities) {
+	*out = *in
+	if in.EarliestRecheckTime != nil {
+		in, out := &in.EarliestRecheckTime, &out.EarliestRecheckTime
+		*out = (*in).DeepCopy()
+	}
+	if in.ControlPlaneOperatorsIdentities != nil {
+		in, out := &in.ControlPlaneOperatorsIdentities, &out.ControlPlaneOperatorsIdentities
+		*out = make(map[string]*ServiceProviderClusterControlPlaneOperatorIdentity, len(*in))
+		for key, val := range *in {
+			var outVal *ServiceProviderClusterControlPlaneOperatorIdentity
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = new(ServiceProviderClusterControlPlaneOperatorIdentity)
+				(*in).DeepCopyInto(*out)
+			}
+			(*out)[key] = outVal
+		}
+	}
+	if in.ServiceManagedIdentity != nil {
+		in, out := &in.ServiceManagedIdentity, &out.ServiceManagedIdentity
+		*out = new(ServiceProviderClusterServiceManagedIdentity)
+		(*in).DeepCopyInto(*out)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceProviderClusterMSIManagedIdentities.
+func (in *ServiceProviderClusterMSIManagedIdentities) DeepCopy() *ServiceProviderClusterMSIManagedIdentities {
+	if in == nil {
+		return nil
+	}
+	out := new(ServiceProviderClusterMSIManagedIdentities)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ServiceProviderClusterServiceManagedIdentity) DeepCopyInto(out *ServiceProviderClusterServiceManagedIdentity) {
+	*out = *in
+	if in.ResourceID != nil {
+		in, out := &in.ResourceID, &out.ResourceID
+		*out = DeepCopyResourceID(*in)
+	}
+	if in.ClientID != nil {
+		in, out := &in.ClientID, &out.ClientID
+		*out = new(string)
+		**out = **in
+	}
+	if in.PrincipalID != nil {
+		in, out := &in.PrincipalID, &out.PrincipalID
+		*out = new(string)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceProviderClusterServiceManagedIdentity.
+func (in *ServiceProviderClusterServiceManagedIdentity) DeepCopy() *ServiceProviderClusterServiceManagedIdentity {
+	if in == nil {
+		return nil
+	}
+	out := new(ServiceProviderClusterServiceManagedIdentity)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ServiceProviderClusterSpec) DeepCopyInto(out *ServiceProviderClusterSpec) {
 	*out = *in
@@ -2116,6 +2216,7 @@ func (in *ServiceProviderClusterStatus) DeepCopyInto(out *ServiceProviderCluster
 		**out = **in
 	}
 	in.AzureResources.DeepCopyInto(&out.AzureResources)
+	in.MSIManagedIdentities.DeepCopyInto(&out.MSIManagedIdentities)
 	return
 }