diff --git a/.gitignore b/.gitignore
index 5614a596..ca0333a3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,3 +11,4 @@
 .DS_Store
 /docs/__pycache__
 .idea/
+temp/
diff --git a/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml b/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml
index 59224836..e5dff587 100644
--- a/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml
+++ b/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml
@@ -387,7 +387,26 @@ spec:
                           the destination side (i.e. the original related object will not be touched, regardless of this
                           option). Leaving this disabled, the syncagent will only create copies of the related objects,
                           but never delete them itself.
+
+                          Deprecated: use CleanupPolicy instead. When CleanupPolicy is empty, cleanup:true is treated
+                          as OnPrimaryDeletion and cleanup:false as Orphan.
                         type: boolean
+                      cleanupPolicy:
+                        description: |-
+                          CleanupPolicy controls when the syncagent deletes destination copies of this related
+                          resource. The original object on the origin side is never deleted.
+
+                            - Orphan (default): copies are never deleted by the agent.
+                            - OnPrimaryDeletion: copies are deleted only when the primary object is deleted.
+                            - MatchOrigin: copies are pruned as soon as their origin object no longer exists, and
+                              all copies are deleted when the primary object is deleted.
+
+                          When left empty, the deprecated Cleanup field is used to derive the effective policy.
+                        enum:
+                          - Orphan
+                          - OnPrimaryDeletion
+                          - MatchOrigin
+                        type: string
                       group:
                         description: |-
                           Group is the API group of the related resource. This should be left blank for resources
@@ -878,6 +897,10 @@ spec:
                         rule: '!has(self.identityHash) || (has(self.group) && has(self.version) && has(self.resource))'
                       - message: watch must be configured when origin is service and syncStatus is true
                         rule: '!(self.origin == ''service'' && has(self.syncStatus) && self.syncStatus) || has(self.watch)'
+                      - message: watch must be configured when cleanupPolicy is MatchOrigin and origin is service
+                        rule: '!(has(self.cleanupPolicy) && self.cleanupPolicy == ''MatchOrigin'' && self.origin == ''service'') || has(self.watch)'
+                      - message: 'cleanup:true conflicts with cleanupPolicy: Orphan'
+                        rule: '!(has(self.cleanup) && self.cleanup && has(self.cleanupPolicy) && self.cleanupPolicy == ''Orphan'')'
                   type: array
                 resource:
                   description: |-
diff --git a/docs/content/publish-resources/related-resources.md b/docs/content/publish-resources/related-resources.md
index 3c0a9751..69f82660 100644
--- a/docs/content/publish-resources/related-resources.md
+++ b/docs/content/publish-resources/related-resources.md
@@ -571,3 +571,58 @@ individually. For each namespace it will again follow the configured source, may
 template or reference. If again a label selector is used, it will be applied in each namespace and
 the configured rewrite rule will be evaluated once per found object. In this case, `.Value` is the
 name of found object.
+
+## Cleanup
+
+By default the agent only ever _creates_ copies of related resources; it never deletes them. This
+is intentional: copies can be useful as audit trails, may have decoupled lifecycles, or may be owned
+by the service provider. The behavior is controlled per related resource via the `cleanupPolicy`
+field, which takes one of three values:
+
+- `Orphan` (default): copies are never deleted by the agent.
+- `OnPrimaryDeletion`: copies are deleted only when the primary object is deleted.
+- `MatchOrigin`: the destination set is kept equal to the origin set — a copy is pruned as soon as
+  its origin object no longer exists, and all copies are deleted when the primary object is deleted.
+
+**The original object on the origin side is never deleted, under any policy.** The agent only ever
+deletes destination copies that it created itself (they carry provenance labels identifying the
+owning primary object); hand-created objects are never touched.
+
+`MatchOrigin` prunes copies while the agent reconciles the primary object. For an `origin: service`
+related resource this means a `watch` must be configured, otherwise a deleted source object would
+only be noticed on the next incidental reconcile of the primary. The CRD enforces this: setting
+`cleanupPolicy: MatchOrigin` together with `origin: service` requires a `watch` block.
+
+```yaml
+apiVersion: syncagent.kcp.io/v1alpha1
+kind: PublishedResource
+metadata:
+  name: publish-crontabs
+spec:
+  resource:
+    apiGroup: example.com
+    version: v1
+    kind: CronTab
+  related:
+    - identifier: credentials
+      origin: service
+      kind: Secret
+      cleanupPolicy: MatchOrigin
+      object:
+        selector:
+          matchLabels:
+            app: credentials
+          rewrite:
+            template:
+              template: "{{ .Value }}"
+      watch:
+        bySelector:
+          matchLabels:
+            example.com/managed: "true"
+```
+
+!!! note
+    The deprecated boolean `cleanup` field still works for backwards compatibility. When
+    `cleanupPolicy` is not set, `cleanup: true` is treated as `OnPrimaryDeletion` and `cleanup: false`
+    (or unset) as `Orphan`. New configurations should use `cleanupPolicy` instead; setting
+    `cleanup: true` together with `cleanupPolicy: Orphan` is rejected as a contradiction.
diff --git a/internal/sync/meta.go b/internal/sync/meta.go
index c5bf76f8..530ca14f 100644
--- a/internal/sync/meta.go
+++ b/internal/sync/meta.go
@@ -158,3 +158,48 @@ func (k objectKey) Annotations() labels.Set {
 
 	return s
 }
