fix: harden PrefectAutomation controller (#294 follow-ups)

Follow-up hardening for the PrefectAutomation CRD merged in #294. None of
these were blockers, but worth addressing before the CRD sees real use:

- Self-heal when the automation is deleted out-of-band in Prefect: a 404 on
  update now clears the stored ID and recreates instead of looping on
  SyncError forever.
- Propagate invalid JSON in action parameters/jobVariables instead of
  silently dropping it (matches the match/matchRelated paths).
- Enforce deploymentId/deploymentName rules in Validate(), grounded in the
  Prefect API schema: the two are mutually exclusive, and for
  run/pause/resume-deployment a "selected" source (the default) requires
  exactly one target while "inferred" requires none.
- Error on ambiguous deployment names in FindDeploymentByName rather than
  silently resolving to an arbitrary match. Deployment names are only unique
  per flow, so a bare name can legitimately match several. Flow-qualified
  references are a larger CRD change left as a follow-up.
- Add the missing controller envtest suite plus unit coverage for the above.

Claude-Session: https://claude.ai/code/session_01YXNFE2AvNggTqhD3oBWmMg

Author: mitchnielsen

api/v1/prefectautomation_types.go

@@ -184,6 +184,13 @@ type PrefectSequenceTrigger struct {
 	Triggers []PrefectChildTrigger `json:"triggers"`
 }
 
+// Deployment-target action types (their target is a deployment).
+const (
+	ActionRunDeployment    = "run-deployment"
+	ActionPauseDeployment  = "pause-deployment"
+	ActionResumeDeployment = "resume-deployment"
+)
+
 // PrefectAutomationAction matches Prefect's action union. `type` selects the
 // action; the remaining fields apply to the relevant action types.
 type PrefectAutomationAction struct {
@@ -326,6 +333,62 @@ func (a *PrefectAutomation) Validate() error {
 	if set != 1 {
 		return fmt.Errorf("exactly one of trigger.event, trigger.metric, trigger.compound, or trigger.sequence must be set (got %d)", set)
 	}
+
+	for _, list := range []struct {
+		field   string
+		actions []PrefectAutomationAction
+	}{
+		{"actions", a.Spec.Actions},
+		{"actionsOnTrigger", a.Spec.ActionsOnTrigger},
+		{"actionsOnResolve", a.Spec.ActionsOnResolve},
+	} {
+		for i := range list.actions {
+			if err := validateAction(list.field, i, list.actions[i]); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// deploymentTargetActions are the action types whose target is a deployment.
+var deploymentTargetActions = map[string]struct{}{
+	ActionRunDeployment:    {},
+	ActionPauseDeployment:  {},
+	ActionResumeDeployment: {},
+}
+
+// validateAction enforces the deploymentId/deploymentName rules that mirror the
+// Prefect API: the two fields are two ways to supply the same deployment_id (so
+// never both), and for deployment-target actions the presence of a target is
+// governed by `source` ("selected" requires exactly one, "inferred" requires
+// neither). Source defaults to "selected" per the API.
+func validateAction(field string, i int, action PrefectAutomationAction) error {
+	hasID := action.DeploymentID != nil && *action.DeploymentID != ""
+	hasName := action.DeploymentName != nil && *action.DeploymentName != ""
+
+	if hasID && hasName {
+		return fmt.Errorf("%s[%d] (%s): deploymentId and deploymentName are mutually exclusive", field, i, action.Type)
+	}
+
+	if _, isDeploymentAction := deploymentTargetActions[action.Type]; !isDeploymentAction {
+		return nil
+	}
+
+	source := "selected"
+	if action.Source != nil && *action.Source != "" {
+		source = *action.Source
+	}
+	switch source {
+	case "inferred":
+		if hasID || hasName {
+			return fmt.Errorf("%s[%d] (%s): deploymentId/deploymentName must not be set when source is 'inferred'", field, i, action.Type)
+		}
+	default: // "selected"
+		if !hasID && !hasName {
+			return fmt.Errorf("%s[%d] (%s): exactly one of deploymentId or deploymentName must be set when source is 'selected'", field, i, action.Type)
+		}
+	}
 	return nil
 }
 

api/v1/prefectautomation_types_test.go

@@ -0,0 +1,162 @@
+/*
+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 v1
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("PrefectAutomation Validate", func() {
+	// eventTrigger is a minimal valid trigger so tests can focus on action rules.
+	eventTrigger := PrefectAutomationTrigger{
+		Event: &PrefectEventTrigger{Posture: "Reactive"},
+	}
+
+	automationWithActions := func(actions ...PrefectAutomationAction) *PrefectAutomation {
+		return &PrefectAutomation{
+			Spec: PrefectAutomationSpec{
+				Name:    "a",
+				Trigger: eventTrigger,
+				Actions: actions,
+			},
+		}
+	}
+
+	Describe("trigger union", func() {
+		It("accepts exactly one trigger member", func() {
+			a := &PrefectAutomation{Spec: PrefectAutomationSpec{Name: "a", Trigger: eventTrigger}}
+			Expect(a.Validate()).To(Succeed())
+		})
+
+		It("rejects zero trigger members", func() {
+			a := &PrefectAutomation{Spec: PrefectAutomationSpec{Name: "a"}}
+			Expect(a.Validate()).To(HaveOccurred())
+		})
+
+		It("rejects more than one trigger member", func() {
+			a := &PrefectAutomation{Spec: PrefectAutomationSpec{
+				Name: "a",
+				Trigger: PrefectAutomationTrigger{
+					Event:  &PrefectEventTrigger{Posture: "Reactive"},
+					Metric: &PrefectMetricTrigger{},
+				},
+			}}
+			Expect(a.Validate()).To(HaveOccurred())
+		})
+	})
+
+	Describe("deployment target actions", func() {
+		It("rejects setting both deploymentId and deploymentName", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:           "run-deployment",
+				Source:         new("selected"),
+				DeploymentID:   new("dep-id"),
+				DeploymentName: new("dep-name"),
+			})
+			err := a.Validate()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("mutually exclusive"))
+		})
+
+		It("rejects both even for non-deployment action types", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:           "pause-automation",
+				DeploymentID:   new("dep-id"),
+				DeploymentName: new("dep-name"),
+			})
+			Expect(a.Validate()).To(HaveOccurred())
+		})
+
+		It("requires a target when source is selected", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:   "run-deployment",
+				Source: new("selected"),
+			})
+			err := a.Validate()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("selected"))
+		})
+
+		It("requires a target when source is unset (defaults to selected)", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type: "run-deployment",
+			})
+			Expect(a.Validate()).To(HaveOccurred())
+		})
+
+		It("accepts a selected action with deploymentId only", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:         "run-deployment",
+				Source:       new("selected"),
+				DeploymentID: new("dep-id"),
+			})
+			Expect(a.Validate()).To(Succeed())
+		})
+
+		It("accepts a selected action with deploymentName only", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:           "pause-deployment",
+				Source:         new("selected"),
+				DeploymentName: new("dep-name"),
+			})
+			Expect(a.Validate()).To(Succeed())
+		})
+
+		It("rejects a target when source is inferred", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:         "run-deployment",
+				Source:       new("inferred"),
+				DeploymentID: new("dep-id"),
+			})
+			err := a.Validate()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("inferred"))
+		})
+
+		It("accepts an inferred action with no target", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:   "resume-deployment",
+				Source: new("inferred"),
+			})
+			Expect(a.Validate()).To(Succeed())
+		})
+
+		It("validates actions across all three lists", func() {
+			a := &PrefectAutomation{Spec: PrefectAutomationSpec{
+				Name:    "a",
+				Trigger: eventTrigger,
+				ActionsOnResolve: []PrefectAutomationAction{
+					{Type: "run-deployment", Source: new("selected")},
+				},
+			}}
+			err := a.Validate()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("actionsOnResolve"))
+		})
+	})
+
+	Describe("non-deployment actions", func() {
+		It("does not require a deployment target", func() {
+			a := automationWithActions(PrefectAutomationAction{
+				Type:    "send-notification",
+				Subject: new("hi"),
+			})
+			Expect(a.Validate()).To(Succeed())
+		})
+	})
+})

internal/controller/prefectautomation_controller.go

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -171,6 +172,18 @@ func (r *PrefectAutomationReconciler) syncWithPrefect(ctx context.Context, autom
 	var result *prefect.Automation
 	if automation.Status.Id != nil && *automation.Status.Id != "" {
 		result, err = prefectClient.UpdateAutomation(ctx, *automation.Status.Id, automationSpec)
+		// If the automation was deleted out-of-band, clear the stale ID and
+		// requeue so the next pass recreates it instead of looping on SyncError.
+		if errors.Is(err, prefect.ErrAutomationNotFound) {
+			log.Info("Automation no longer exists in Prefect, recreating", "automation", automation.Name, "prefectId", *automation.Status.Id)
+			automation.Status.Id = nil
+			r.setCondition(automation, PrefectAutomationConditionSynced, metav1.ConditionFalse, "Recreating", "Automation was deleted in Prefect; recreating")
+			automation.Status.Ready = false
+			if updateErr := r.Status().Update(ctx, automation); updateErr != nil {
+				log.Error(updateErr, "Failed to update automation status", "automation", automation.Name)
+			}
+			return ctrl.Result{RequeueAfter: time.Second}, nil
+		}
 	} else {
 		result, err = prefectClient.CreateAutomation(ctx, automationSpec)
 	}