feat(watcher): add --required_annotations to hold the finalizer until extra keys are true

cmd/watcher/main.go

@@ -85,6 +85,7 @@ var (
 	storeDeadline                = flag.Duration("store_deadline", 10*time.Minute, "How long to wait for storing the PipelineRun and TaskRun resources before aborting and clearing the finalizer in case of delete event")
 	forwardBuffer                = flag.Duration("forward_buffer", 150*time.Second, "This determines duration since completion time of TaskRun to wait for forwarder to finish")
 	managedByValues              = flag.String("managed_by_values", "", "Comma-separated list of additional spec.managedBy values the watcher will process. Runs with unset, empty, whitespace-only or \"tekton.dev/pipeline\" managedBy values are always accepted.")
+	requiredAnnotations          = newAnnotationFlag("required_annotation", "Repeatable flag. Use \"key=value\" to require an exact match, or just \"key\" to require existence of the annotation regardless of its value. The stored annotation is always implicitly required.")
 )
 
 func main() {
@@ -139,6 +140,7 @@ func main() {
 		SummaryAnnotations:           *summaryAnnotations,
 		DisableStoringIncompleteRuns: *disableStoringIncompleteRuns,
 		AllowedManagedByValues:       reconciler.ParseManagedByValues(*managedByValues),
+		RequiredAnnotations:          map[string]reconciler.AnnotationRequirement(*requiredAnnotations),
 	}
 
 	log.Printf("dynamic reconcile timeout %s and update log timeout is %s", cfg.DynamicReconcileTimeout.String(), cfg.UpdateLogTimeout.String())
@@ -246,3 +248,9 @@ func loadCerts() (*x509.CertPool, error) {
 	}
 	return certs, nil
 }
+
+func newAnnotationFlag(name, usage string) *reconciler.AnnotationFlag {
+	f := &reconciler.AnnotationFlag{}
+	flag.Var(f, name, usage)
+	return f
+}

docs/watcher/README.md

@@ -76,6 +76,32 @@ Watcher implements a finalizer to block deletion by an external pruner when obje
 
 When deletion request comes, it will block until completion time + `completed_run_grace_period` period is passed. A hard limit could be set as `store_deadline` (default 10m), after which the object will be removed from the cluster even without confirmation it's been stored in the DB.
 
+### Required annotations for finalizer release
+
+The `--required_annotation` flag (repeatable) specifies annotations that must be present on a PipelineRun or TaskRun before the Watcher clears its finalizer. This is useful in multicluster scenarios where external controllers (e.g. a Hub scheduler or syncer service) need to write annotations on a resource before it can be safely deleted.
+
+The flag supports two modes:
+1. **Value matching:** Use `key=value` to require that the annotation exists AND its value exactly matches the provided value.
+2. **Existence only:** Use `key` (without an `=`) to require that the annotation exists, regardless of what its value is.
+
+The `results.tekton.dev/stored` annotation is always implicitly required and does not need to be listed. When the flag is not provided (the default), only the stored annotation is checked and behavior is unchanged from previous versions.
+
+For example, to require that a Hub scheduler has annotated the resource (existence only) before the finalizer is cleared:
+
+```
+--required_annotation "hub.example.com/scheduled"
+```
+
+Multiple annotations with mixed requirements:
+
+```
+--required_annotation "hub.example.com/scheduled"
+--required_annotation "ci.example.com/status=passed"
+```
+
+If any required annotation is missing or does not match its expected value, the Watcher re-queues the resource and checks again after the `FinalizerRequeueInterval` (10 seconds). The `store_deadline` safety limit still applies — if the deadline passes, the finalizer is cleared regardless of whether the required annotations are present.
+
+> **Note:** This flag applies to both PipelineRun and TaskRun resources. CustomRuns are not affected.
 
 ## Filtering by `spec.managedBy`
 

pkg/watcher/reconciler/config.go

