diff --git a/pkg/operator/encryption/controllers.go b/pkg/operator/encryption/controllers.go
index 0dc0d3199c..ead7f67869 100644
--- a/pkg/operator/encryption/controllers.go
+++ b/pkg/operator/encryption/controllers.go
@@ -123,6 +123,19 @@ func NewControllers(
 			encryptionSecretSelector,
 			eventRecorder,
 		),
+		controllers.NewEncryptionRotationController(
+			component,
+			provider,
+			deployer,
+			encryptionEnabledChecker.PreconditionFulfilled,
+			migrator,
+			operatorClient,
+			apiServerInformer,
+			kubeInformersForNamespaces,
+			secretsClient,
+			encryptionSecretSelector,
+			eventRecorder,
+		),
 	}, nil
 }
 
diff --git a/pkg/operator/encryption/controllers/encryption_rotation_controller.go b/pkg/operator/encryption/controllers/encryption_rotation_controller.go
new file mode 100644
index 0000000000..8ccd890797
--- /dev/null
+++ b/pkg/operator/encryption/controllers/encryption_rotation_controller.go
@@ -0,0 +1,293 @@
+package controllers
+
+import (
+	"context"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/util/retry"
+	"k8s.io/klog/v2"
+
+	configv1informers "github.com/openshift/client-go/config/informers/externalversions/config/v1"
+
+	"github.com/openshift/library-go/pkg/controller/factory"
+	"github.com/openshift/library-go/pkg/operator/encryption/controllers/migrators"
+	"github.com/openshift/library-go/pkg/operator/encryption/encryptionstatus"
+	"github.com/openshift/library-go/pkg/operator/encryption/secrets"
+	"github.com/openshift/library-go/pkg/operator/encryption/statemachine"
+	"github.com/openshift/library-go/pkg/operator/events"
+	operatorv1helpers "github.com/openshift/library-go/pkg/operator/v1helpers"
+)
+
+const (
+	encryptionRotationConvergenceDelay = 5 * time.Minute
+)
+
+// encryptionRotationController orchestrates KMS KEK rotation by resetting migration state on the
+// encryption key secret and recording rotation progress on operator status.
+type encryptionRotationController struct {
+	instanceName             string
+	controllerInstanceName   string
+	operatorClient           operatorv1helpers.OperatorClient
+	encryptionSecretSelector metav1.ListOptions
+	secretClient             corev1client.SecretsGetter
+	deployer                 statemachine.Deployer
+	migrator                 migrators.Migrator
+	provider                 Provider
+	preconditionsFulfilledFn preconditionsFulfilled
+}
+
+func NewEncryptionRotationController(
+	instanceName string,
+	provider Provider,
+	deployer statemachine.Deployer,
+	preconditionsFulfilledFn preconditionsFulfilled,
+	migrator migrators.Migrator,
+	operatorClient operatorv1helpers.OperatorClient,
+	apiServerConfigInformer configv1informers.APIServerInformer,
+	kubeInformersForNamespaces operatorv1helpers.KubeInformersForNamespaces,
+	secretClient corev1client.SecretsGetter,
+	encryptionSecretSelector metav1.ListOptions,
+	eventRecorder events.Recorder,
+) factory.Controller {
+	c := &encryptionRotationController{
+		instanceName:             instanceName,
+		controllerInstanceName:   factory.ControllerInstanceName(instanceName, "EncryptionRotation"),
+		operatorClient:           operatorClient,
+		encryptionSecretSelector: encryptionSecretSelector,
+		secretClient:             secretClient,
+		deployer:                 deployer,
+		migrator:                 migrator,
+		provider:                 provider,
+		preconditionsFulfilledFn: preconditionsFulfilledFn,
+	}
+
+	return factory.New().ResyncEvery(time.Minute).WithSync(c.sync).WithControllerInstanceName(c.controllerInstanceName).WithInformers(
+		operatorClient.Informer(),
+		kubeInformersForNamespaces.InformersFor("openshift-config-managed").Core().V1().Secrets().Informer(),
+		apiServerConfigInformer.Informer(),
+		deployer,
+	).ToController(
+		c.controllerInstanceName,
+		eventRecorder.WithComponentSuffix("encryption-rotation-controller"),
+	)
+}
+
+func (c *encryptionRotationController) sync(ctx context.Context, syncCtx factory.SyncContext) error {
+	if ready, err := shouldRunEncryptionController(c.operatorClient, c.preconditionsFulfilledFn, c.provider.ShouldRunEncryptionControllers); err != nil || !ready {
+		return err
+	}
+	return c.reconcile(ctx)
+}
+
+func (c *encryptionRotationController) reconcile(ctx context.Context) error {
+	currentConfig, _, encryptionSecrets, _, err := statemachine.GetEncryptionConfigAndState(
+		ctx, c.deployer, c.secretClient, c.encryptionSecretSelector, c.provider.EncryptedGRs(),
+	)
+	if err != nil {
+		return err
+	}
+
+	if currentConfig == nil || !currentConfig.UsesKMS() {
+		return nil
+	}
+
+	_, operatorStatus, _, err := c.operatorClient.GetOperatorState()
+	if err != nil {
+		return err
+	}
+
+	healthReports := encryptionstatus.HealthReportsFromOperatorStatus(operatorStatus)
+	rotations, err := encryptionstatus.KeyRotationStatusFromOperatorStatus(operatorStatus)
+	if err != nil {
+		return err
+	}
+
+	encryptedGRs := currentConfig.EncryptedGroupResources()
+	klog.Infof("%s: reconciling KMS key rotation (%d plugin(s), %d health report(s), %d rotation entr(ies))",
+		c.instanceName, len(currentConfig.KMSPlugins), len(healthReports), len(rotations))
+	for keyID := range currentConfig.KMSPlugins {
+		writeKeySecret := secrets.FindKeySecret(encryptionSecrets, c.instanceName, keyID)
+		if writeKeySecret == nil {
+			klog.Infof("%s: no write key secret for KMS plugin keyID %q, skipping", c.instanceName, keyID)
+			continue
+		}
+
+		rotations, err = c.reconcileKMSPlugin(
+			ctx, keyID, encryptedGRs, writeKeySecret, healthReports, rotations,
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	if err := c.persistRotations(ctx, rotations); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *encryptionRotationController) reconcileKMSPlugin(
+	ctx context.Context,
+	keyID string,
+	encryptedGRs []schema.GroupResource,
+	writeKeySecret *corev1.Secret,
+	healthReports []encryptionstatus.KMSPluginHealthReport,
+	rotations []encryptionstatus.KMSPluginRotationStatus,
+) ([]encryptionstatus.KMSPluginRotationStatus, error) {
+	// Check KEK convergence across all nodes for this keyID.
+	convergedKEKID, converged := encryptionstatus.ConvergedKEKForKeyID(healthReports, keyID)
+	if !converged {
+		klog.Infof("%s: KEK not yet converged for keyID %q, waiting for all nodes to report the same kekID", c.instanceName, keyID)
+		return rotations, nil
+	}
+	klog.Infof("%s: KEK converged for keyID %q: kekID=%q", c.instanceName, keyID, convergedKEKID)
+
+	// If the converged kekID matches the last completed rotation, we are in steady state.
+	lastCompleted, hasCompleted := encryptionstatus.LatestCompletedRotationForKeyID(rotations, keyID)
+	if hasCompleted && convergedKEKID == lastCompleted.KEKID {
+		klog.V(4).Infof("%s: converged kekID %q matches last completed rotation for keyID %q, nothing to do", c.instanceName, convergedKEKID, keyID)
+		return rotations, nil
+	}
+
+	// Ensure an open rotation entry exists with discoveryTime for the converged kekID.
+	now := metav1.Now()
+	rotations, openIdx := encryptionstatus.GetOrCreateOpenRotation(rotations, keyID, convergedKEKID, now)
+	klog.Infof("%s: tracking open rotation for keyID=%q kekID=%q (discoveryTime=%s)",
+		c.instanceName, keyID, convergedKEKID, rotations[openIdx].DiscoveryTime.Format(time.RFC3339))
+
+	// Mirror migration finish time from the write key secret annotations when all
+	// encrypted group resources have been migrated by the migration controller.
+	migrated, err := encryptionstatus.AllEncryptedGRsMigrated(writeKeySecret, encryptedGRs)
+	if err != nil {
+		return rotations, err
+	}
+	rotations = mirrorMigrationFinish(c.instanceName, rotations, openIdx, migrated, writeKeySecret)
+
+	// Bootstrap: if no prior completed rotation exists, this is the initial provider
+	// migration driven by the migration controller. We only track convergence and mirror
+	// the finish time — never prune migrations or clear annotations.
+	if !hasCompleted {
+		klog.Infof("%s: bootstrap for keyID %q — no prior completed rotation, waiting for initial migration to complete", c.instanceName, keyID)
+		return rotations, nil
+	}
+
+	// From here on a KEK change was detected (convergedKEKID != lastCompleted.KEKID).
+	// Check guards before starting a storage re-migration.
+	entry := rotations[openIdx]
+
+	if entry.MigrationStartTime != nil {
+		klog.Infof("%s: rotation for kekID %q already started at %s, waiting for migration to complete",
+			c.instanceName, convergedKEKID, entry.MigrationStartTime.Format(time.RFC3339))
+		return rotations, nil
+	}
+
+	if entry.DiscoveryTime != nil && time.Since(entry.DiscoveryTime.Time) < encryptionRotationConvergenceDelay {
+		remaining := encryptionRotationConvergenceDelay - time.Since(entry.DiscoveryTime.Time)
+		klog.Infof("%s: rotation for kekID %q waiting for convergence delay (%s remaining)",
+			c.instanceName, convergedKEKID, remaining.Round(time.Second))
+		return rotations, nil
+	}
+
+	// All guards passed — start the rotation by pruning existing storage version
+	// migrations and clearing migration annotations on the write key secret so the
+	// migration controller picks up the work again.
+	klog.Infof("%s: starting storage re-migration for keyID=%q: kekID changed from %q to %q",
+		c.instanceName, keyID, lastCompleted.KEKID, convergedKEKID)
+	if err := c.startRotation(ctx, encryptedGRs, writeKeySecret); err != nil {
+		return rotations, err
+	}
+
+	rotations = encryptionstatus.SetMigrationStartTime(rotations, openIdx, now)
+	klog.Infof("%s: rotation migration started for kekID %q at %s", c.instanceName, convergedKEKID, now.Format(time.RFC3339))
+	return rotations, nil
+}
+
+func (c *encryptionRotationController) startRotation(ctx context.Context, encryptedGRs []schema.GroupResource, secret *corev1.Secret) error {
+	for _, gr := range encryptedGRs {
+		klog.Infof("%s: pruning storage migration for %s", c.instanceName, gr)
+		if err := c.migrator.PruneMigration(gr); err != nil {
+			return err
+		}
+	}
+	klog.Infof("%s: clearing migration annotations on secret %s/%s", c.instanceName, secret.Namespace, secret.Name)
+	return c.clearMigrationAnnotations(ctx, secret)
+}
+
+func mirrorMigrationFinish(
+	instanceName string,
+	rotations []encryptionstatus.KMSPluginRotationStatus,
+	idx int,
+	migrated bool,
+	secret *corev1.Secret,
+) []encryptionstatus.KMSPluginRotationStatus {
+	if !migrated || idx < 0 || idx >= len(rotations) || rotations[idx].MigrationFinishTime != nil {
+		return rotations
+	}
+	finish, ok := migrationFinishTimeFromSecret(secret)
+	if !ok {
+		return rotations
+	}
+	entry := rotations[idx]
+	klog.Infof("%s: mirroring migrationFinishTime for keyID %q kekID %q from secret %s/%s at %s",
+		instanceName, entry.KeyID, entry.KEKID, secret.Namespace, secret.Name, finish.Format(time.RFC3339))
+	return encryptionstatus.SetMigrationFinishTime(rotations, idx, finish)
+}
+
+func migrationFinishTimeFromSecret(secret *corev1.Secret) (metav1.Time, bool) {
+	if secret == nil || secret.Annotations == nil {
+		return metav1.Time{}, false
+	}
+	raw, ok := secret.Annotations[secrets.EncryptionSecretMigratedTimestamp]
+	if !ok || raw == "" {
+		return metav1.Time{}, false
+	}
+	ts, err := time.Parse(time.RFC3339, raw)
+	if err != nil {
+		klog.Warningf("ignoring invalid %s annotation on secret %s/%s: %v",
+			secrets.EncryptionSecretMigratedTimestamp, secret.Namespace, secret.Name, err)
+		return metav1.Time{}, false
+	}
+	return metav1.NewTime(ts), true
+}
+
+func (c *encryptionRotationController) clearMigrationAnnotations(ctx context.Context, secret *corev1.Secret) error {
+	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+		current, err := c.secretClient.Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+		if current.Annotations == nil {
+			return nil
+		}
+		changed := false
+		if _, ok := current.Annotations[secrets.EncryptionSecretMigratedTimestamp]; ok {
+			delete(current.Annotations, secrets.EncryptionSecretMigratedTimestamp)
+			changed = true
+		}
+		if _, ok := current.Annotations[secrets.EncryptionSecretMigratedResources]; ok {
+			delete(current.Annotations, secrets.EncryptionSecretMigratedResources)
+			changed = true
+		}
+		if !changed {
+			return nil
+		}
+		_, err = c.secretClient.Secrets(current.Namespace).Update(ctx, current, metav1.UpdateOptions{})
+		return err
+	})
+}
+
+func (c *encryptionRotationController) persistRotations(ctx context.Context, rotations []encryptionstatus.KMSPluginRotationStatus) error {
+	_, updated, err := operatorv1helpers.UpdateStatus(ctx, c.operatorClient, encryptionstatus.SetKeyRotationStatusCondition(rotations))
+	if err != nil {
+		return err
+	}
+	if updated {
+		klog.Infof("%s: updated key rotation status (%d entries)", c.instanceName, len(rotations))
+	}
+	return nil
+}
diff --git a/pkg/operator/encryption/encryptiondata/config.go b/pkg/operator/encryption/encryptiondata/config.go
index 40a855a81f..cf24ab7d84 100644
--- a/pkg/operator/encryption/encryptiondata/config.go
+++ b/pkg/operator/encryption/encryptiondata/config.go
@@ -36,6 +36,25 @@ func (c *Config) HasEncryptionConfiguration() bool {
 	return c != nil && c.Encryption != nil
 }
 
