From ba9cc6067333a87c61c53fbf00e57a42fecde5aa Mon Sep 17 00:00:00 2001 From: Paolo Dettori Date: Wed, 10 Jun 2026 07:54:09 -0400 Subject: [PATCH 1/4] fix: wrap status update with RetryOnConflict to prevent linkedSkills loss The reconciler's final status update could fail with an optimistic concurrency conflict when another reconciliation loop modified the same AgentRuntime concurrently. Without retry, the linkedSkills field was never populated after the conflict, causing the E2E test to time out. Wrap the status update with retry.RetryOnConflict (the same pattern used throughout this controller) so the update re-fetches the latest object and re-applies all status fields on conflict. Closes #422 Assisted-By: Claude (Anthropic AI) Signed-off-by: Paolo Dettori --- .../controller/agentruntime_controller.go | 41 +++++++++++-------- .../agentruntime_controller_test.go | 38 +++++++++++++++++ 2 files changed, 63 insertions(+), 16 deletions(-) diff --git a/kagenti-operator/internal/controller/agentruntime_controller.go b/kagenti-operator/internal/controller/agentruntime_controller.go index 814f68f5..2680fff0 100644 --- a/kagenti-operator/internal/controller/agentruntime_controller.go +++ b/kagenti-operator/internal/controller/agentruntime_controller.go @@ -248,17 +248,9 @@ func (r *AgentRuntimeReconciler) Reconcile(ctx context.Context, req ctrl.Request // 6.5. Discover linked skills from workload annotation (set by kagenti backend or user) fg := r.getFeatureGates() + var linkedSkills []string if fg.SkillDiscovery { - rt.Status.LinkedSkills = r.readLinkedSkills(ctx, rt) - if len(rt.Status.LinkedSkills) > 0 { - r.setCondition(rt, ConditionTypeSkillsDiscovered, metav1.ConditionTrue, "SkillsFound", - fmt.Sprintf("%d linked skill(s) discovered from workload annotation", len(rt.Status.LinkedSkills))) - } else { - meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) - } - } else { - rt.Status.LinkedSkills = nil - meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) + linkedSkills = r.readLinkedSkills(ctx, rt) } // 7. Count configured pods @@ -267,12 +259,29 @@ func (r *AgentRuntimeReconciler) Reconcile(ctx context.Context, req ctrl.Request logger.V(1).Info("Failed to count configured pods", "error", err) } - // 8. Update status - rt.Status.ConfiguredPods = configuredPods - r.setPhase(rt, agentv1alpha1.RuntimePhaseActive) - r.setCondition(rt, ConditionTypeReady, metav1.ConditionTrue, "Configured", - fmt.Sprintf("Workload %s configured with config-hash %s", rt.Spec.TargetRef.Name, configResult.Hash[:12])) - if err := r.Status().Update(ctx, rt); err != nil { + // 8. Update status (retry on conflict to avoid losing linkedSkills updates) + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := r.Get(ctx, req.NamespacedName, rt); err != nil { + return err + } + rt.Status.ConfiguredPods = configuredPods + r.setPhase(rt, agentv1alpha1.RuntimePhaseActive) + r.setCondition(rt, ConditionTypeReady, metav1.ConditionTrue, "Configured", + fmt.Sprintf("Workload %s configured with config-hash %s", rt.Spec.TargetRef.Name, configResult.Hash[:12])) + if fg.SkillDiscovery { + rt.Status.LinkedSkills = linkedSkills + if len(linkedSkills) > 0 { + r.setCondition(rt, ConditionTypeSkillsDiscovered, metav1.ConditionTrue, "SkillsFound", + fmt.Sprintf("%d linked skill(s) discovered from workload annotation", len(linkedSkills))) + } else { + meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) + } + } else { + rt.Status.LinkedSkills = nil + meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) + } + return r.Status().Update(ctx, rt) + }); err != nil { logger.Error(err, "Failed to update status") return ctrl.Result{}, err } diff --git a/kagenti-operator/internal/controller/agentruntime_controller_test.go b/kagenti-operator/internal/controller/agentruntime_controller_test.go index 69178b89..7e5f1243 100644 --- a/kagenti-operator/internal/controller/agentruntime_controller_test.go +++ b/kagenti-operator/internal/controller/agentruntime_controller_test.go @@ -209,6 +209,44 @@ var _ = Describe("AgentRuntime Controller", func() { Expect(k8sClient.Get(ctx, nn, updatedRT)).To(Succeed()) Expect(updatedRT.Status.LinkedSkills).To(ConsistOf("summarizer", "translator")) }) + + It("should persist linkedSkills even after a concurrent status modification", func() { + dep := newDeployment("skills-conflict-deploy", namespace) + dep.Annotations = map[string]string{ + AnnotationSkills: `["skill-a","skill-b"]`, + } + Expect(k8sClient.Create(ctx, dep)).To(Succeed()) + defer func() { _ = k8sClient.Delete(ctx, dep) }() + + rt := newAgentRuntime("skills-conflict-rt", namespace, "skills-conflict-deploy", agentv1alpha1.RuntimeTypeAgent) + Expect(k8sClient.Create(ctx, rt)).To(Succeed()) + defer func() { _ = k8sClient.Delete(ctx, rt) }() + + r := newReconciler() + r.GetFeatureGates = func() *webhookconfig.FeatureGates { + return &webhookconfig.FeatureGates{SkillDiscovery: true} + } + nn := types.NamespacedName{Name: "skills-conflict-rt", Namespace: namespace} + + // First reconcile to set up finalizer + _, _ = r.Reconcile(ctx, reconcile.Request{NamespacedName: nn}) + + // Simulate a concurrent modification by updating the object's status + // from another controller (this bumps resourceVersion) + current := &agentv1alpha1.AgentRuntime{} + Expect(k8sClient.Get(ctx, nn, current)).To(Succeed()) + current.Status.ConfiguredPods = 99 + Expect(k8sClient.Status().Update(ctx, current)).To(Succeed()) + + // Now reconcile — the status update inside Reconcile will initially + // hit a conflict (stale resourceVersion), but retry should succeed + _, err := r.Reconcile(ctx, reconcile.Request{NamespacedName: nn}) + Expect(err).NotTo(HaveOccurred()) + + updatedRT := &agentv1alpha1.AgentRuntime{} + Expect(k8sClient.Get(ctx, nn, updatedRT)).To(Succeed()) + Expect(updatedRT.Status.LinkedSkills).To(ConsistOf("skill-a", "skill-b")) + }) }) Context("When setting status", func() { From fdd22e1d061ea3af539fea389ff409df951cefb6 Mon Sep 17 00:00:00 2001 From: Paolo Dettori Date: Wed, 10 Jun 2026 09:28:09 -0400 Subject: [PATCH 2/4] fix: snapshot status before retry to preserve all conditions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback: the re-fetch inside RetryOnConflict was discarding conditions computed earlier in the same reconcile (ConfigResolved, IstioMeshEnrolled) because they weren't persisted yet. Snapshot rt.Status.DeepCopy() before the retry loop and restore it after the fresh Get — this preserves all computed conditions while still picking up the latest resourceVersion for the update. Also add ConfigResolved assertion to the conflict unit test to lock down the full status-preservation contract. Assisted-By: Claude (Anthropic AI) Signed-off-by: Paolo Dettori --- .../controller/agentruntime_controller.go | 36 ++++++++++--------- .../agentruntime_controller_test.go | 3 ++ 2 files changed, 22 insertions(+), 17 deletions(-) diff --git a/kagenti-operator/internal/controller/agentruntime_controller.go b/kagenti-operator/internal/controller/agentruntime_controller.go index 2680fff0..3aba6099 100644 --- a/kagenti-operator/internal/controller/agentruntime_controller.go +++ b/kagenti-operator/internal/controller/agentruntime_controller.go @@ -259,27 +259,29 @@ func (r *AgentRuntimeReconciler) Reconcile(ctx context.Context, req ctrl.Request logger.V(1).Info("Failed to count configured pods", "error", err) } - // 8. Update status (retry on conflict to avoid losing linkedSkills updates) + // 8. Update status (retry on conflict to preserve all conditions computed above) + rt.Status.ConfiguredPods = configuredPods + r.setPhase(rt, agentv1alpha1.RuntimePhaseActive) + r.setCondition(rt, ConditionTypeReady, metav1.ConditionTrue, "Configured", + fmt.Sprintf("Workload %s configured with config-hash %s", rt.Spec.TargetRef.Name, configResult.Hash[:12])) + if fg.SkillDiscovery { + rt.Status.LinkedSkills = linkedSkills + if len(linkedSkills) > 0 { + r.setCondition(rt, ConditionTypeSkillsDiscovered, metav1.ConditionTrue, "SkillsFound", + fmt.Sprintf("%d linked skill(s) discovered from workload annotation", len(linkedSkills))) + } else { + meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) + } + } else { + rt.Status.LinkedSkills = nil + meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) + } + desired := rt.Status.DeepCopy() if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { if err := r.Get(ctx, req.NamespacedName, rt); err != nil { return err } - rt.Status.ConfiguredPods = configuredPods - r.setPhase(rt, agentv1alpha1.RuntimePhaseActive) - r.setCondition(rt, ConditionTypeReady, metav1.ConditionTrue, "Configured", - fmt.Sprintf("Workload %s configured with config-hash %s", rt.Spec.TargetRef.Name, configResult.Hash[:12])) - if fg.SkillDiscovery { - rt.Status.LinkedSkills = linkedSkills - if len(linkedSkills) > 0 { - r.setCondition(rt, ConditionTypeSkillsDiscovered, metav1.ConditionTrue, "SkillsFound", - fmt.Sprintf("%d linked skill(s) discovered from workload annotation", len(linkedSkills))) - } else { - meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) - } - } else { - rt.Status.LinkedSkills = nil - meta.RemoveStatusCondition(&rt.Status.Conditions, ConditionTypeSkillsDiscovered) - } + rt.Status = *desired return r.Status().Update(ctx, rt) }); err != nil { logger.Error(err, "Failed to update status") diff --git a/kagenti-operator/internal/controller/agentruntime_controller_test.go b/kagenti-operator/internal/controller/agentruntime_controller_test.go index 7e5f1243..5d5fa179 100644 --- a/kagenti-operator/internal/controller/agentruntime_controller_test.go +++ b/kagenti-operator/internal/controller/agentruntime_controller_test.go @@ -246,6 +246,9 @@ var _ = Describe("AgentRuntime Controller", func() { updatedRT := &agentv1alpha1.AgentRuntime{} Expect(k8sClient.Get(ctx, nn, updatedRT)).To(Succeed()) Expect(updatedRT.Status.LinkedSkills).To(ConsistOf("skill-a", "skill-b")) + + configCond := meta.FindStatusCondition(updatedRT.Status.Conditions, ConditionTypeConfigResolved) + Expect(configCond).NotTo(BeNil(), "ConfigResolved condition must survive the retry re-fetch") }) }) From 143256955eb9790f48555fb0060301c88bb4d21a Mon Sep 17 00:00:00 2001 From: Paolo Dettori Date: Wed, 10 Jun 2026 10:07:58 -0400 Subject: [PATCH 3/4] style: add intent comment on status overwrite Assisted-By: Claude (Anthropic AI) Signed-off-by: Paolo Dettori --- kagenti-operator/internal/controller/agentruntime_controller.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kagenti-operator/internal/controller/agentruntime_controller.go b/kagenti-operator/internal/controller/agentruntime_controller.go index 3aba6099..b16eb5ea 100644 --- a/kagenti-operator/internal/controller/agentruntime_controller.go +++ b/kagenti-operator/internal/controller/agentruntime_controller.go @@ -281,7 +281,7 @@ func (r *AgentRuntimeReconciler) Reconcile(ctx context.Context, req ctrl.Request if err := r.Get(ctx, req.NamespacedName, rt); err != nil { return err } - rt.Status = *desired + rt.Status = *desired // safe: this controller is the sole status owner return r.Status().Update(ctx, rt) }); err != nil { logger.Error(err, "Failed to update status") From 85616c9b32d90b4355640188da243a94b6330e09 Mon Sep 17 00:00:00 2001 From: Paolo Dettori Date: Wed, 10 Jun 2026 11:02:57 -0400 Subject: [PATCH 4/4] fix: don't strip user-managed kagenti.io/skills annotation on deletion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deletion handler was removing the kagenti.io/skills annotation from the target Deployment. This annotation is user-managed (set by the kagenti backend or user) — the operator reads but never sets it, so it should not remove it either. This caused the E2E skill discovery test to fail: the "feature gate disabled" test deletes the AgentRuntime, the finalizer strips the annotation, and the subsequent "feature gate enabled" test finds an empty annotation on the same Deployment. Also re-apply the Deployment fixture in the E2E "feature gate enabled" BeforeAll to guard against any future annotation-stripping regressions. Assisted-By: Claude (Anthropic AI) Signed-off-by: Paolo Dettori --- .../internal/controller/agentruntime_controller.go | 4 ---- kagenti-operator/test/e2e/e2e_test.go | 7 +++++++ 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/kagenti-operator/internal/controller/agentruntime_controller.go b/kagenti-operator/internal/controller/agentruntime_controller.go index b16eb5ea..18f69271 100644 --- a/kagenti-operator/internal/controller/agentruntime_controller.go +++ b/kagenti-operator/internal/controller/agentruntime_controller.go @@ -719,10 +719,6 @@ func (r *AgentRuntimeReconciler) handleDeletion(ctx context.Context, rt *agentv1 delete(workloadLabels, LabelManagedBy) acc.obj.SetLabels(workloadLabels) - // Remove skills annotation from workload metadata. - workloadAnnotations := acc.obj.GetAnnotations() - delete(workloadAnnotations, AnnotationSkills) - acc.obj.SetAnnotations(workloadAnnotations) // Remove kagenti.io/type from PodTemplateSpec pod labels so future pods // are not presented to the webhook with the type label. diff --git a/kagenti-operator/test/e2e/e2e_test.go b/kagenti-operator/test/e2e/e2e_test.go index da60d992..c35c12ac 100644 --- a/kagenti-operator/test/e2e/e2e_test.go +++ b/kagenti-operator/test/e2e/e2e_test.go @@ -2133,6 +2133,13 @@ rules: Context("Feature gate enabled", Ordered, func() { BeforeAll(func() { + By("re-applying target Deployment to restore skills annotation after prior deletion cleanup") + _, err := utils.KubectlApplyStdin(skillDiscoveryDeploymentFixture(), skillDiscoveryTestNamespace) + Expect(err).NotTo(HaveOccurred()) + Expect(utils.WaitForDeploymentReady( + "skill-discovery-agent", skillDiscoveryTestNamespace, 2*time.Minute, + )).To(Succeed()) + By("enabling skillDiscovery feature gate") Expect(utils.EnableSkillDiscovery(controllerNamespace, controllerDeployment)).To(Succeed())