@@ -16,6 +16,8 @@
 package reconciler
 
 import (
+	"fmt"
+	"strings"
 	"time"
 
 	"k8s.io/apimachinery/pkg/labels"
@@ -83,6 +85,23 @@ type Config struct {
 	// included. Runs with nil or empty managedBy are also always accepted.
 	// This set must not be mutated after initialization to avoid data races.
 	AllowedManagedByValues sets.Set[string]
+
+	// RequiredAnnotations maps annotation keys to their requirements.
+	// Every listed annotation must be present before the watcher clears its
+	// finalizer. If the requirement specifies ExactMatch=true, the value must
+	// also match. The stored annotation (results.tekton.dev/stored) is always
+	// implicitly required and does not need to be listed. When nil/empty,
+	// only the stored annotation is checked (existing behavior).
+	RequiredAnnotations map[string]AnnotationRequirement
+}
+
+// AnnotationRequirement defines the condition an annotation must meet.
+type AnnotationRequirement struct {
+	// ExactMatch indicates if the annotation's value must exactly match Value.
+	// If false, only the existence of the annotation key is checked.
+	ExactMatch bool
+	// Value is the expected value of the annotation when ExactMatch is true.
+	Value string
 }
 
 // GetDisableAnnotationUpdate returns whether annotation updates should be
@@ -129,3 +148,66 @@ func (c *Config) SetLabelSelector(selector string) error {
 	c.labelSelector = parsedSelector
 	return nil
 }
+
+// AnnotationFlag implements flag.Value for a repeatable --required_annotation
+// flag. Each invocation adds one entry to the map.
+type AnnotationFlag map[string]AnnotationRequirement
+
+// String returns a display representation used by flag --help.
+func (f *AnnotationFlag) String() string {
+	if f == nil || len(*f) == 0 {
+		return ""
+	}
+	parts := make([]string, 0, len(*f))
+	for k, req := range *f {
+		if req.ExactMatch {
+			parts = append(parts, k+"="+req.Value)
+		} else {
+			parts = append(parts, k)
+		}
+	}
+	return strings.Join(parts, ", ")
+}
+
+// Set is called once per --required_annotation occurrence. It splits the
+// value on the first "=" to extract the annotation key and expected value.
+// If no "=" is present, it registers an existence-only requirement.
+func (f *AnnotationFlag) Set(val string) error {
+	val = strings.TrimSpace(val)
+	if val == "" {
+		return fmt.Errorf("invalid annotation: cannot be empty")
+	}
+	idx := strings.Index(val, "=")
+	if idx == 0 {
+		return fmt.Errorf(`invalid annotation %q, key cannot be empty`, val)
+	}
+	if *f == nil {
+		*f = make(map[string]AnnotationRequirement)
+	}
+	if idx < 0 {
+		(*f)[val] = AnnotationRequirement{ExactMatch: false}
+	} else {
+		(*f)[val[:idx]] = AnnotationRequirement{ExactMatch: true, Value: val[idx+1:]}
+	}
+	return nil
+}
+
+// AreRequiredAnnotationsReady checks whether every required annotation is
+// present and meets its requirement (existence or exact value). Returns the
+// key of the first unsatisfied annotation and false, or "" and true when all
+// are satisfied.
+func (c *Config) AreRequiredAnnotationsReady(annotations map[string]string) (missingKey string, ready bool) {
+	if c == nil || len(c.RequiredAnnotations) == 0 {
+		return "", true
+	}
+	for key, req := range c.RequiredAnnotations {
+		val, exists := annotations[key]
+		if !exists {
+			return key, false
+		}
+		if req.ExactMatch && val != req.Value {
+			return key, false
+		}
+	}
+	return "", true
+}

pkg/watcher/reconciler/config_test.go

@@ -44,6 +44,188 @@ func TestGetDisableAnnotationUpdate(t *testing.T) {
 	}
 }
 
