fix: re-sync PrefectAutomation every reconcile for continuous self-heal

PrefectAutomation persisted Status.LastSyncTime, so needsSync short-circuited on
a 10-minute drift timer and never re-checked Prefect in between. A manual edit or
delete of the automation in Prefect was therefore not reconciled until the drift
window elapsed or the operator restarted.

PrefectDeployment reconciles against Prefect on every pass (its status carries no
LastSyncTime), so it self-heals continuously. Align PrefectAutomation with that:
needsSync now returns true each reconcile. syncWithPrefect is idempotent
(update-in-place; with #297, recreate-on-404), so re-running it every pass is safe.

Adds a unit test asserting needsSync returns true in steady state.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ShmuelOps

internal/controller/prefectautomation_controller.go

@@ -120,10 +120,14 @@ func (r *PrefectAutomationReconciler) needsSync(automation *prefectiov1.PrefectA
 	if automation.Status.ObservedGeneration < automation.Generation {
 		return true
 	}
-	if automation.Status.LastSyncTime == nil {
-		return true
-	}
-	return time.Since(automation.Status.LastSyncTime.Time) > 10*time.Minute
+	// Otherwise re-sync on every reconcile, rather than gating on a coarse drift
+	// timer. PrefectDeployment reconciles against Prefect each loop (its status
+	// carries no LastSyncTime) and so self-heals out-of-band changes continuously;
+	// PrefectAutomation persisted LastSyncTime and so only re-checked every 10m,
+	// meaning a manual edit/delete in Prefect wasn't reconciled until the drift
+	// window elapsed or the operator restarted. syncWithPrefect is idempotent, so
+	// re-running it each pass is safe.
+	return true
 }
 
 // syncWithPrefect creates or updates the automation in the Prefect API

internal/controller/prefectautomation_needssync_test.go

@@ -0,0 +1,68 @@
+/*
+Copyright 2024.
+
+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 controller
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	prefectiov1 "github.com/PrefectHQ/prefect-operator/api/v1"
+)
+
+// needsSync must return true on every reconcile so the controller re-checks the
+// automation against Prefect each loop and self-heals an out-of-band edit/delete
+// (matching PrefectDeployment's continuous reconcile) rather than only on a
+// coarse drift timer or an operator restart.
+func TestAutomationNeedsSync(t *testing.T) {
+	r := &PrefectAutomationReconciler{}
+	id := "automation-1"
+	now := metav1.Now()
+
+	newHash := "hash-1"
+	base := func() *prefectiov1.PrefectAutomation {
+		a := &prefectiov1.PrefectAutomation{}
+		a.Generation = 2
+		a.Status.Id = &id
+		a.Status.SpecHash = newHash
+		a.Status.ObservedGeneration = 2
+		a.Status.LastSyncTime = &now
+		return a
+	}
+
+	cases := []struct {
+		name    string
+		mutate  func(*prefectiov1.PrefectAutomation)
+		hash    string
+		wantMsg string
+	}{
+		{"no id", func(a *prefectiov1.PrefectAutomation) { a.Status.Id = nil }, newHash, "unsynced automation"},
+		{"spec changed", func(a *prefectiov1.PrefectAutomation) {}, "different-hash", "spec change"},
+		{"generation ahead", func(a *prefectiov1.PrefectAutomation) { a.Generation = 3 }, newHash, "generation bump"},
+		{"steady state (just synced)", func(a *prefectiov1.PrefectAutomation) {}, newHash, "continuous self-heal"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			a := base()
+			tc.mutate(a)
+			if !r.needsSync(a, tc.hash) {
+				t.Fatalf("needsSync = false, want true (%s)", tc.wantMsg)
+			}
+		})
+	}
+}