+// UsesKMS returns whether the deployed encryption configuration includes KMS plugins.
+func (c *Config) UsesKMS() bool {
+	return c != nil && len(c.KMSPlugins) > 0
+}
+
+// EncryptedGroupResources returns group resources from the deployed encryption configuration.
+func (c *Config) EncryptedGroupResources() []schema.GroupResource {
+	if !c.HasEncryptionConfiguration() {
+		return nil
+	}
+	grs := make([]schema.GroupResource, 0, len(c.Encryption.Resources))
+	for _, resourceConfig := range c.Encryption.Resources {
+		for _, resource := range resourceConfig.Resources {
+			grs = append(grs, schema.ParseGroupResource(resource))
+		}
+	}
+	return grs
+}
+
 // FromEncryptionState converts encryption state to Config.
 func FromEncryptionState(encryptionState map[schema.GroupResource]state.GroupResourceState) (*Config, error) {
 	resourceConfigs := make([]apiserverconfigv1.ResourceConfiguration, 0, len(encryptionState))
diff --git a/pkg/operator/encryption/encryptionstatus/convergence.go b/pkg/operator/encryption/encryptionstatus/convergence.go
new file mode 100644
index 0000000000..d875455532
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/convergence.go
@@ -0,0 +1,50 @@
+package encryptionstatus
+
+import (
+	"strings"
+)
+
+// KEKByKeyID groups observed kekIds per plugin keyId across health reports.
+// Only healthy reports with non-empty keyId and kekId are included.
+func KEKByKeyID(reports []KMSPluginHealthReport) map[string][]string {
+	result := map[string][]string{}
+	for _, report := range reports {
+		if !isHealthyReport(report) {
+			continue
+		}
+		result[report.KeyID] = append(result[report.KeyID], report.KEKID)
+	}
+	return result
+}
+
+// ConvergedKEKForKeyID returns the unanimous kekId for keyID when every healthy report for that keyId agrees.
+func ConvergedKEKForKeyID(reports []KMSPluginHealthReport, keyID string) (kekID string, ok bool) {
+	if keyID == "" {
+		return "", false
+	}
+
+	byKeyID := KEKByKeyID(reports)
+	kekIDs, found := byKeyID[keyID]
+	if !found || len(kekIDs) == 0 {
+		return "", false
+	}
+
+	uniq := map[string]struct{}{}
+	for _, id := range kekIDs {
+		if id == "" {
+			return "", false
+		}
+		uniq[id] = struct{}{}
+	}
+	if len(uniq) != 1 {
+		return "", false
+	}
+	for id := range uniq {
+		return id, true
+	}
+	return "", false
+}
+
+func isHealthyReport(report KMSPluginHealthReport) bool {
+	return strings.EqualFold(report.Status, "healthy")
+}
diff --git a/pkg/operator/encryption/encryptionstatus/convergence_test.go b/pkg/operator/encryption/encryptionstatus/convergence_test.go
new file mode 100644
index 0000000000..489d51d7c1
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/convergence_test.go
@@ -0,0 +1,34 @@
+package encryptionstatus
+
+import "testing"
+
+func TestConvergedKEKForKeyID(t *testing.T) {
+	reports := []KMSPluginHealthReport{
+		{KeyID: "1", NodeName: "master-0", KEKID: "kek-a", Status: "healthy"},
+		{KeyID: "1", NodeName: "master-1", KEKID: "kek-a", Status: "healthy"},
+		{KeyID: "2", NodeName: "master-0", KEKID: "kek-b", Status: "healthy"},
+		{KeyID: "2", NodeName: "master-1", KEKID: "kek-c", Status: "healthy"},
+	}
+
+	kekID, ok := ConvergedKEKForKeyID(reports, "1")
+	if !ok || kekID != "kek-a" {
+		t.Fatalf("expected converged kek-a for key 1, got %q ok=%v", kekID, ok)
+	}
+
+	_, ok = ConvergedKEKForKeyID(reports, "2")
+	if ok {
+		t.Fatal("expected key 2 to be divergent")
+	}
+}
+
+func TestKEKByKeyIDIgnoresUnhealthy(t *testing.T) {
+	reports := []KMSPluginHealthReport{
+		{KeyID: "1", NodeName: "master-0", KEKID: "kek-a", Status: "healthy"},
+		{KeyID: "1", NodeName: "master-1", KEKID: "kek-b", Status: "healthy"},
+		{KeyID: "1", NodeName: "master-2", KEKID: "kek-a", Status: "unhealthy"},
+	}
+	_, ok := ConvergedKEKForKeyID(reports, "1")
+	if ok {
+		t.Fatal("expected unhealthy node to be ignored so healthy nodes still diverge")
+	}
+}
diff --git a/pkg/operator/encryption/encryptionstatus/migration.go b/pkg/operator/encryption/encryptionstatus/migration.go
new file mode 100644
index 0000000000..1fe6e57c3a
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/migration.go
@@ -0,0 +1,47 @@
+package encryptionstatus
+
+import (
+	"encoding/json"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/openshift/library-go/pkg/operator/encryption/secrets"
+)
+
+func parseMigratedResources(secret *corev1.Secret) ([]schema.GroupResource, error) {
+	if secret == nil || secret.Annotations == nil {
+		return nil, nil
+	}
+	raw, ok := secret.Annotations[secrets.EncryptionSecretMigratedResources]
+	if !ok || len(raw) == 0 {
+		return nil, nil
+	}
+	migrated := secrets.MigratedGroupResources{}
+	if err := json.Unmarshal([]byte(raw), &migrated); err != nil {
+		return nil, fmt.Errorf("invalid %s annotation: %w", secrets.EncryptionSecretMigratedResources, err)
+	}
+	return migrated.Resources, nil
+}
+
+// AllEncryptedGRsMigrated returns true when every encrypted GR is listed in migrated-resources.
+func AllEncryptedGRsMigrated(secret *corev1.Secret, encryptedGRs []schema.GroupResource) (bool, error) {
+	migrated, err := parseMigratedResources(secret)
+	if err != nil {
+		return false, err
+	}
+	for _, gr := range encryptedGRs {
+		found := false
+		for _, migratedGR := range migrated {
+			if migratedGR == gr {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return false, nil
+		}
+	}
+	return len(encryptedGRs) > 0, nil
+}
diff --git a/pkg/operator/encryption/encryptionstatus/migration_test.go b/pkg/operator/encryption/encryptionstatus/migration_test.go
new file mode 100644
index 0000000000..a62468da19
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/migration_test.go
@@ -0,0 +1,36 @@
+package encryptionstatus
+
+import (
+	"encoding/json"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/openshift/library-go/pkg/operator/encryption/secrets"
+)
+
+func TestAllEncryptedGRsMigrated(t *testing.T) {
+	grs := []schema.GroupResource{{Group: "", Resource: "secrets"}, {Group: "", Resource: "configmaps"}}
+	migrated := secrets.MigratedGroupResources{Resources: grs}
+	raw, err := json.Marshal(migrated)
+	if err != nil {
+		t.Fatal(err)
+	}
+	secret := &corev1.Secret{}
+	secret.Annotations = map[string]string{
+		secrets.EncryptionSecretMigratedResources: string(raw),
+	}
+	ok, err := AllEncryptedGRsMigrated(secret, grs)
+	if err != nil || !ok {
+		t.Fatalf("expected all migrated, ok=%v err=%v", ok, err)
+	}
+
+	partial := secrets.MigratedGroupResources{Resources: grs[:1]}
+	raw, _ = json.Marshal(partial)
+	secret.Annotations[secrets.EncryptionSecretMigratedResources] = string(raw)
+	ok, err = AllEncryptedGRsMigrated(secret, grs)
+	if err != nil || ok {
+		t.Fatalf("expected partial migration, ok=%v err=%v", ok, err)
+	}
+}
diff --git a/pkg/operator/encryption/encryptionstatus/operator.go b/pkg/operator/encryption/encryptionstatus/operator.go
new file mode 100644
index 0000000000..2ec91de0f3
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/operator.go
@@ -0,0 +1,142 @@
+package encryptionstatus
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	operatorv1 "github.com/openshift/api/operator/v1"
+)
+
+// interimHealthReport is the JSON shape published in KMSHealthReporter_* condition messages.
+type interimHealthReport struct {
+	KEKID       string `json:"kekID"`
+	KeyID       string `json:"keyID"`
+	Status      string `json:"status"`
+	LastChecked string `json:"lastChecked"`
+}
+
+// encryptionStatusFromOperator reads structured encryption status from operator status.
+// Interim conditions remain the source until operator API types expose EncryptionStatus.
+func encryptionStatusFromOperator(status *operatorv1.OperatorStatus) APIServerEncryptionStatus {
+	_ = status
+	return APIServerEncryptionStatus{}
+}
+
+// HealthReportsFromOperatorStatus returns health reports from structured status when present,
+// otherwise parses interim KMSHealthReporter_* operator conditions.
+func HealthReportsFromOperatorStatus(status *operatorv1.OperatorStatus) []KMSPluginHealthReport {
+	encryptionStatus := encryptionStatusFromOperator(status)
+	if len(encryptionStatus.HealthReports) > 0 {
+		return encryptionStatus.HealthReports
+	}
+	return healthReportsFromConditions(status)
+}
+
+func healthReportsFromConditions(status *operatorv1.OperatorStatus) []KMSPluginHealthReport {
+	if status == nil {
+		return nil
+	}
+
+	var reports []KMSPluginHealthReport
+	for _, condition := range status.Conditions {
+		if !strings.HasPrefix(condition.Type, KMSHealthReporterConditionPrefix) {
+			continue
+		}
+		nodeName := strings.TrimPrefix(condition.Type, KMSHealthReporterConditionPrefix)
+		parsed, err := parseInterimHealthMessage(condition.Message)
+		if err != nil {
+			continue
+		}
+		for _, entry := range parsed {
+			reports = append(reports, KMSPluginHealthReport{
+				KeyID:    entry.KeyID,
+				NodeName: nodeName,
+				KEKID:    entry.KEKID,
+				Status:   entry.Status,
+				// LastChecked left zero when only interim JSON timestamp is available without parsing.
+			})
+		}
+	}
+	return reports
+}
+
+func parseInterimHealthMessage(message string) ([]interimHealthReport, error) {
+	if len(message) == 0 {
+		return nil, nil
+	}
+	var parsed []interimHealthReport
+	if err := json.Unmarshal([]byte(message), &parsed); err != nil {
+		return nil, err
+	}
+	return parsed, nil
+}
+
+// KeyRotationStatusFromOperatorStatus reads rotation history from structured status when present,
+// otherwise from the interim EncryptionKeyRotationStatus condition.
+func KeyRotationStatusFromOperatorStatus(status *operatorv1.OperatorStatus) ([]KMSPluginRotationStatus, error) {
+	encryptionStatus := encryptionStatusFromOperator(status)
+	if len(encryptionStatus.KeyRotationStatus) > 0 {
+		return encryptionStatus.KeyRotationStatus, nil
+	}
+	return keyRotationStatusFromCondition(status)
+}
+
+func keyRotationStatusFromCondition(status *operatorv1.OperatorStatus) ([]KMSPluginRotationStatus, error) {
+	if status == nil {
+		return nil, nil
+	}
+	for _, condition := range status.Conditions {
+		if condition.Type != KeyRotationStatusConditionType {
+			continue
+		}
+		if len(condition.Message) == 0 {
+			return nil, nil
+		}
+		var rotations []KMSPluginRotationStatus
+		if err := json.Unmarshal([]byte(condition.Message), &rotations); err != nil {
+			return nil, fmt.Errorf("failed to parse %s condition: %w", KeyRotationStatusConditionType, err)
+		}
+		return rotations, nil
+	}
+	return nil, nil
+}
+
+// SetKeyRotationStatusCondition returns an update func that stores keyRotationStatus in operator conditions
+// until status.encryptionStatus is available on the operator API type.
+func SetKeyRotationStatusCondition(rotations []KMSPluginRotationStatus) func(*operatorv1.OperatorStatus) error {
+	return func(status *operatorv1.OperatorStatus) error {
+		if status == nil {
+			return fmt.Errorf("operator status is nil")
+		}
+		message := ""
+		if len(rotations) > 0 {
+			bs, err := json.Marshal(rotations)
+			if err != nil {
+				return err
+			}
+			message = string(bs)
+		}
+		condition := operatorv1.OperatorCondition{
+			Type:    KeyRotationStatusConditionType,
+			Status:  operatorv1.ConditionTrue,
+			Reason:  KeyRotationStatusConditionReason,
+			Message: message,
+		}
+		setOperatorCondition(&status.Conditions, condition)
+		return nil
+	}
+}
+
+func setOperatorCondition(conditions *[]operatorv1.OperatorCondition, newCondition operatorv1.OperatorCondition) {
+	if conditions == nil {
+		return
+	}
+	for i, existing := range *conditions {
+		if existing.Type == newCondition.Type {
+			(*conditions)[i] = newCondition
+			return
+		}
+	}
+	*conditions = append(*conditions, newCondition)
+}
diff --git a/pkg/operator/encryption/encryptionstatus/operator_test.go b/pkg/operator/encryption/encryptionstatus/operator_test.go
new file mode 100644
index 0000000000..7bd9602037
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/operator_test.go
@@ -0,0 +1,41 @@
+package encryptionstatus
+
+import (
+	"testing"
+
+	operatorv1 "github.com/openshift/api/operator/v1"
+)
+
+func TestHealthReportsFromConditions(t *testing.T) {
+	status := &operatorv1.OperatorStatus{
+		Conditions: []operatorv1.OperatorCondition{
+			{
+				Type:    KMSHealthReporterConditionPrefix + "master-0",
+				Message: `[{"kekID":"kek-1","keyID":"2","status":"healthy","lastChecked":"2026-05-08T12:34:56Z"}]`,
+			},
+		},
+	}
+	reports := HealthReportsFromOperatorStatus(status)
+	if len(reports) != 1 {
+		t.Fatalf("expected 1 report, got %d", len(reports))
+	}
+	if reports[0].NodeName != "master-0" || reports[0].KeyID != "2" || reports[0].KEKID != "kek-1" {
+		t.Fatalf("unexpected report: %+v", reports[0])
+	}
+}
+
+func TestKeyRotationStatusRoundTrip(t *testing.T) {
+	rotations := []KMSPluginRotationStatus{{KeyID: "1", KEKID: "kek-a"}}
+	update := SetKeyRotationStatusCondition(rotations)
+	status := &operatorv1.OperatorStatus{}
+	if err := update(status); err != nil {
+		t.Fatal(err)
+	}
+	got, err := KeyRotationStatusFromOperatorStatus(status)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 || got[0].KeyID != "1" || got[0].KEKID != "kek-a" {
+		t.Fatalf("unexpected rotations: %+v", got)
+	}
+}
diff --git a/pkg/operator/encryption/encryptionstatus/rotation.go b/pkg/operator/encryption/encryptionstatus/rotation.go
new file mode 100644
index 0000000000..07cefae724
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/rotation.go
@@ -0,0 +1,101 @@
+package encryptionstatus
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// LatestCompletedRotationForKeyID returns the most recent completed entry for keyID.
+// Rotations are stored newest-first.
+func LatestCompletedRotationForKeyID(rotations []KMSPluginRotationStatus, keyID string) (KMSPluginRotationStatus, bool) {
+	for _, rotation := range rotations {
+		if rotation.KeyID != keyID {
+			continue
+		}
+		if rotation.MigrationFinishTime != nil {
+			return rotation, true
+		}
+	}
+	return KMSPluginRotationStatus{}, false
+}
+
+// OpenRotation returns the in-progress rotation for keyID and kekID (no migrationFinishTime).
+func OpenRotation(rotations []KMSPluginRotationStatus, keyID, kekID string) (KMSPluginRotationStatus, int, bool) {
+	for i, rotation := range rotations {
+		if rotation.KeyID != keyID || rotation.KEKID != kekID {
+			continue
+		}
+		if rotation.MigrationFinishTime == nil {
+			return rotation, i, true
+		}
+	}
+	return KMSPluginRotationStatus{}, -1, false
+}
+
+// GetOrCreateOpenRotation ensures an open rotation entry exists for keyID and kekID with discoveryTime set.
+func GetOrCreateOpenRotation(rotations []KMSPluginRotationStatus, keyID, kekID string, now metav1.Time) ([]KMSPluginRotationStatus, int) {
+	if idx := indexRotation(rotations, keyID, kekID); idx >= 0 {
+		if rotations[idx].MigrationFinishTime != nil {
+			rotations = prependRotation(rotations, newOpenRotation(keyID, kekID, now))
+			return rotations, 0
+		}
+		return SetDiscoveryTime(rotations, idx, now), idx
+	}
+	rotations = prependRotation(rotations, newOpenRotation(keyID, kekID, now))
+	return rotations, 0
+}
+
+func newOpenRotation(keyID, kekID string, now metav1.Time) KMSPluginRotationStatus {
+	discoveryTime := now.DeepCopy()
+	return KMSPluginRotationStatus{
+		KeyID:         keyID,
+		KEKID:         kekID,
+		DiscoveryTime: discoveryTime,
+	}
+}
+
+// SetDiscoveryTime sets discoveryTime on the rotation at index when unset.
+func SetDiscoveryTime(rotations []KMSPluginRotationStatus, index int, discoveryTime metav1.Time) []KMSPluginRotationStatus {
+	if index < 0 || index >= len(rotations) {
+		return rotations
+	}
+	if rotations[index].DiscoveryTime != nil {
+		return rotations
+	}
+	rotations[index].DiscoveryTime = discoveryTime.DeepCopy()
+	return rotations
+}
+
+// SetMigrationStartTime sets migrationStartTime on the rotation at index.
+func SetMigrationStartTime(rotations []KMSPluginRotationStatus, index int, startTime metav1.Time) []KMSPluginRotationStatus {
+	if index < 0 || index >= len(rotations) {
+		return rotations
+	}
+	rotations[index].MigrationStartTime = startTime.DeepCopy()
+	return rotations
+}
+
+// SetMigrationFinishTime sets migrationFinishTime on the rotation at index.
+func SetMigrationFinishTime(rotations []KMSPluginRotationStatus, index int, finishTime metav1.Time) []KMSPluginRotationStatus {
+	if index < 0 || index >= len(rotations) {
+		return rotations
+	}
+	rotations[index].MigrationFinishTime = finishTime.DeepCopy()
+	return rotations
+}
+
+func indexRotation(rotations []KMSPluginRotationStatus, keyID, kekID string) int {
+	for i, rotation := range rotations {
+		if rotation.KeyID == keyID && rotation.KEKID == kekID {
+			return i
+		}
+	}
+	return -1
+}
+
+func prependRotation(rotations []KMSPluginRotationStatus, rotation KMSPluginRotationStatus) []KMSPluginRotationStatus {
+	rotations = append([]KMSPluginRotationStatus{rotation}, rotations...)
+	if len(rotations) > MaxKeyRotationStatusEntries {
+		rotations = rotations[:MaxKeyRotationStatusEntries]
+	}
+	return rotations
+}
diff --git a/pkg/operator/encryption/encryptionstatus/rotation_test.go b/pkg/operator/encryption/encryptionstatus/rotation_test.go
new file mode 100644
index 0000000000..e96fa53e51
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/rotation_test.go
@@ -0,0 +1,55 @@
+package encryptionstatus
+
+import (
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestPrependRotationMaxEntries(t *testing.T) {
+	rotations := make([]KMSPluginRotationStatus, MaxKeyRotationStatusEntries)
+	for i := range rotations {
+		rotations[i] = KMSPluginRotationStatus{KEKID: string(rune('a' + i))}
+	}
+	rotations = prependRotation(rotations, KMSPluginRotationStatus{KeyID: "1", KEKID: "new"})
+	if len(rotations) != MaxKeyRotationStatusEntries {
+		t.Fatalf("expected %d entries, got %d", MaxKeyRotationStatusEntries, len(rotations))
+	}
+	if rotations[0].KEKID != "new" {
+		t.Fatalf("expected newest entry first, got %q", rotations[0].KEKID)
+	}
+}
+
+func TestUpsertOpenRotationUsesProvidedNowForMissingDiscoveryTime(t *testing.T) {
+	fixed := time.Date(2026, 5, 21, 12, 0, 0, 0, time.UTC)
+	now := metav1.NewTime(fixed)
+	rotations := []KMSPluginRotationStatus{
+		{KeyID: "1", KEKID: "kek-a"},
+	}
+
+	updated, idx := GetOrCreateOpenRotation(rotations, "1", "kek-a", now)
+	if idx != 0 {
+		t.Fatalf("expected index 0, got %d", idx)
+	}
+	if updated[0].DiscoveryTime == nil || !updated[0].DiscoveryTime.Equal(&now) {
+		t.Fatalf("expected discoveryTime %v, got %v", now, updated[0].DiscoveryTime)
+	}
+	if updated[0].KeyID != "1" || updated[0].KEKID != "kek-a" {
+		t.Fatalf("unexpected entry identity: %+v", updated[0])
+	}
+}
+
+func TestLatestCompletedRotationForKeyID(t *testing.T) {
+	finish := metav1.Now()
+	rotations := []KMSPluginRotationStatus{
+		{KeyID: "1", KEKID: "open"},
+		{KeyID: "1", KEKID: "new", MigrationFinishTime: &finish},
+		{KeyID: "1", KEKID: "old", MigrationFinishTime: &finish},
+		{KeyID: "2", KEKID: "other", MigrationFinishTime: &finish},
+	}
+	latest, ok := LatestCompletedRotationForKeyID(rotations, "1")
+	if !ok || latest.KEKID != "new" {
+		t.Fatalf("expected first completed entry for key 1, got %q ok=%v", latest.KEKID, ok)
+	}
+}
diff --git a/pkg/operator/encryption/encryptionstatus/types.go b/pkg/operator/encryption/encryptionstatus/types.go
new file mode 100644
index 0000000000..96af1ed5fd
--- /dev/null
+++ b/pkg/operator/encryption/encryptionstatus/types.go
@@ -0,0 +1,44 @@
+package encryptionstatus
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// APIServerEncryptionStatus mirrors the future operator/config status field for KMS encryption.
+type APIServerEncryptionStatus struct {
+	HealthReports     []KMSPluginHealthReport   `json:"healthReports,omitempty"`
+	KeyRotationStatus []KMSPluginRotationStatus `json:"keyRotationStatus,omitempty"`
+}
+
+// KMSPluginHealthReport describes per-node KMS plugin health (filled by the health controller).
+type KMSPluginHealthReport struct {
+	KeyID       string      `json:"keyId"`
+	NodeName    string      `json:"nodeName"`
+	KEKID       string      `json:"kekId,omitempty"`
+	Status      string      `json:"status"`
+	LastChecked metav1.Time `json:"lastChecked"`
+	Detail      string      `json:"detail,omitempty"`
+}
+
+// KMSPluginRotationStatus tracks one KEK rotation episode on the operand operator.
+type KMSPluginRotationStatus struct {
+	KeyID               string       `json:"keyId"`
+	KEKID               string       `json:"kekId"`
+	DiscoveryTime       *metav1.Time `json:"discoveryTime,omitempty"`
+	MigrationStartTime  *metav1.Time `json:"migrationStartTime,omitempty"`
+	MigrationFinishTime *metav1.Time `json:"migrationFinishTime,omitempty"`
+}
+
+const (
+	// MaxKeyRotationStatusEntries is the maximum number of rotation history entries kept in status.
+	MaxKeyRotationStatusEntries = 10
+
+	// KMSHealthReporterConditionPrefix is the interim condition type prefix used by the health controller.
+	KMSHealthReporterConditionPrefix = "KMSHealthReporter_"
+
+	// KeyRotationStatusConditionType stores keyRotationStatus JSON until status.encryptionStatus is available on the operator API.
+	KeyRotationStatusConditionType = "EncryptionKeyRotationStatus"
+
+	// KeyRotationStatusConditionReason is set when the condition carries the current rotation status snapshot.
+	KeyRotationStatusConditionReason = "AsExpected"
+)
diff --git a/pkg/operator/encryption/secrets/secrets.go b/pkg/operator/encryption/secrets/secrets.go
index e2e59efd6d..ae9f891bef 100644
--- a/pkg/operator/encryption/secrets/secrets.go
+++ b/pkg/operator/encryption/secrets/secrets.go
@@ -172,6 +172,17 @@ func (m *MigratedGroupResources) HasResource(resource schema.GroupResource) bool
 	return false
 }
 
+// FindKeySecret returns the encryption key secret for keyID from a ListKeySecrets result.
+func FindKeySecret(encryptionSecrets []*corev1.Secret, component, keyID string) *corev1.Secret {
+	name := fmt.Sprintf("encryption-key-%s-%s", component, keyID)
+	for _, secret := range encryptionSecrets {
+		if secret.Namespace == "openshift-config-managed" && secret.Name == name {
+			return secret
+		}
+	}
+	return nil
+}
+
 // ListKeySecrets returns the current key secrets from openshift-config-managed.
 func ListKeySecrets(ctx context.Context, secretClient corev1client.SecretsGetter, encryptionSecretSelector metav1.ListOptions) ([]*corev1.Secret, error) {
 	encryptionSecretList, err := secretClient.Secrets("openshift-config-managed").List(ctx, encryptionSecretSelector)