+
+// relatedCopyLabels builds the provenance labels put on a destination copy of a related resource.
+// They tie the copy to its owning primary object and the related resource identifier so that all
+// copies of a given (primary, identifier) can be enumerated via relatedCopySelector. Names and
+// namespaces are hashed because they can exceed the label value limit or contain invalid
+// characters; the identifier is validated to be alphanumeric by the API, so it is used verbatim.
+// The agent name (when set) is included so that each agent only ever prunes its own copies.
+func relatedCopyLabels(primary ctrlruntimeclient.Object, identifier, agentName string) map[string]string {
+	set := map[string]string{
+		relatedPrimaryNameHashLabel: crypto.Hash(primary.GetName()),
+		relatedIdentifierLabel:      identifier,
+	}
+
+	if namespace := primary.GetNamespace(); namespace != "" {
+		set[relatedPrimaryNamespaceHashLabel] = crypto.Hash(namespace)
+	}
+
+	if agentName != "" {
+		set[agentNameLabel] = agentName
+	}
+
+	return set
+}
+
+// relatedCopyAnnotations builds the human-facing provenance annotations (plaintext primary
+// namespace/name) for a destination copy of a related resource.
+func relatedCopyAnnotations(primary ctrlruntimeclient.Object) map[string]string {
+	set := map[string]string{
+		relatedPrimaryNameAnnotation: primary.GetName(),
+	}
+
+	if namespace := primary.GetNamespace(); namespace != "" {
+		set[relatedPrimaryNamespaceAnnotation] = namespace
+	}
+
+	return set
+}
+
+// relatedCopySelector returns a label selector matching exactly the destination copies created for
+// the given primary object and related resource identifier (scoped to the agent when set). Because
+// it mirrors relatedCopyLabels, only objects the agent itself labelled are ever selected, so
+// hand-created objects are never in scope for pruning.
+func relatedCopySelector(primary ctrlruntimeclient.Object, identifier, agentName string) labels.Selector {
+	return labels.SelectorFromSet(relatedCopyLabels(primary, identifier, agentName))
+}
diff --git a/internal/sync/object_syncer.go b/internal/sync/object_syncer.go
index cabbae03..0ee6124d 100644
--- a/internal/sync/object_syncer.go
+++ b/internal/sync/object_syncer.go
@@ -57,6 +57,14 @@ type objectSyncer struct {
 	blockSourceDeletion bool
 	// whether or not to place sync-related metadata on the destination object
 	metadataOnDestination bool
+	// destLabels are additional labels stamped onto the destination object on every apply.
+	// Used to attach related-resource provenance (owning primary + identifier) to copies so
+	// that they can later be enumerated and pruned. These are applied additively and never
+	// clobber labels the copy already carries.
+	destLabels map[string]string
+	// destAnnotations are additional annotations stamped onto the destination object on every
+	// apply; the human-facing counterpart to destLabels.
+	destAnnotations map[string]string
 	// optional mutations for both directions of the sync
 	mutator mutation.Mutator
 	// stateStore is capable of remembering the state of a Kubernetes object
@@ -420,6 +428,9 @@ func (s *objectSyncer) applyServerSide(ctx context.Context, log *zap.SugaredLogg
 		s.labelWithAgent(desired)
 	}
 
+	// stamp any additional provenance labels/annotations (e.g. related-resource ownership)
+	s.ensureDestMetadata(desired)
+
 	// do not claim ownership of subresource content via the main resource
 	// apply; status is reconciled separately and "scale" is never written.
 	s.removeSubresources(desired)
@@ -503,6 +514,9 @@ func (s *objectSyncer) ensureDestinationObject(ctx context.Context, log *zap.Sug
 		s.labelWithAgent(destObj)
 	}
 
+	// stamp any additional provenance labels/annotations (e.g. related-resource ownership)
+	s.ensureDestMetadata(destObj)
+
 	// finally, we can create the destination object
 	objectLog := log.With("dest-object", newObjectKey(destObj, dest.clusterName, logicalcluster.None))
 	objectLog.Debugw("Creating destination object…")
@@ -551,6 +565,7 @@ func (s *objectSyncer) adoptExistingDestinationObject(ctx context.Context, log *
 	ensureAnnotations(existingDestObj, sourceKey.Annotations())
 
 	s.labelWithAgent(existingDestObj)
+	s.ensureDestMetadata(existingDestObj)
 
 	if err := dest.client.Update(ctx, existingDestObj); err != nil {
 		return fmt.Errorf("failed to upsert current destination object labels: %w", err)
@@ -654,3 +669,16 @@ func (s *objectSyncer) labelWithAgent(obj *unstructured.Unstructured) {
 		ensureLabels(obj, map[string]string{agentNameLabel: s.agentName})
 	}
 }
+
+// ensureDestMetadata stamps the syncer's additional destination labels/annotations onto the given
+// object. It is additive (it never removes labels the object already carries) and is used to
+// attach related-resource provenance to destination copies.
+func (s *objectSyncer) ensureDestMetadata(obj *unstructured.Unstructured) {
+	if len(s.destLabels) > 0 {
+		ensureLabels(obj, s.destLabels)
+	}
+
+	if len(s.destAnnotations) > 0 {
+		ensureAnnotations(obj, s.destAnnotations)
+	}
+}
diff --git a/internal/sync/syncer_related.go b/internal/sync/syncer_related.go
index 84a059ee..bb7be6e3 100644
--- a/internal/sync/syncer_related.go
+++ b/internal/sync/syncer_related.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"regexp"
 	"slices"
 	"strings"
@@ -36,7 +37,10 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -80,17 +84,15 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 		eventObjSide = syncSideSource
 	}
 
+	// normalize the deprecated cleanup bool and the cleanup policy field into a single policy.
+	policy := relRes.EffectiveCleanupPolicy()
+
 	// find the all objects on the origin side that match the given criteria
 	resolvedObjects, err := resolveRelatedResourceObjects(ctx, origin, dest, relRes)
 	if err != nil {
 		return false, fmt.Errorf("failed to get resolve origin objects: %w", err)
 	}
 
-	// no objects were found yet, that's okay
-	if len(resolvedObjects) == 0 {
-		return false, nil
-	}
-
 	slices.SortStableFunc(resolvedObjects, func(a, b resolvedObject) int {
 		aKey := ctrlruntimeclient.ObjectKeyFromObject(a.original).String()
 		bKey := ctrlruntimeclient.ObjectKeyFromObject(b.original).String()
@@ -106,7 +108,27 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 		return false, fmt.Errorf("failed to lookup %v: %w", projectedGVR, err)
 	}
 
-	for idx, resolved := range resolvedObjects {
+	// The primary object (always the kcp side) owns these related copies; stamp its coordinates
+	// onto every copy so that all copies of this (primary, identifier) can be enumerated later to
+	// prune stale copies or tear them all down.
+	primary := remote.object
+	destLabels := relatedCopyLabels(primary, relRes.Identifier, s.agentName)
+	destAnnotations := relatedCopyAnnotations(primary)
+
+	// remember which destination copies we (re)synced this pass, so a MatchOrigin prune can delete
+	// the copies that no longer have a matching origin object.
+	synced := sets.New[string]()
+
+	// We "forward" the deletion to the related objects only if the primary is already in deletion
+	// and the related object either originated from the user (so on the service cluster we just
+	// have a useless copy once the main object has been cleared up) OR the admin explicitly opted
+	// into a cleanup policy that removes copies on primary deletion.
+	forceDelete := primaryDeleting &&
+		(policy == syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion ||
+			policy == syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin ||
+			relRes.Origin == syncagentv1alpha1.RelatedResourceOriginKcp)
+
+	for _, resolved := range resolvedObjects {
 		destObject := &unstructured.Unstructured{}
 		destObject.SetAPIVersion(projectedGVK.GroupVersion().String())
 		destObject.SetKind(projectedGVK.Kind)
@@ -127,12 +149,6 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 			object:      destObject,
 		}
 
-		// We "forward" the deletion to the related objects only if the primary is already in deletion
-		// and the related object either originated from the user (so on the service cluster we just
-		// have a useless copy once the main object has been cleared up) OR the admin explicitly opted
-		// into the cleanup procedure to ensure that copies of the related object are removed.
-		forceDelete := primaryDeleting && (relRes.Cleanup || relRes.Origin == syncagentv1alpha1.RelatedResourceOriginKcp)
-
 		// When status sync is enabled, include "status" in subresources so it is stripped from
 		// the spec patch (avoiding a no-op write on resources that have a status subresource).
 		// The status is then separately written via the status subresource endpoint by syncStatusForward.
@@ -169,6 +185,9 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 			mutator: s.relatedMutators[relRes.Identifier],
 			// we never want to store sync-related metadata inside kcp
 			metadataOnDestination: false,
+			// stamp owning-primary provenance so that the copies can be enumerated for pruning.
+			destLabels:      destLabels,
+			destAnnotations: destAnnotations,
 			// events are always created on the kcp side
 			eventObjSide: eventObjSide,
 			// force deletion of related resources when the primary object is being deleted
@@ -188,43 +207,188 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su
 		// too many unnecessary requeues.
 		requeue = requeue || req
 
-		// now that the related object was successfully synced, we can remember its details on the
-		// main object
-		if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginService {
-			// TODO: Improve this logic, the added index is just a hack until we find a better solution
-			// to let the user know about the related object (this annotation is not relevant for the
-			// syncing logic, it's purely for the end-user).
-			annotation := fmt.Sprintf("%s%s.%d", relatedObjectAnnotationPrefix, relRes.Identifier, idx)
-
-			value, err := json.Marshal(relatedObjectAnnotation{
-				Namespace:  resolved.destination.Namespace,
-				Name:       resolved.destination.Name,
-				APIVersion: resolved.original.GetAPIVersion(),
-				Kind:       resolved.original.GetKind(),
-			})
-			if err != nil {
-				return false, fmt.Errorf("failed to encode related object annotation: %w", err)
+		synced.Insert(relatedCopyKey(resolved.destination.Namespace, resolved.destination.Name))
+	}
+
+	// Remember the related objects on the primary object for the end-user. This is rebuilt from the
+	// freshly-resolved set so stale entries for pruned objects do not linger.
+	if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginService {
+		annRequeue, err := s.rememberRelatedObjects(ctx, log, remote, relRes.Identifier, resolvedObjects)
+		if err != nil {
+			return false, err
+		}
+
+		// we updated the main object, so we requeue immediately because successive patches would
+		// fail anyway; the prune below then runs on the next reconciliation.
+		if annRequeue {
+			return true, nil
+		}
+	}
+
+	// Prune / teardown destination copies as configured. Only objects carrying our provenance
+	// labels are ever considered, and we only ever act on the destination client, so the original
+	// origin-side objects are never touched.
+	switch {
+	case primaryDeleting && (policy == syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion ||
+		policy == syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin):
+		// On primary teardown, delete ALL labelled copies. This is a superset of the per-object
+		// deletion performed in the loop above and additionally reclaims copies whose origin object
+		// had already disappeared mid-life (which the loop can no longer resolve).
+		selector := relatedCopySelector(primary, relRes.Identifier, s.agentName)
+
+		pruneRequeue, err := s.pruneRelatedCopies(ctx, log, dest, primary, projectedGVK, selector, nil, true)
+		if err != nil {
+			return false, fmt.Errorf("failed to tear down related copies: %w", err)
+		}
+
+		requeue = requeue || pruneRequeue
+
+	case !primaryDeleting && policy == syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin:
+		// Keep the destination set equal to the origin set: prune every labelled copy whose origin
+		// object was not resolved this pass.
+		selector := relatedCopySelector(primary, relRes.Identifier, s.agentName)
+
+		pruneRequeue, err := s.pruneRelatedCopies(ctx, log, dest, primary, projectedGVK, selector, synced, false)
+		if err != nil {
+			return false, fmt.Errorf("failed to prune related copies: %w", err)
+		}
+
+		requeue = requeue || pruneRequeue
+	}
+
+	return requeue, nil
+}
+
+// relatedCopyKey builds the namespace/name key used to track and match destination copies.
+func relatedCopyKey(namespace, name string) string {
+	return namespace + "/" + name
+}
+
+// rememberRelatedObjects writes the human-facing provenance annotations onto the primary object,
+// one per related copy (indexed). It rebuilds the full set for the given identifier from the
+// currently resolved objects, so annotations for pruned copies are removed and do not accumulate.
+// It reports requeue=true when it patched the primary object.
+func (s *ResourceSyncer) rememberRelatedObjects(ctx context.Context, log *zap.SugaredLogger, remote syncSide, identifier string, resolvedObjects []resolvedObject) (requeue bool, err error) {
+	// TODO: Improve this logic, the added index is just a hack until we find a better solution to
+	// let the user know about the related object (this annotation is not relevant for the syncing
+	// logic, it's purely for the end-user).
+	prefix := fmt.Sprintf("%s%s.", relatedObjectAnnotationPrefix, identifier)
+
+	desired := map[string]string{}
+	for idx, resolved := range resolvedObjects {
+		value, err := json.Marshal(relatedObjectAnnotation{
+			Namespace:  resolved.destination.Namespace,
+			Name:       resolved.destination.Name,
+			APIVersion: resolved.original.GetAPIVersion(),
+			Kind:       resolved.original.GetKind(),
+		})
+		if err != nil {
+			return false, fmt.Errorf("failed to encode related object annotation: %w", err)
+		}
+
+		desired[fmt.Sprintf("%s%d", prefix, idx)] = string(value)
+	}
+
+	annotations := remote.object.GetAnnotations()
+	if annotations == nil {
+		annotations = map[string]string{}
+	}
+
+	// determine whether the existing annotations for this identifier already match the desired set.
+	changed := false
+	for key := range annotations {
+		if strings.HasPrefix(key, prefix) {
+			if _, ok := desired[key]; !ok {
+				changed = true
+				break
+			}
+		}
+	}
+	if !changed {
+		for key, value := range desired {
+			if annotations[key] != value {
+				changed = true
+				break
 			}
+		}
+	}
 
-			annotations := remote.object.GetAnnotations()
-			existing := annotations[annotation]
+	if !changed {
+		return false, nil
+	}
 
-			if existing != string(value) {
-				oldState := remote.object.DeepCopy()
+	oldState := remote.object.DeepCopy()
 
-				annotations[annotation] = string(value)
-				remote.object.SetAnnotations(annotations)
+	// drop all existing entries for this identifier, then add the freshly computed ones.
+	for key := range annotations {
+		if strings.HasPrefix(key, prefix) {
+			delete(annotations, key)
+		}
+	}
+	maps.Copy(annotations, desired)
+	remote.object.SetAnnotations(annotations)
 
-				log.Debug("Remembering related object in main object…")
-				if err := remote.client.Patch(ctx, remote.object, ctrlruntimeclient.MergeFrom(oldState)); err != nil {
-					return false, fmt.Errorf("failed to update related data in remote object: %w", err)
-				}
+	log.Debug("Remembering related objects in main object…")
+	if err := remote.client.Patch(ctx, remote.object, ctrlruntimeclient.MergeFrom(oldState)); err != nil {
+		return false, fmt.Errorf("failed to update related data in remote object: %w", err)
+	}
 
-				// requeue (since this updated the main object, we do actually want to
-				// requeue immediately because successive patches would fail anyway)
-				return true, nil
+	return true, nil
+}
+
+// pruneRelatedCopies lists the destination copies matching the given (primary + identifier + agent)
+// selector and deletes those that are no longer wanted. When deleteAll is true, every matching copy
+// is deleted (primary teardown); otherwise only copies whose key is not in the keep set are deleted
+// (mid-life prune). It only ever operates on the destination client, so origin objects are never
+// touched, and it only ever sees objects that carry our provenance labels, so hand-created objects
+// are never in scope.
+func (s *ResourceSyncer) pruneRelatedCopies(ctx context.Context, log *zap.SugaredLogger, dest syncSide, primary *unstructured.Unstructured, projectedGVK schema.GroupVersionKind, selector labels.Selector, keep sets.Set[string], deleteAll bool) (requeue bool, err error) {
+	list := &unstructured.UnstructuredList{}
+	list.SetAPIVersion(projectedGVK.GroupVersion().String())
+	list.SetKind(projectedGVK.Kind + "List")
+
+	// List cluster-wide (across all namespaces). A related resource can map its copies into a
+	// namespace other than the primary's (via spec.object.namespace rewrites), so scoping the List
+	// to a single namespace would miss — and therefore never prune — copies that landed elsewhere.
+	// The label selector already scopes the result to this primary + identifier + agent, so only
+	// copies this agent created for this primary can match.
+	listOpts := []ctrlruntimeclient.ListOption{
+		ctrlruntimeclient.MatchingLabelsSelector{Selector: selector},
+	}
+
+	if err := dest.client.List(ctx, list, listOpts...); err != nil {
+		return false, fmt.Errorf("failed to list related copies: %w", err)
+	}
+
+	recorder := recorderFromContext(ctx)
+
+	for i := range list.Items {
+		item := &list.Items[i]
+
+		if !deleteAll && keep.Has(relatedCopyKey(item.GetNamespace(), item.GetName())) {
+			continue
+		}
+
+		// already being deleted; come back once it is gone.
+		if item.GetDeletionTimestamp() != nil {
+			requeue = true
+			continue
+		}
+
+		log.Debugw("Pruning related object copy…", "namespace", item.GetNamespace(), "name", item.GetName())
+		if err := dest.client.Delete(ctx, item); err != nil {
+			if apierrors.IsNotFound(err) {
+				continue
 			}
+
+			return false, fmt.Errorf("failed to delete related copy: %w", err)
+		}
+
+		if recorder != nil {
+			recorder.Eventf(primary, corev1.EventTypeNormal, "ObjectCleanup", "Deleted orphaned copy %s/%s of a related resource.", item.GetNamespace(), item.GetName())
 		}
+
+		requeue = true
 	}
 
 	return requeue, nil
diff --git a/internal/sync/syncer_related_test.go b/internal/sync/syncer_related_test.go
index 92dcd248..26f1bb92 100644
--- a/internal/sync/syncer_related_test.go
+++ b/internal/sync/syncer_related_test.go
@@ -24,8 +24,114 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 )
 
+func TestEffectiveCleanupPolicy(t *testing.T) {
+	testcases := []struct {
+		name     string
+		cleanup  bool
+		policy   syncagentv1alpha1.RelatedResourceCleanupPolicy
+		expected syncagentv1alpha1.RelatedResourceCleanupPolicy
+	}{
+		{
+			name:     "no cleanup, no policy defaults to Orphan",
+			expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan,
+		},
+		{
+			name:     "legacy cleanup:true maps to OnPrimaryDeletion",
+			cleanup:  true,
+			expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion,
+		},
+		{
+			name:     "explicit Orphan wins over cleanup:false",
+			policy:   syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan,
+			expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan,
+		},
+		{
+			name:     "explicit policy wins over legacy cleanup:true",
+			cleanup:  true,
+			policy:   syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+			expected: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+		},
+		{
+			name:     "explicit OnPrimaryDeletion without cleanup",
+			policy:   syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion,
+			expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion,
+		},
+		{
+			name:     "explicit MatchOrigin without cleanup",
+			policy:   syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+			expected: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			spec := &syncagentv1alpha1.RelatedResourceSpec{
+				Cleanup:       tc.cleanup,
+				CleanupPolicy: tc.policy,
+			}
+
+			if got := spec.EffectiveCleanupPolicy(); got != tc.expected {
+				t.Errorf("expected %q, got %q", tc.expected, got)
+			}
+		})
+	}
+}
+
+func TestRelatedCopyLabelsSelectorRoundTrip(t *testing.T) {
+	primary := &unstructured.Unstructured{}
+	primary.SetName("my-primary")
+	primary.SetNamespace("some-namespace")
+
+	const (
+		identifier = "credentials"
+		agentName  = "agent-1"
+	)
+
+	labelSet := relatedCopyLabels(primary, identifier, agentName)
+
+	// the labels produced for a copy must match the selector used to find them again.
+	selector := relatedCopySelector(primary, identifier, agentName)
+	if !selector.Matches(labels.Set(labelSet)) {
+		t.Errorf("selector %q does not match its own labels %v", selector, labelSet)
+	}
+
+	// a selector for a different identifier must not match.
+	otherIdentifier := relatedCopySelector(primary, "other", agentName)
+	if otherIdentifier.Matches(labels.Set(labelSet)) {
+		t.Errorf("selector for a different identifier unexpectedly matched labels %v", labelSet)
+	}
+
+	// a selector for a different agent must not match.
+	otherAgent := relatedCopySelector(primary, identifier, "agent-2")
+	if otherAgent.Matches(labels.Set(labelSet)) {
+		t.Errorf("selector for a different agent unexpectedly matched labels %v", labelSet)
+	}
+
+	// a selector for a different primary object must not match.
+	otherPrimary := &unstructured.Unstructured{}
+	otherPrimary.SetName("other-primary")
+	otherPrimary.SetNamespace("some-namespace")
+	if relatedCopySelector(otherPrimary, identifier, agentName).Matches(labels.Set(labelSet)) {
+		t.Errorf("selector for a different primary unexpectedly matched labels %v", labelSet)
+	}
+
+	// a cluster-scoped primary (no namespace) must omit the namespace-hash label and still
+	// round-trip.
+	clusterPrimary := &unstructured.Unstructured{}
+	clusterPrimary.SetName("cluster-primary")
+	clusterLabels := relatedCopyLabels(clusterPrimary, identifier, agentName)
+	if _, ok := clusterLabels[relatedPrimaryNamespaceHashLabel]; ok {
+		t.Error("expected no namespace-hash label for a cluster-scoped primary")
+	}
+	if !relatedCopySelector(clusterPrimary, identifier, agentName).Matches(labels.Set(clusterLabels)) {
+		t.Errorf("cluster-scoped selector does not match its own labels %v", clusterLabels)
+	}
+}
+
 func TestResolveRelatedResourceObjects(t *testing.T) {
 	// in kcp
 	primaryObject := newUnstructured(&dummyv1alpha1.Thing{
diff --git a/internal/sync/types.go b/internal/sync/types.go
index e048ec58..d3c6012c 100644
--- a/internal/sync/types.go
+++ b/internal/sync/types.go
@@ -53,6 +53,21 @@ const (
 	// full annotation name, the annotation value is a JSON string containing GVK and
 	// metadata of the related object.
 	relatedObjectAnnotationPrefix = "related-resources.syncagent.kcp.io/"
+
+	// The following labels/annotations are put on the destination copies of related resources
+	// to link them back to their owning primary object and the related resource identifier.
+	// They allow the agent to List all copies belonging to a specific primary + identifier so
+	// that it can prune copies whose origin object no longer exists (cleanupPolicy: MatchOrigin)
+	// or delete all copies on primary teardown. Names/namespaces can exceed the 63-character
+	// label limit or contain invalid characters, so they are hashed; the plaintext values are
+	// kept as annotations for humans.
+
+	relatedPrimaryNamespaceHashLabel = "syncagent.kcp.io/related-primary-namespace-hash"
+	relatedPrimaryNameHashLabel      = "syncagent.kcp.io/related-primary-name-hash"
+	relatedIdentifierLabel           = "syncagent.kcp.io/related-identifier"
+
+	relatedPrimaryNamespaceAnnotation = "syncagent.kcp.io/related-primary-namespace"
+	relatedPrimaryNameAnnotation      = "syncagent.kcp.io/related-primary-name"
 )
 
 func OwnedBy(obj ctrlruntimeclient.Object, agentName string) bool {
diff --git a/sdk/apis/syncagent/v1alpha1/published_resource.go b/sdk/apis/syncagent/v1alpha1/published_resource.go
index 77737290..b3fa5095 100644
--- a/sdk/apis/syncagent/v1alpha1/published_resource.go
+++ b/sdk/apis/syncagent/v1alpha1/published_resource.go
@@ -201,6 +201,25 @@ const (
 	RelatedResourceOriginKcp     RelatedResourceOrigin = "kcp"
 )
 
+// RelatedResourceCleanupPolicy controls when the syncagent deletes the destination copies of a
+// related resource. The original object on the origin side is never deleted, regardless of the
+// chosen policy.
+//
+// +kubebuilder:validation:Enum=Orphan;OnPrimaryDeletion;MatchOrigin
+type RelatedResourceCleanupPolicy string
+
+const (
+	// RelatedResourceCleanupPolicyOrphan never deletes copies (default; == legacy cleanup:false).
+	RelatedResourceCleanupPolicyOrphan RelatedResourceCleanupPolicy = "Orphan"
+	// RelatedResourceCleanupPolicyOnPrimaryDeletion deletes copies only when the primary object
+	// is deleted (== legacy cleanup:true).
+	RelatedResourceCleanupPolicyOnPrimaryDeletion RelatedResourceCleanupPolicy = "OnPrimaryDeletion"
+	// RelatedResourceCleanupPolicyMatchOrigin keeps the destination set equal to the origin set:
+	// a copy is pruned as soon as its origin object is gone, and all copies are deleted when the
+	// primary object is deleted.
+	RelatedResourceCleanupPolicyMatchOrigin RelatedResourceCleanupPolicy = "MatchOrigin"
+)
+
 // RelatedResourceSpec describes a single related resource, which might point to
 // any number of actual Kubernetes objects.
 //
@@ -211,6 +230,8 @@ const (
 // group is included here because when an identityHash is used, core/v1 cannot possible be targetted
 // +kubebuilder:validation:XValidation:rule="!has(self.identityHash) || (has(self.group) && has(self.version) && has(self.resource))",message="identity hashes can only be used with GVRs"
 // +kubebuilder:validation:XValidation:rule="!(self.origin == 'service' && has(self.syncStatus) && self.syncStatus) || has(self.watch)",message="watch must be configured when origin is service and syncStatus is true"
+// +kubebuilder:validation:XValidation:rule="!(has(self.cleanupPolicy) && self.cleanupPolicy == 'MatchOrigin' && self.origin == 'service') || has(self.watch)",message="watch must be configured when cleanupPolicy is MatchOrigin and origin is service"
+// +kubebuilder:validation:XValidation:rule="!(has(self.cleanup) && self.cleanup && has(self.cleanupPolicy) && self.cleanupPolicy == 'Orphan')",message="cleanup:true conflicts with cleanupPolicy: Orphan"
 type RelatedResourceSpec struct {
 	// Identifier is a unique name for this related resource. The name must be unique within one
 	// PublishedResource and is the key by which consumers (end users) can identify and consume the
@@ -248,8 +269,24 @@ type RelatedResourceSpec struct {
 	// the destination side (i.e. the original related object will not be touched, regardless of this
 	// option). Leaving this disabled, the syncagent will only create copies of the related objects,
 	// but never delete them itself.
+	//
+	// Deprecated: use CleanupPolicy instead. When CleanupPolicy is empty, cleanup:true is treated
+	// as OnPrimaryDeletion and cleanup:false as Orphan.
 	Cleanup bool `json:"cleanup,omitempty"`
 
+	// CleanupPolicy controls when the syncagent deletes destination copies of this related
+	// resource. The original object on the origin side is never deleted.
+	//
+	//   - Orphan (default): copies are never deleted by the agent.
+	//   - OnPrimaryDeletion: copies are deleted only when the primary object is deleted.
+	//   - MatchOrigin: copies are pruned as soon as their origin object no longer exists, and
+	//     all copies are deleted when the primary object is deleted.
+	//
+	// When left empty, the deprecated Cleanup field is used to derive the effective policy.
+	//
+	// +optional
+	CleanupPolicy RelatedResourceCleanupPolicy `json:"cleanupPolicy,omitempty"`
+
 	// Projection is used to change the GVK of a related resource on the opposite side of
 	// its origin.
 	// All fields in the projection are optional. If a field is set, it will overwrite
@@ -285,6 +322,21 @@ type RelatedResourceSpec struct {
 	SyncStatus bool `json:"syncStatus,omitempty"`
 }
 
+// EffectiveCleanupPolicy normalizes the deprecated Cleanup bool and the CleanupPolicy field into
+// a single effective policy. CleanupPolicy takes precedence; when it is empty, cleanup:true maps
+// to OnPrimaryDeletion and cleanup:false to Orphan.
+func (r *RelatedResourceSpec) EffectiveCleanupPolicy() RelatedResourceCleanupPolicy {
+	if r.CleanupPolicy != "" {
+		return r.CleanupPolicy
+	}
+
+	if r.Cleanup {
+		return RelatedResourceCleanupPolicyOnPrimaryDeletion
+	}
+
+	return RelatedResourceCleanupPolicyOrphan
+}
+
 // RelatedResourceWatch configures how the watch handler maps a changed related resource
 // back to its owning primary object.
 // Exactly one of ByOwner or BySelector must be set.
diff --git a/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go b/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go
index 5517ac2b..4196a14b 100644
--- a/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go
+++ b/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go
@@ -25,19 +25,20 @@ import (
 // RelatedResourceSpecApplyConfiguration represents a declarative configuration of the RelatedResourceSpec type for use
 // with apply.
 type RelatedResourceSpecApplyConfiguration struct {
-	Identifier   *string                                      `json:"identifier,omitempty"`
-	Origin       *syncagentv1alpha1.RelatedResourceOrigin     `json:"origin,omitempty"`
-	Group        *string                                      `json:"group,omitempty"`
-	Version      *string                                      `json:"version,omitempty"`
-	Resource     *string                                      `json:"resource,omitempty"`
-	Kind         *string                                      `json:"kind,omitempty"`
-	IdentityHash *string                                      `json:"identityHash,omitempty"`
-	Cleanup      *bool                                        `json:"cleanup,omitempty"`
-	Projection   *RelatedResourceProjectionApplyConfiguration `json:"projection,omitempty"`
-	Object       *RelatedResourceObjectApplyConfiguration     `json:"object,omitempty"`
-	Mutation     *ResourceMutationSpecApplyConfiguration      `json:"mutation,omitempty"`
-	Watch        *RelatedResourceWatchApplyConfiguration      `json:"watch,omitempty"`
-	SyncStatus   *bool                                        `json:"syncStatus,omitempty"`
+	Identifier    *string                                         `json:"identifier,omitempty"`
+	Origin        *syncagentv1alpha1.RelatedResourceOrigin        `json:"origin,omitempty"`
+	Group         *string                                         `json:"group,omitempty"`
+	Version       *string                                         `json:"version,omitempty"`
+	Resource      *string                                         `json:"resource,omitempty"`
+	Kind          *string                                         `json:"kind,omitempty"`
+	IdentityHash  *string                                         `json:"identityHash,omitempty"`
+	Cleanup       *bool                                           `json:"cleanup,omitempty"`
+	CleanupPolicy *syncagentv1alpha1.RelatedResourceCleanupPolicy `json:"cleanupPolicy,omitempty"`
+	Projection    *RelatedResourceProjectionApplyConfiguration    `json:"projection,omitempty"`
+	Object        *RelatedResourceObjectApplyConfiguration        `json:"object,omitempty"`
+	Mutation      *ResourceMutationSpecApplyConfiguration         `json:"mutation,omitempty"`
+	Watch         *RelatedResourceWatchApplyConfiguration         `json:"watch,omitempty"`
+	SyncStatus    *bool                                           `json:"syncStatus,omitempty"`
 }
 
 // RelatedResourceSpecApplyConfiguration constructs a declarative configuration of the RelatedResourceSpec type for use with
@@ -110,6 +111,14 @@ func (b *RelatedResourceSpecApplyConfiguration) WithCleanup(value bool) *Related
 	return b
 }
 
+// WithCleanupPolicy sets the CleanupPolicy field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the CleanupPolicy field is set to the value of the last call.
+func (b *RelatedResourceSpecApplyConfiguration) WithCleanupPolicy(value syncagentv1alpha1.RelatedResourceCleanupPolicy) *RelatedResourceSpecApplyConfiguration {
+	b.CleanupPolicy = &value
+	return b
+}
+
 // WithProjection sets the Projection field in the declarative configuration to the given value
 // and returns the receiver, so that objects can be built by chaining "With" function invocations.
 // If called multiple times, the Projection field is set to the value of the last call.
diff --git a/test/e2e/sdk/cel_validations_test.go b/test/e2e/sdk/cel_validations_test.go
index 96da0369..f91a6401 100644
--- a/test/e2e/sdk/cel_validations_test.go
+++ b/test/e2e/sdk/cel_validations_test.go
@@ -209,6 +209,61 @@ func TestValidateRelatedResourceSpec(t *testing.T) {
 				SyncStatus: true,
 			},
 		},
+		{
+			name:  "cleanupPolicy MatchOrigin with origin service requires watch",
+			valid: false,
+			spec: syncagentv1alpha1.RelatedResourceSpec{
+				Origin:        syncagentv1alpha1.RelatedResourceOriginService,
+				Resource:      "things",
+				Version:       "v1",
+				CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+			},
+		},
+		{
+			name:  "cleanupPolicy MatchOrigin with origin service and watch is valid",
+			valid: true,
+			spec: syncagentv1alpha1.RelatedResourceSpec{
+				Origin:        syncagentv1alpha1.RelatedResourceOriginService,
+				Resource:      "things",
+				Version:       "v1",
+				CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+				Watch: &syncagentv1alpha1.RelatedResourceWatch{
+					ByOwner: &syncagentv1alpha1.RelatedResourceWatchByOwner{},
+				},
+			},
+		},
+		{
+			name:  "cleanupPolicy MatchOrigin with origin kcp does not require watch",
+			valid: true,
+			spec: syncagentv1alpha1.RelatedResourceSpec{
+				Origin:        syncagentv1alpha1.RelatedResourceOriginKcp,
+				Resource:      "things",
+				Version:       "v1",
+				CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+			},
+		},
+		{
+			name:  "cleanup true conflicts with cleanupPolicy Orphan",
+			valid: false,
+			spec: syncagentv1alpha1.RelatedResourceSpec{
+				Origin:        syncagentv1alpha1.RelatedResourceOriginService,
+				Resource:      "things",
+				Version:       "v1",
+				Cleanup:       true,
+				CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan,
+			},
+		},
+		{
+			name:  "cleanup true with cleanupPolicy OnPrimaryDeletion is valid",
+			valid: true,
+			spec: syncagentv1alpha1.RelatedResourceSpec{
+				Origin:        syncagentv1alpha1.RelatedResourceOriginService,
+				Resource:      "things",
+				Version:       "v1",
+				Cleanup:       true,
+				CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion,
+			},
+		},
 	}
 
 	alphaNum := regexp.MustCompile(`[^a-z0-9]`)
diff --git a/test/e2e/sync/cleanup_test.go b/test/e2e/sync/cleanup_test.go
index a1891b10..faca8175 100644
--- a/test/e2e/sync/cleanup_test.go
+++ b/test/e2e/sync/cleanup_test.go
@@ -414,3 +414,399 @@ spec:
 		t.Errorf("Failed to get Secret on service cluster: %v", err)
 	}
 }
+
+// matchOriginPublishedResource builds a PublishedResource that publishes CronTabs with a related
+// Secret (found via a label selector so that a whole set of objects can match) using
+// cleanupPolicy: MatchOrigin. The related object name is preserved on both sides.
+func matchOriginPublishedResource(origin syncagentv1alpha1.RelatedResourceOrigin, watchLabels map[string]string) *syncagentv1alpha1.PublishedResource {
+	return &syncagentv1alpha1.PublishedResource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "publish-crontabs",
+		},
+		Spec: syncagentv1alpha1.PublishedResourceSpec{
+			Resource: syncagentv1alpha1.SourceResourceDescriptor{
+				APIGroup: "example.com",
+				Version:  "v1",
+				Kind:     "CronTab",
+			},
+			Naming: &syncagentv1alpha1.ResourceNaming{
+				Name:      "{{ .Object.metadata.name }}",
+				Namespace: "synced-{{ .Object.metadata.namespace }}",
+			},
+			Projection: &syncagentv1alpha1.ResourceProjection{
+				Group: "kcp.example.com",
+			},
+			Related: []syncagentv1alpha1.RelatedResourceSpec{
+				{
+					Identifier:    "credentials",
+					Origin:        origin,
+					Kind:          "Secret",
+					CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin,
+					Object: syncagentv1alpha1.RelatedResourceObject{
+						RelatedResourceObjectSpec: syncagentv1alpha1.RelatedResourceObjectSpec{
+							Selector: &syncagentv1alpha1.RelatedResourceObjectSelector{
+								LabelSelector: metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"app": "credentials",
+									},
+								},
+								Rewrite: syncagentv1alpha1.RelatedResourceSelectorRewrite{
+									Template: &syncagentv1alpha1.TemplateExpression{
+										// keep the same name on both sides
+										Template: "{{ .Value }}",
+									},
+								},
+							},
+						},
+					},
+					Watch: &syncagentv1alpha1.RelatedResourceWatch{
+						BySelector: &metav1.LabelSelector{
+							MatchLabels: watchLabels,
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func credentialSecret(name, namespace string) *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+			Labels: map[string]string{
+				"app": "credentials",
+			},
+		},
+		Data: map[string][]byte{
+			"password": []byte("hunter2"),
+		},
+		Type: corev1.SecretTypeOpaque,
+	}
+}
+
+func waitForSecret(t *testing.T, ctx context.Context, client ctrlruntimeclient.Client, key types.NamespacedName) {
+	t.Helper()
+
+	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) {
+		return client.Get(ctx, key, &corev1.Secret{}) == nil, nil
+	})
+	if err != nil {
+		t.Fatalf("Secret %v did not appear: %v", key, err)
+	}
+}
+
+func waitForSecretGone(t *testing.T, ctx context.Context, client ctrlruntimeclient.Client, key types.NamespacedName) {
+	t.Helper()
+
+	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) {
+		return apierrors.IsNotFound(client.Get(ctx, key, &corev1.Secret{})), nil
+	})
+	if err != nil {
+		t.Fatalf("Secret %v was not pruned: %v", key, err)
+	}
+}
+
+func requireSecretExists(t *testing.T, ctx context.Context, client ctrlruntimeclient.Client, key types.NamespacedName, msg string) {
+	t.Helper()
+
+	if err := client.Get(ctx, key, &corev1.Secret{}); err != nil {
+		t.Errorf("%s: %v", msg, err)
+	}
+}
+
+// TestRelatedResourceMatchOriginServiceOrigin verifies that with cleanupPolicy: MatchOrigin and
+// origin: service, deleting one source object mid-life prunes exactly its kcp copy while the other
+// copy and the surviving source object remain, and that primary teardown removes all kcp copies
+// without touching the service-cluster sources.
+func TestRelatedResourceMatchOriginServiceOrigin(t *testing.T) {
+	const (
+		apiExportName = "kcp.example.com"
+		orgWorkspace  = "match-origin-service"
+	)
+
+	ctx := t.Context()
+	ctrlruntime.SetLogger(logr.Discard())
+
+	orgKubconfig := utils.CreateOrganization(t, ctx, orgWorkspace, apiExportName)
+
+	envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{
+		"test/crds/crontab.yaml",
+	})
+
+	watchLabels := map[string]string{"syncagent-e2e": "find-me"}
+
+	t.Log("Publishing CRDs…")
+	if err := envtestClient.Create(ctx, matchOriginPublishedResource(syncagentv1alpha1.RelatedResourceOriginService, watchLabels)); err != nil {
+		t.Fatalf("Failed to create PublishedResource: %v", err)
+	}
+
+	utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "")
+
+	kcpClusterClient := utils.GetKcpAdminClusterClient(t)
+	teamClusterPath := logicalcluster.NewPath("root").Join(orgWorkspace).Join("team-1")
+	teamClient := kcpClusterClient.Cluster(teamClusterPath)
+
+	utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{
+		Group:   "kcp.example.com",
+		Version: "v1",
+		Kind:    "CronTab",
+	})
+
+	t.Log("Creating CronTab in kcp…")
+	crontab := utils.YAMLToUnstructured(t, `
+apiVersion: kcp.example.com/v1
+kind: CronTab
+metadata:
+  namespace: default
+  name: my-crontab
+spec:
+  cronSpec: '* * *'
+  image: ubuntu:latest
+`)
+	crontab.SetLabels(watchLabels)
+
+	if err := teamClient.Create(ctx, crontab); err != nil {
+		t.Fatalf("Failed to create CronTab in kcp: %v", err)
+	}
+
+	// wait for the CronTab to sync down (this also creates the synced-default namespace)
+	t.Log("Waiting for CronTab to be synced to service cluster…")
+	serviceCrontab := &unstructured.Unstructured{}
+	serviceCrontab.SetAPIVersion("example.com/v1")
+	serviceCrontab.SetKind("CronTab")
+	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) {
+		return envtestClient.Get(ctx, types.NamespacedName{Namespace: "synced-default", Name: "my-crontab"}, serviceCrontab) == nil, nil
+	})
+	if err != nil {
+		t.Fatalf("CronTab was not synced to service cluster: %v", err)
+	}
+
+	// create two related Secrets on the service cluster
+	t.Log("Creating credential Secrets on service cluster…")
+	for _, name := range []string{"cred-1", "cred-2"} {
+		if err := envtestClient.Create(ctx, credentialSecret(name, "synced-default")); err != nil {
+			t.Fatalf("Failed to create Secret %s: %v", name, err)
+		}
+	}
+
+	// wait for both to be synced up to kcp
+	t.Log("Waiting for both Secrets to be synced to kcp…")
+	waitForSecret(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"})
+	waitForSecret(t, ctx, teamClient, types.NamespacedName{Name: "cred-2", Namespace: "default"})
+
+	// delete one source object while the primary still lives
+	t.Log("Deleting cred-1 on the service cluster…")
+	if err := envtestClient.Delete(ctx, credentialSecret("cred-1", "synced-default")); err != nil {
+		t.Fatalf("Failed to delete source Secret cred-1: %v", err)
+	}
+
+	// its kcp copy should be pruned…
+	t.Log("Verifying that the cred-1 copy in kcp is pruned…")
+	waitForSecretGone(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"})
+
+	// …while the other copy and the surviving source object remain untouched.
+	requireSecretExists(t, ctx, teamClient, types.NamespacedName{Name: "cred-2", Namespace: "default"}, "cred-2 copy in kcp should remain")
+	requireSecretExists(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"}, "cred-2 source on service cluster should remain")
+
+	// now tear down the primary
+	t.Log("Deleting CronTab in kcp…")
+	if err := teamClient.Delete(ctx, crontab); err != nil {
+		t.Fatalf("Failed to delete CronTab: %v", err)
+	}
+
+	// all kcp copies should be gone
+	t.Log("Verifying that all related copies in kcp are gone…")
+	waitForSecretGone(t, ctx, teamClient, types.NamespacedName{Name: "cred-2", Namespace: "default"})
+
+	// the service-cluster source must never be touched
+	requireSecretExists(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"}, "cred-2 source on service cluster should remain after teardown")
+}
+
+// TestRelatedResourceMatchOriginTeardownReclaimsOrphan reproduces the teardown-orphan bug: a source
+// object is deleted mid-life (orphaning its copy), then the primary is deleted. Even the orphaned
+// copy must be reclaimed on teardown.
+func TestRelatedResourceMatchOriginTeardownReclaimsOrphan(t *testing.T) {
+	const (
+		apiExportName = "kcp.example.com"
+		orgWorkspace  = "match-origin-orphan"
+	)
+
+	ctx := t.Context()
+	ctrlruntime.SetLogger(logr.Discard())
+
+	orgKubconfig := utils.CreateOrganization(t, ctx, orgWorkspace, apiExportName)
+
+	envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{
+		"test/crds/crontab.yaml",
+	})
+
+	watchLabels := map[string]string{"syncagent-e2e": "find-me"}
+
+	t.Log("Publishing CRDs…")
+	if err := envtestClient.Create(ctx, matchOriginPublishedResource(syncagentv1alpha1.RelatedResourceOriginService, watchLabels)); err != nil {
+		t.Fatalf("Failed to create PublishedResource: %v", err)
+	}
+
+	utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "")
+
+	kcpClusterClient := utils.GetKcpAdminClusterClient(t)
+	teamClusterPath := logicalcluster.NewPath("root").Join(orgWorkspace).Join("team-1")
+	teamClient := kcpClusterClient.Cluster(teamClusterPath)
+
+	utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{
+		Group:   "kcp.example.com",
+		Version: "v1",
+		Kind:    "CronTab",
+	})
+
+	t.Log("Creating CronTab in kcp…")
+	crontab := utils.YAMLToUnstructured(t, `
+apiVersion: kcp.example.com/v1
+kind: CronTab
+metadata:
+  namespace: default
+  name: my-crontab
+spec:
+  cronSpec: '* * *'
+  image: ubuntu:latest
+`)
+	crontab.SetLabels(watchLabels)
+
+	if err := teamClient.Create(ctx, crontab); err != nil {
+		t.Fatalf("Failed to create CronTab in kcp: %v", err)
+	}
+
+	t.Log("Waiting for CronTab to be synced to service cluster…")
+	serviceCrontab := &unstructured.Unstructured{}
+	serviceCrontab.SetAPIVersion("example.com/v1")
+	serviceCrontab.SetKind("CronTab")
+	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) {
+		return envtestClient.Get(ctx, types.NamespacedName{Namespace: "synced-default", Name: "my-crontab"}, serviceCrontab) == nil, nil
+	})
+	if err != nil {
+		t.Fatalf("CronTab was not synced to service cluster: %v", err)
+	}
+
+	t.Log("Creating credential Secret on service cluster…")
+	if err := envtestClient.Create(ctx, credentialSecret("cred-1", "synced-default")); err != nil {
+		t.Fatalf("Failed to create Secret: %v", err)
+	}
+
+	t.Log("Waiting for Secret to be synced to kcp…")
+	waitForSecret(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"})
+
+	// Delete the source object, but immediately tear down the primary as well. We do NOT wait for
+	// the mid-life prune to run first, so that the copy may still be orphaned when teardown begins;
+	// the List-based teardown must reclaim it regardless.
+	t.Log("Deleting source Secret and then the CronTab…")
+	if err := envtestClient.Delete(ctx, credentialSecret("cred-1", "synced-default")); err != nil {
+		t.Fatalf("Failed to delete source Secret: %v", err)
+	}
+	if err := teamClient.Delete(ctx, crontab); err != nil {
+		t.Fatalf("Failed to delete CronTab: %v", err)
+	}
+
+	t.Log("Verifying that no orphaned copy survives in kcp…")
+	waitForSecretGone(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"})
+}
+
+// TestRelatedResourceMatchOriginKcpOrigin verifies the selector-shrink case for origin: kcp: when a
+// source object is relabelled so it no longer matches the selector (but is not deleted), its
+// service-side copy is pruned. This is a case the finalizer path alone misses.
+func TestRelatedResourceMatchOriginKcpOrigin(t *testing.T) {
+	const (
+		apiExportName = "kcp.example.com"
+		orgWorkspace  = "match-origin-kcp"
+	)
+
+	ctx := t.Context()
+	ctrlruntime.SetLogger(logr.Discard())
+
+	orgKubconfig := utils.CreateOrganization(t, ctx, orgWorkspace, apiExportName)
+
+	envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{
+		"test/crds/crontab.yaml",
+	})
+
+	watchLabels := map[string]string{"syncagent-e2e": "find-me"}
+
+	t.Log("Publishing CRDs…")
+	if err := envtestClient.Create(ctx, matchOriginPublishedResource(syncagentv1alpha1.RelatedResourceOriginKcp, watchLabels)); err != nil {
+		t.Fatalf("Failed to create PublishedResource: %v", err)
+	}
+
+	utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "")
+
+	kcpClusterClient := utils.GetKcpAdminClusterClient(t)
+	teamClusterPath := logicalcluster.NewPath("root").Join(orgWorkspace).Join("team-1")
+	teamClient := kcpClusterClient.Cluster(teamClusterPath)
+
+	utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{
+		Group:   "kcp.example.com",
+		Version: "v1",
+		Kind:    "CronTab",
+	})
+
+	t.Log("Creating CronTab in kcp…")
+	crontab := utils.YAMLToUnstructured(t, `
+apiVersion: kcp.example.com/v1
+kind: CronTab
+metadata:
+  namespace: default
+  name: my-crontab
+spec:
+  cronSpec: '* * *'
+  image: ubuntu:latest
+`)
+	crontab.SetLabels(watchLabels)
+
+	if err := teamClient.Create(ctx, crontab); err != nil {
+		t.Fatalf("Failed to create CronTab in kcp: %v", err)
+	}
+
+	// wait for the CronTab to be synced down so the synced-default namespace exists
+	t.Log("Waiting for CronTab to be synced to service cluster…")
+	serviceCrontab := &unstructured.Unstructured{}
+	serviceCrontab.SetAPIVersion("example.com/v1")
+	serviceCrontab.SetKind("CronTab")
+	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) {
+		return envtestClient.Get(ctx, types.NamespacedName{Namespace: "synced-default", Name: "my-crontab"}, serviceCrontab) == nil, nil
+	})
+	if err != nil {
+		t.Fatalf("CronTab was not synced to service cluster: %v", err)
+	}
+
+	// create two related Secrets in kcp (origin: kcp)
+	t.Log("Creating credential Secrets in kcp…")
+	for _, name := range []string{"cred-1", "cred-2"} {
+		if err := teamClient.Create(ctx, credentialSecret(name, "default")); err != nil {
+			t.Fatalf("Failed to create Secret %s: %v", name, err)
+		}
+	}
+
+	// both should be synced down to the service cluster
+	t.Log("Waiting for both Secrets to be synced to the service cluster…")
+	waitForSecret(t, ctx, envtestClient, types.NamespacedName{Name: "cred-1", Namespace: "synced-default"})
+	waitForSecret(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"})
+
+	// relabel cred-1 in kcp so it no longer matches the selector (but is NOT deleted)
+	t.Log("Relabelling cred-1 in kcp so it no longer matches…")
+	kcpSecret := &corev1.Secret{}
+	if err := teamClient.Get(ctx, types.NamespacedName{Name: "cred-1", Namespace: "default"}, kcpSecret); err != nil {
+		t.Fatalf("Failed to get cred-1 in kcp: %v", err)
+	}
+	kcpSecret.Labels = map[string]string{"app": "something-else"}
+	if err := teamClient.Update(ctx, kcpSecret); err != nil {
+		t.Fatalf("Failed to relabel cred-1 in kcp: %v", err)
+	}
+
+	// its service-side copy should be pruned…
+	t.Log("Verifying that the cred-1 copy on the service cluster is pruned…")
+	waitForSecretGone(t, ctx, envtestClient, types.NamespacedName{Name: "cred-1", Namespace: "synced-default"})
+
+	// …while the still-matching copy remains and the relabelled source object is untouched.
+	requireSecretExists(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"}, "cred-2 copy on service cluster should remain")
+	requireSecretExists(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"}, "relabelled cred-1 source in kcp should remain")
+}