+func TestAnnotationFlagSet(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		inputs  []string
+		want    map[string]AnnotationRequirement
+		wantErr bool
+	}{
+		{
+			name:   "key=value with arbitrary value",
+			inputs: []string{"ci.example.com/status=passed"},
+			want:   map[string]AnnotationRequirement{"ci.example.com/status": {ExactMatch: true, Value: "passed"}},
+		},
+		{
+			name:   "value containing equals - split on first only",
+			inputs: []string{"key=a=b=c"},
+			want:   map[string]AnnotationRequirement{"key": {ExactMatch: true, Value: "a=b=c"}},
+		},
+		{
+			name:   "existence only (no equals)",
+			inputs: []string{"hub.example.com/scheduled"},
+			want:   map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: false}},
+		},
+		{
+			name:   "multiple calls accumulate entries",
+			inputs: []string{"a=1", "b"},
+			want:   map[string]AnnotationRequirement{"a": {ExactMatch: true, Value: "1"}, "b": {ExactMatch: false}},
+		},
+		{
+			name:   "whitespace trimmed",
+			inputs: []string{"  hub.example.com/scheduled=true  "},
+			want:   map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: true, Value: "true"}},
+		},
+		{
+			name:    "empty string is rejected",
+			inputs:  []string{""},
+			wantErr: true,
+		},
+		{
+			name:    "empty key is rejected",
+			inputs:  []string{"=value"},
+			wantErr: true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var f AnnotationFlag
+			var err error
+			for _, input := range tc.inputs {
+				if setErr := f.Set(input); setErr != nil {
+					err = setErr
+				}
+			}
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("AnnotationFlag.Set(%v) expected error, got nil", tc.inputs)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("AnnotationFlag.Set(%v) unexpected error: %v", tc.inputs, err)
+			}
+			if len(f) != len(tc.want) {
+				t.Fatalf("AnnotationFlag after Set(%v) = %v (len %d), want %v (len %d)", tc.inputs, f, len(f), tc.want, len(tc.want))
+			}
+			for k, wantReq := range tc.want {
+				if gotReq, ok := f[k]; !ok {
+					t.Errorf("AnnotationFlag missing key %q", k)
+				} else if gotReq != wantReq {
+					t.Errorf("AnnotationFlag[%q] = %+v, want %+v", k, gotReq, wantReq)
+				}
+			}
+		})
+	}
+}
+
+func TestAreRequiredAnnotationsReady(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		cfg         *Config
+		annotations map[string]string
+		wantReady   bool
+		wantMissing string
+	}{
+		{
+			name:        "nil config is always ready",
+			cfg:         nil,
+			annotations: nil,
+			wantReady:   true,
+		},
+		{
+			name:        "empty required map is always ready",
+			cfg:         &Config{},
+			annotations: nil,
+			wantReady:   true,
+		},
+		{
+			name:        "required annotation present and matching",
+			cfg:         &Config{RequiredAnnotations: map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: true, Value: "true"}}},
+			annotations: map[string]string{"hub.example.com/scheduled": "true"},
+			wantReady:   true,
+		},
+		{
+			name:        "required annotation missing",
+			cfg:         &Config{RequiredAnnotations: map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: true, Value: "true"}}},
+			annotations: map[string]string{},
+			wantReady:   false,
+			wantMissing: "hub.example.com/scheduled",
+		},
+		{
+			name:        "required annotation present but wrong value",
+			cfg:         &Config{RequiredAnnotations: map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: true, Value: "true"}}},
+			annotations: map[string]string{"hub.example.com/scheduled": "false"},
+			wantReady:   false,
+			wantMissing: "hub.example.com/scheduled",
+		},
+		{
+			name:        "existence only - present",
+			cfg:         &Config{RequiredAnnotations: map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: false}}},
+			annotations: map[string]string{"hub.example.com/scheduled": "anything"},
+			wantReady:   true,
+		},
+		{
+			name:        "existence only - missing",
+			cfg:         &Config{RequiredAnnotations: map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: false}}},
+			annotations: map[string]string{},
+			wantReady:   false,
+			wantMissing: "hub.example.com/scheduled",
+		},
+		{
+			name:        "nil annotations map",
+			cfg:         &Config{RequiredAnnotations: map[string]AnnotationRequirement{"hub.example.com/scheduled": {ExactMatch: true, Value: "true"}}},
+			annotations: nil,
+			wantReady:   false,
+			wantMissing: "hub.example.com/scheduled",
+		},
+		{
+			name: "multiple required - all satisfied",
+			cfg: &Config{RequiredAnnotations: map[string]AnnotationRequirement{
+				"hub.example.com/scheduled": {ExactMatch: true, Value: "true"},
+				"syncer.example.com/synced": {ExactMatch: true, Value: "true"},
+			}},
+			annotations: map[string]string{
+				"hub.example.com/scheduled": "true",
+				"syncer.example.com/synced": "true",
+			},
+			wantReady: true,
+		},
+		{
+			name: "multiple required - one missing",
+			cfg: &Config{RequiredAnnotations: map[string]AnnotationRequirement{
+				"hub.example.com/scheduled": {ExactMatch: true, Value: "true"},
+				"syncer.example.com/synced": {ExactMatch: true, Value: "true"},
+			}},
+			annotations: map[string]string{
+				"hub.example.com/scheduled": "true",
+			},
+			wantReady: false,
+		},
+		{
+			name: "mixed true and arbitrary values - all satisfied",
+			cfg: &Config{RequiredAnnotations: map[string]AnnotationRequirement{
+				"hub.example.com/scheduled": {ExactMatch: false},
+				"ci.example.com/status":     {ExactMatch: true, Value: "passed"},
+			}},
+			annotations: map[string]string{
+				"hub.example.com/scheduled": "true",
+				"ci.example.com/status":     "passed",
+			},
+			wantReady: true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			missingKey, ready := tc.cfg.AreRequiredAnnotationsReady(tc.annotations)
+			if ready != tc.wantReady {
+				t.Errorf("AreRequiredAnnotationsReady() ready = %t, want %t", ready, tc.wantReady)
+			}
+			if tc.wantMissing != "" && missingKey != tc.wantMissing {
+				t.Errorf("AreRequiredAnnotationsReady() missingKey = %q, want %q", missingKey, tc.wantMissing)
+			}
+		})
+	}
+}
+
 func TestCompletedResourceGracePeriod(t *testing.T) {
 	for _, tc := range []struct {
 		cfg  *Config

pkg/watcher/reconciler/pipelinerun/reconciler.go

@@ -318,6 +318,12 @@ func (r *Reconciler) finalize(ctx context.Context, pr *pipelinev1.PipelineRun, r
 		return controller.NewRequeueAfter(r.cfg.FinalizerRequeueInterval)
 	}
 
+	if missingKey, ready := r.cfg.AreRequiredAnnotationsReady(pr.Annotations); !ready {
+		logging.FromContext(ctx).Debugf("required annotation %q is not ready on pipelinerun %s/%s, requeuing",
+			missingKey, pr.Namespace, pr.Name)
+		return controller.NewRequeueAfter(r.cfg.FinalizerRequeueInterval)
+	}
+
 	return nil
 }