Harden PBS backup attribution against shared sources (#1639)

Adversarial review of 84dba861b found three ways the new evidence paths
could still attribute a snapshot to the wrong cluster.

The submission-source learner was asymmetric. Clusters only became known
to it through snapshots that were already attributable, so a cluster with
no uniquely-attributable snapshot was invisible - and a source token both
clusters share then mapped to exactly one visible cluster and looked
decisive. The visible cluster got the other's backups while the other
guest stayed at zero. Callers now declare every connection owning a
candidate guest for a PBS instance, and the learner refuses to resolve
anything for that instance until each of them has had a snapshot
attributed to it. Observation is not scoped per PBS instance, so a
cluster seen submitting to its own PBS server still counts as visible -
the reported two-server topology keeps working.

PVE storage confirmations were treated as authorship. A pbs-type storage
listing proves the connection can SEE a snapshot, which a shared token, a
synced datastore, or an offsite copy all arrange without the connection
having made it, and a single confirmer previously outscored everything
else. Confirmations now carry the storage they came from, and only a
storage view that never lists a snapshot some other connection also lists
can attribute a colliding VMID. An overlapping view has demonstrated it
sees other clusters' snapshots, so nothing it lists attributes anything.
Where an exclusive view and the learned source mapping both speak they
must agree, otherwise the snapshot drops as it did before #1639. The
disjoint case - each cluster mounting only its own datastore - is
unchanged.

Confirmations were evicted by partial poll failures. A storage whose
content query failed contributed nothing, and the partial set overwrote
the previous one, flipping attribution between cycles. They now go
through the same per-storage preservation as storage backups.

Contract text calling the PVE listing "the only deterministic
attribution" is reworded to match the weakened semantics.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/release-control/v6/internal/subsystems/agent-lifecycle.md

@@ -414,7 +414,11 @@ update, profile rollout, command reachability, or fleet-control authority.
     is monitoring-owned attribution state for Proxmox backup freshness: it is
     internal-only, never serialized into state payloads or snapshots, and does
     not describe agent enrollment, report admission, tombstones,
-    re-enrollment, or command authority.
+    re-enrollment, or command authority. The evidence is keyed by the PVE
+    storage it was listed from and is readable per connection so monitoring
+    can carry it forward across a failed storage query; that accessor is
+    likewise monitoring-internal and never becomes an agent- or API-facing
+    surface.
 27. `internal/monitoring/monitor.go` shared with `monitoring`: monitor construction owns both monitoring runtime initialization and fail-closed agent lifecycle journal hydration before report admission.
 28. `internal/monitoring/monitor_agents.go` shared with `monitoring`: server-side Unified Agent report, removal, token binding, tombstone expiry, and re-enrollment semantics are jointly owned by agent lifecycle authority and monitoring ingest.
 29. `pkg/agents/host/report.go` shared with `monitoring`: the Unified Agent host report is both an agent lifecycle authored-state contract and a monitoring ingest contract for host maintenance posture.

docs/release-control/v6/internal/subsystems/alerts.md

@@ -38,7 +38,12 @@ may match a candidate guest's instance or node only by exact normalized
 equality. Suffix matching is reserved for real PBS namespaces inside the
 shared identity helpers; applying it to a PBS connection name
 cross-attributes clusters that share a VMID, so an unlinked backup whose
-label identifies no guest exactly stays on its generic rollup key.
+label identifies no guest exactly stays on its generic rollup key. The PBS
+submission-source learner that shares those identity helpers is backup
+attribution state, not alert identity: it is built per evaluation from
+positively attributed snapshots, never persists across evaluations, and is
+inconclusive by default, so it can only narrow which guest an already-linked
+backup belongs to and can never widen, merge, or move an alert's subject.
 Availability incident and alert identity belongs to the source-owned
 `network-endpoint` check. Correlation may project probe evidence onto a matched
 machine, but it must not copy the check incident onto that machine or create a

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -86,18 +86,29 @@ PBS snapshot-to-guest attribution for VMIDs that exist on more than one PVE
 location is evidence-driven, never guessed. When the direct PBS connection is
 authoritative and pbs-type storage contents are dropped from the PVE backup
 list, storage backup polling must still harvest each listed snapshot as a
-per-connection guest confirmation (type, VMID, backup time): which cluster
-listed a snapshot through its own storage is the only deterministic
-attribution for collision VMIDs with no namespace or comment evidence. Guest
-backup-time sync consumes those confirmations first, then a
-submission-source mapping (owner token, datastore, PBS instance — scoped to
-the PBS instance, strongest first) learned from the same poll's attributable
-snapshots; a source component that was never positively attributed stops
-resolution rather than deferring to weaker components. Snapshots that remain
-unattributable stay dropped for colliding guests, and the confirmation
-evidence is monitoring-internal state: cleared when a PVE instance is
-retired or its storage poll returns no pbs-type content, and never
-serialized into state payloads or snapshots.
+per-connection guest confirmation carrying the storage it was listed from
+(storage, type, VMID, backup time). A storage listing is evidence that the
+connection can see a snapshot, not that it authored it: a shared owner token,
+a synced datastore, or an offsite copy all surface another cluster's
+snapshots. A confirmation may therefore attribute a collision VMID only from
+a storage view that never lists a snapshot another connection also lists; a
+view that overlaps another connection's has demonstrated it sees snapshots it
+did not author, so nothing it lists attributes anything and it can never
+override other evidence. Guest backup-time sync weighs those confirmations
+alongside a submission-source mapping (owner token, datastore, PBS instance —
+scoped to the PBS instance, strongest first) learned from the same poll's
+attributable snapshots, and where both speak they must agree. A source
+component that was never positively attributed stops
+resolution rather than deferring to weaker components, and the source mapping
+stays inconclusive for a PBS instance while any PVE connection owning a
+candidate guest there has had no snapshot attributed to it at all — an
+unobserved connection may be submitting through the very same source.
+Snapshots that remain unattributable stay dropped for colliding guests. The
+confirmation evidence is monitoring-internal state: cleared when a PVE
+instance is retired or its storage poll returns no pbs-type content, carried
+forward per storage when that storage's content query fails so a partial poll
+failure cannot evict attribution, and never serialized into state payloads or
+snapshots.
 Removed host-agent reconnect blocks are identity-scoped: matching may use the
 canonical host ID or token-qualified machine/hostname continuity, but must never
 block a distinct live host by hostname alone.

docs/release-control/v6/internal/subsystems/storage-recovery.md

@@ -57,7 +57,12 @@ learned from the same batch's attributable snapshots. A source component
 never positively attributed stops resolution instead of deferring to weaker
 components, and a snapshot whose source decisively belongs to a connection
 with no matching guest stays unlinked; it must never be linked to another
-cluster's guest with the same VMID.
+cluster's guest with the same VMID. The learned mapping is positive evidence
+only, so a source resolving to a single connection proves nothing while some
+other connection owning a candidate guest has had no snapshot attributed to
+it: that connection is invisible to the mapping rather than absent from the
+source, and resolution stays inconclusive for the PBS instance until every
+candidate connection has been observed.
 
 
 Physical-disk to node association is evidence-directional. A disk reported by a

internal/models/issue1639_pbs_collision_test.go

@@ -147,9 +147,11 @@ func TestIssue1639PBSCollisionSharedSourceStaysDropped(t *testing.T) {
 }
 
 // TestIssue1639PBSCollisionResolvedByPVEStorageConfirmation verifies the
-// deterministic path: even with identical submission sources (fully
-// mirrored clusters pushing to one datastore with one token), each PVE
-// connection's own pbs-type storage listing attributes its snapshots.
+// storage-view path: even with identical submission sources (mirrored
+// clusters pushing with one token), each PVE connection's own pbs-type
+// storage listing attributes its snapshots — as long as the two listings are
+// disjoint, i.e. each cluster mounts only its own datastore, so seeing a
+// snapshot there really does mean authoring it.
 func TestIssue1639PBSCollisionResolvedByPVEStorageConfirmation(t *testing.T) {
 	state := NewState()
 
@@ -170,10 +172,10 @@ func TestIssue1639PBSCollisionResolvedByPVEStorageConfirmation(t *testing.T) {
 	state.mu.Unlock()
 
 	state.UpdatePBSGuestConfirmationsForInstance("cluster-a", []PBSGuestConfirmation{
-		{BackupType: "vm", VMID: 173, Time: backupTimeA.Unix()},
+		{Storage: "pbs-a", BackupType: "vm", VMID: 173, Time: backupTimeA.Unix()},
 	})
 	state.UpdatePBSGuestConfirmationsForInstance("cluster-b", []PBSGuestConfirmation{
-		{BackupType: "vm", VMID: 173, Time: backupTimeB.Unix()},
+		{Storage: "pbs-b", BackupType: "vm", VMID: 173, Time: backupTimeB.Unix()},
 	})
 
 	state.SyncGuestBackupTimes()
@@ -206,13 +208,16 @@ func TestIssue1639PBSCollisionForeignSourceRejected(t *testing.T) {
 		{VMID: 173, Name: "web-a", Instance: "cluster-a", Node: "pve-a1"},
 		{VMID: 173, Name: "web-b", Instance: "cluster-b", Node: "pve-b1"},
 		{VMID: 100, Name: "db-a", Instance: "cluster-a", Node: "pve-a1"},
+		{VMID: 150, Name: "db-b", Instance: "cluster-b", Node: "pve-b1"},
 	})
 
 	state.mu.Lock()
 	state.PBSBackups = []PBSBackup{
-		// Only cluster-a's source is learnable, and the only 173 snapshot
-		// belongs to it. cluster-b must stay at zero.
+		// Both clusters are visible with a source of their own, so the
+		// singleton owner token really is cluster-a's. The only 173 snapshot
+		// belongs to cluster-a and cluster-b must stay at zero.
 		{ID: "a-100", VMID: "100", BackupType: "vm", BackupTime: now.Add(-26 * time.Hour), Instance: "pbs-main", Datastore: "store-a", Owner: "cluster-a@pbs!token"},
+		{ID: "b-150", VMID: "150", BackupType: "vm", BackupTime: now.Add(-27 * time.Hour), Instance: "pbs-main", Datastore: "store-b", Owner: "cluster-b@pbs!token"},
 		{ID: "a-173", VMID: "173", BackupType: "vm", BackupTime: now.Add(-2 * time.Hour), Instance: "pbs-main", Datastore: "store-a", Owner: "cluster-a@pbs!token"},
 	}
 	state.mu.Unlock()
@@ -236,3 +241,151 @@ func TestIssue1639PBSCollisionForeignSourceRejected(t *testing.T) {
 		}
 	}
 }
+
+// TestIssue1639PBSCollisionInvisibleClusterKeepsSourceInconclusive is the
+// asymmetric case the first #1639 fix got wrong. Clusters only become known
+// to the submission-source learner through snapshots that were already
+// attributable, so cluster-b — sharing cluster-a's owner token and owning no
+// uniquely-attributable guest — is invisible. The shared token then looks
+// like cluster-a's alone and cluster-a is handed cluster-b's snapshot.
+// Without further evidence both guests must stay unattributed.
+func TestIssue1639PBSCollisionInvisibleClusterKeepsSourceInconclusive(t *testing.T) {
+	state := NewState()
+
+	now := time.Now()
+	backupTimeB := now.Add(-2 * time.Hour).Truncate(time.Second)
+
+	state.UpdateVMs([]VM{
+		{VMID: 173, Name: "web-a", Instance: "cluster-a", Node: "pve-a1"},
+		{VMID: 173, Name: "web-b", Instance: "cluster-b", Node: "pve-b1"},
+		// Only cluster-a owns a guest whose snapshots are attributable on
+		// their own; cluster-b's single guest is the colliding one.
+		{VMID: 100, Name: "db-a", Instance: "cluster-a", Node: "pve-a1"},
+	})
+
+	state.mu.Lock()
+	state.PBSBackups = []PBSBackup{
+		{ID: "a-100", VMID: "100", BackupType: "vm", BackupTime: now.Add(-26 * time.Hour),
+			Instance: "pbs-main", Datastore: "backups", Owner: "shared@pbs!token"},
+		// Authored by cluster-b, indistinguishable from cluster-a's own
+		// submissions because the token and datastore are shared.
+		{ID: "b-173", VMID: "173", BackupType: "vm", BackupTime: backupTimeB,
+			Instance: "pbs-main", Datastore: "backups", Owner: "shared@pbs!token"},
+	}
+	state.mu.Unlock()
+
+	state.SyncGuestBackupTimes()
+
+	for _, vm := range state.GetSnapshot().VMs {
+		if vm.VMID != 173 {
+			continue
+		}
+		if !vm.LastBackup.IsZero() {
+			t.Errorf("VM 173 on %s = %v, want zero while cluster-b has no attributable evidence", vm.Instance, vm.LastBackup)
+		}
+	}
+}
+
+// TestIssue1639PBSCollisionInvisibleClusterResolvedByOwnStorageView is the
+// same topology as above with the one piece of evidence that settles it:
+// cluster-b's own pbs-type storage lists the snapshot, and no other
+// connection's listing overlaps that storage. The snapshot goes to
+// cluster-b, and cluster-a still does not get it.
+func TestIssue1639PBSCollisionInvisibleClusterResolvedByOwnStorageView(t *testing.T) {
+	state := NewState()
+
+	now := time.Now()
+	uniqueTimeA := now.Add(-26 * time.Hour).Truncate(time.Second)
+	backupTimeB := now.Add(-2 * time.Hour).Truncate(time.Second)
+
+	state.UpdateVMs([]VM{
+		{VMID: 173, Name: "web-a", Instance: "cluster-a", Node: "pve-a1"},
+		{VMID: 173, Name: "web-b", Instance: "cluster-b", Node: "pve-b1"},
+		{VMID: 100, Name: "db-a", Instance: "cluster-a", Node: "pve-a1"},
+	})
+
+	state.mu.Lock()
+	state.PBSBackups = []PBSBackup{
+		{ID: "a-100", VMID: "100", BackupType: "vm", BackupTime: uniqueTimeA,
+			Instance: "pbs-main", Datastore: "backups", Owner: "shared@pbs!token"},
+		{ID: "b-173", VMID: "173", BackupType: "vm", BackupTime: backupTimeB,
+			Instance: "pbs-main", Datastore: "backups", Owner: "shared@pbs!token"},
+	}
+	state.mu.Unlock()
+
+	state.UpdatePBSGuestConfirmationsForInstance("cluster-a", []PBSGuestConfirmation{
+		{Storage: "pbs-a", BackupType: "vm", VMID: 100, Time: uniqueTimeA.Unix()},
+	})
+	state.UpdatePBSGuestConfirmationsForInstance("cluster-b", []PBSGuestConfirmation{
+		{Storage: "pbs-b", BackupType: "vm", VMID: 173, Time: backupTimeB.Unix()},
+	})
+
+	state.SyncGuestBackupTimes()
+
+	for _, vm := range state.GetSnapshot().VMs {
+		if vm.VMID != 173 {
+			continue
+		}
+		switch vm.Instance {
+		case "cluster-a":
+			if !vm.LastBackup.IsZero() {
+				t.Errorf("cluster-a VM 173 = %v, want zero (the snapshot is cluster-b's)", vm.LastBackup)
+			}
+		case "cluster-b":
+			if !vm.LastBackup.Equal(backupTimeB) {
+				t.Errorf("cluster-b VM 173 = %v, want its own confirmed snapshot %v", vm.LastBackup, backupTimeB)
+			}
+		}
+	}
+}
+
+// TestIssue1639PBSStorageConfirmationFromSharedViewIsNotAuthorship covers the
+// second half of the confirmation weakening: a connection whose storage view
+// also lists snapshots another connection lists is looking at a shared,
+// synced, or offsite datastore. Seeing a snapshot there proves visibility,
+// not authorship, so it must not attribute a colliding VMID on its own.
+func TestIssue1639PBSStorageConfirmationFromSharedViewIsNotAuthorship(t *testing.T) {
+	state := NewState()
+
+	now := time.Now()
+	sharedTime := now.Add(-25 * time.Hour).Truncate(time.Second)
+	disputedTime := now.Add(-2 * time.Hour).Truncate(time.Second)
+
+	state.UpdateVMs([]VM{
+		{VMID: 173, Name: "web-a", Instance: "cluster-a", Node: "pve-a1"},
+		{VMID: 173, Name: "web-b", Instance: "cluster-b", Node: "pve-b1"},
+		{VMID: 200, Name: "shared-guest", Instance: "cluster-b", Node: "pve-b1"},
+	})
+
+	state.mu.Lock()
+	state.PBSBackups = []PBSBackup{
+		{ID: "b-200", VMID: "200", BackupType: "vm", BackupTime: sharedTime,
+			Instance: "pbs-main", Datastore: "backups", Owner: "shared@pbs!token"},
+		{ID: "x-173", VMID: "173", BackupType: "vm", BackupTime: disputedTime,
+			Instance: "pbs-main", Datastore: "backups", Owner: "shared@pbs!token"},
+	}
+	state.mu.Unlock()
+
+	// Both connections mount the same datastore: their listings overlap on
+	// VM 200, which is cluster-b's guest. cluster-a additionally lists the
+	// disputed 173 snapshot, but its view has already proven it can see
+	// other clusters' snapshots.
+	state.UpdatePBSGuestConfirmationsForInstance("cluster-a", []PBSGuestConfirmation{
+		{Storage: "pbs-shared", BackupType: "vm", VMID: 200, Time: sharedTime.Unix()},
+		{Storage: "pbs-shared", BackupType: "vm", VMID: 173, Time: disputedTime.Unix()},
+	})
+	state.UpdatePBSGuestConfirmationsForInstance("cluster-b", []PBSGuestConfirmation{
+		{Storage: "pbs-shared", BackupType: "vm", VMID: 200, Time: sharedTime.Unix()},
+	})
+
+	state.SyncGuestBackupTimes()
+
+	for _, vm := range state.GetSnapshot().VMs {
+		if vm.VMID != 173 {
+			continue
+		}
+		if !vm.LastBackup.IsZero() {
+			t.Errorf("VM 173 on %s = %v, want zero: a shared storage view is visibility, not authorship", vm.Instance, vm.LastBackup)
+		}
+	}
+}

internal/models/metrics_types_test.go

@@ -218,8 +218,11 @@ func TestNativePoolHealthEvidenceNormalizesWithoutChangingIdentity(t *testing.T)
 func TestPBSGuestConfirmationEvidenceStaysOutOfSerializedState(t *testing.T) {
 	state := NewState()
 	state.UpdatePBSGuestConfirmationsForInstance("cluster-a", []PBSGuestConfirmation{
-		{BackupType: "vm", VMID: 173, Time: time.Now().Unix()},
+		{Storage: "pbs-store", BackupType: "vm", VMID: 173, Time: time.Now().Unix()},
 	})
+	if len(state.PBSGuestConfirmationsForInstance("cluster-a")) != 1 {
+		t.Fatal("precondition: confirmation should be retained as internal evidence")
+	}
 
 	payload, err := json.Marshal(state)
 	if err != nil {

internal/models/models.go

@@ -49,22 +49,27 @@ type State struct {
 	PVETagColors                 map[string]string          `json:"pveTagColors,omitempty"`
 	PVETagStyles                 map[string]PVETagStyle     `json:"pveTagStyles,omitempty"`
 
-	// pbsGuestConfirmations records, per PVE connection, the (type, vmid,
-	// time) triples that connection's own pbs-type storage listings
+	// pbsGuestConfirmations records, per PVE connection, the (storage, type,
+	// vmid, time) tuples that connection's own pbs-type storage listings
 	// reported. When direct PBS polling is authoritative the raw storage
-	// contents are not kept as StorageBackups, but which cluster listed a
-	// snapshot is the only deterministic way to attribute a PBS snapshot
-	// whose VMID exists on several clusters (#1639). Internal evidence only:
-	// never serialized and never part of snapshots.
+	// contents are not kept as StorageBackups, but which storage view listed
+	// a snapshot is corroborating evidence for attributing a PBS snapshot
+	// whose VMID exists on several clusters (#1639). A listing proves the
+	// connection can see the snapshot, not that it authored it, so the
+	// storage name is retained and views that overlap another connection's
+	// are discarded as decisive evidence. Internal evidence only: never
+	// serialized and never part of snapshots.
 	pbsGuestConfirmations map[string]map[PBSGuestConfirmation]struct{}
 }
 
 // PBSGuestConfirmation identifies one PBS snapshot as seen through a PVE
-// connection's own pbs-type storage content listing. BackupType uses PBS
-// nomenclature ("vm" or "ct") and Time is the snapshot's backup time in
-// unix seconds, which matches PBSBackup.BackupTime exactly because both
-// derive from the snapshot identifier.
+// connection's own pbs-type storage content listing. Storage is the PVE
+// storage the listing came from, BackupType uses PBS nomenclature ("vm" or
+// "ct") and Time is the snapshot's backup time in unix seconds, which
+// matches PBSBackup.BackupTime exactly because both derive from the snapshot
+// identifier.
 type PBSGuestConfirmation struct {
+	Storage    string
 	BackupType string
 	VMID       int
 	Time       int64
@@ -4145,6 +4150,12 @@ func (s *State) UpdatePBSGuestConfirmationsForInstance(instanceName string, conf
 	set := make(map[PBSGuestConfirmation]struct{}, len(confirmations))
 	for _, confirmation := range confirmations {
 		confirmation.BackupType = strings.ToLower(strings.TrimSpace(confirmation.BackupType))
+		confirmation.Storage = strings.TrimSpace(confirmation.Storage)
+		if confirmation.Storage == "" {
+			// Without the originating storage the evidence cannot be tested
+			// for exclusivity or preserved across a partial poll failure.
+			continue
+		}
 		if (confirmation.BackupType != "vm" && confirmation.BackupType != "ct") || confirmation.VMID <= 0 || confirmation.Time <= 0 {
 			continue
 		}
@@ -4157,6 +4168,38 @@ func (s *State) UpdatePBSGuestConfirmationsForInstance(instanceName string, conf
 	s.pbsGuestConfirmations[instanceName] = set
 }
 
+// PBSGuestConfirmationsForInstance returns the PVE-side PBS snapshot evidence
+// currently held for one connection. Callers use it to carry forward the
+// evidence of storages whose content query failed this cycle, so a partial
+// poll failure cannot evict attribution and flip guests between cycles
+// (#1639).
+func (s *State) PBSGuestConfirmationsForInstance(instanceName string) []PBSGuestConfirmation {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	set, ok := s.pbsGuestConfirmations[instanceName]
+	if !ok || len(set) == 0 {
+		return nil
+	}
+	out := make([]PBSGuestConfirmation, 0, len(set))
+	for confirmation := range set {
+		out = append(out, confirmation)
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Storage != out[j].Storage {
+			return out[i].Storage < out[j].Storage
+		}
+		if out[i].BackupType != out[j].BackupType {
+			return out[i].BackupType < out[j].BackupType
+		}
+		if out[i].VMID != out[j].VMID {
+			return out[i].VMID < out[j].VMID
+		}
+		return out[i].Time < out[j].Time
+	})
+	return out
+}
+
 // SyncGuestBackupTimes updates LastBackup on VMs and Containers from storage backups and PBS backups.
 // Call this after updating storage backups or PBS backups to ensure guest backup indicators are accurate.
 // Matching is done by instance+VMID to prevent cross-instance VMID collisions.
@@ -4239,20 +4282,56 @@ func (s *State) SyncGuestBackupTimes() {
 		}
 	}
 
-	// PVE-side confirmations deterministically attribute a PBS snapshot to
-	// the connection whose own pbs-type storage listed it. Index them by
-	// snapshot identity, keeping the set of confirming connections so a
-	// snapshot listed by several connections (shouldn't happen, but trust
-	// nothing) stays inconclusive.
-	confirmingInstances := make(map[PBSGuestConfirmation]map[string]struct{})
+	// PVE-side confirmations say a connection's own pbs-type storage listed a
+	// snapshot. That is visibility, not authorship: a shared token, a synced
+	// datastore, or an offsite copy all make another cluster's snapshots
+	// appear in this cluster's listing. Only a storage view that never lists
+	// a snapshot some other connection also lists can be treated as
+	// authorship evidence; an overlapping view has demonstrated it sees other
+	// clusters' snapshots, so nothing it lists attributes anything (#1639).
+	type pbsSnapshotKey struct {
+		backupType string
+		vmid       int
+		unixTime   int64
+	}
+	type pbsStorageView struct {
+		instance string
+		storage  string
+	}
+	viewsBySnapshot := make(map[pbsSnapshotKey][]pbsStorageView)
 	for instanceName, confirmations := range s.pbsGuestConfirmations {
 		for confirmation := range confirmations {
-			set, ok := confirmingInstances[confirmation]
-			if !ok {
-				set = make(map[string]struct{})
-				confirmingInstances[confirmation] = set
+			key := pbsSnapshotKey{backupType: confirmation.BackupType, vmid: confirmation.VMID, unixTime: confirmation.Time}
+			viewsBySnapshot[key] = append(viewsBySnapshot[key], pbsStorageView{instance: instanceName, storage: confirmation.Storage})
+		}
+	}
+	sharedViews := make(map[pbsStorageView]struct{})
+	for _, views := range viewsBySnapshot {
+		for i := range views {
+			for j := i + 1; j < len(views); j++ {
+				if views[i].instance == views[j].instance {
+					continue
+				}
+				sharedViews[views[i]] = struct{}{}
+				sharedViews[views[j]] = struct{}{}
+			}
+		}
+	}
+	// exclusiveConfirmer names the connection whose non-overlapping storage
+	// view listed a snapshot. A snapshot two connections both list makes both
+	// their views overlapping, so it has no exclusive confirmer at all.
+	exclusiveConfirmer := make(map[pbsSnapshotKey]string)
+	for key, views := range viewsBySnapshot {
+		exclusive := make(map[string]struct{}, len(views))
+		for _, view := range views {
+			if _, shared := sharedViews[view]; !shared {
+				exclusive[view.instance] = struct{}{}
+			}
+		}
+		if len(exclusive) == 1 {
+			for instanceName := range exclusive {
+				exclusiveConfirmer[key] = instanceName
 			}
-			set[instanceName] = struct{}{}
 		}
 	}
 
@@ -4262,6 +4341,17 @@ func (s *State) SyncGuestBackupTimes() {
 	// learned mapping then resolves collision-VMID snapshots that carry no
 	// namespace or comment evidence of their own (#1639).
 	sourceLearner := proxmoxidentity.NewPBSSourceLearner()
+	for key, backups := range pbsBackupsBySubject {
+		guests := subjectGuests[key]
+		if len(guests) == 0 {
+			continue
+		}
+		for _, backup := range backups {
+			for _, guest := range guests {
+				sourceLearner.RegisterCandidate(backup.Instance, guest.instance)
+			}
+		}
+	}
 	for key, backups := range pbsBackupsBySubject {
 		guests := subjectGuests[key]
 		if len(guests) == 0 {
@@ -4283,6 +4373,12 @@ func (s *State) SyncGuestBackupTimes() {
 						attributed = instance
 					}
 				}
+				if attributed == "" {
+					// An exclusive storage view is positive attribution too,
+					// and it makes the cluster behind it visible to the
+					// coverage check.
+					attributed = exclusiveConfirmer[pbsSnapshotKey{backupType: key.backupType, vmid: key.vmid, unixTime: backup.BackupTime.Unix()}]
+				}
 			}
 			if attributed != "" {
 				sourceLearner.Observe(backup.Instance, backup.Datastore, backup.Owner, attributed)
@@ -4312,24 +4408,33 @@ func (s *State) SyncGuestBackupTimes() {
 				instance,
 				node,
 			)
+			snapshotKey := pbsSnapshotKey{backupType: backupType, vmid: vmid, unixTime: backup.BackupTime.Unix()}
 			if score == 0 {
 				if !subjectIsAmbiguous[subjectKey] {
 					score = 1
-				} else if confirmers := confirmingInstances[PBSGuestConfirmation{BackupType: backupType, VMID: vmid, Time: backup.BackupTime.Unix()}]; len(confirmers) == 1 {
-					// Exactly one connection's own PBS storage listed this
-					// snapshot — it belongs to that cluster, full stop.
-					if _, ours := confirmers[instance]; ours {
-						score = 2
-					}
 				} else {
-					// No usable storage-view evidence (unconfirmed, or a
-					// shared datastore mount several clusters list); fall
-					// back to the learned submission-source mapping. A
-					// decisive attribution to another cluster keeps the
-					// snapshot away from this guest; an inconclusive one
-					// keeps the pre-#1639 drop.
-					if attributed, decisive := sourceLearner.Resolve(backup.Instance, backup.Datastore, backup.Owner); decisive && attributed == instance {
-						score = 1
+					// Two independent evidence paths, neither authoritative
+					// over the other. An exclusive storage view (one no other
+					// connection's listing overlaps) is authorship evidence;
+					// the learned submission-source mapping is inference from
+					// the batch's attributable snapshots. Where both speak
+					// they must agree, otherwise the snapshot stays dropped
+					// as it did before #1639.
+					confirmer, confirmed := exclusiveConfirmer[snapshotKey]
+					attributed, learned := sourceLearner.Resolve(backup.Instance, backup.Datastore, backup.Owner)
+					switch {
+					case confirmed && learned:
+						if confirmer == instance && attributed == instance {
+							score = 1
+						}
+					case confirmed:
+						if confirmer == instance {
+							score = 1
+						}
+					case learned:
+						if attributed == instance {
+							score = 1
+						}
 					}
 				}
 			}

internal/monitoring/monitor_backups.go

@@ -308,13 +308,17 @@ func (m *Monitor) pollStorageBackupsWithNodes(ctx context.Context, instanceName
 				isPBSStorage := storage.Type == "pbs"
 				if isPBSStorage && hasPBSDirectConnection {
 					// The direct PBS connection is authoritative for the
-					// backup list, but which cluster listed a snapshot is
-					// evidence only this storage view has: it attributes a
-					// snapshot whose VMID exists on several clusters to the
-					// cluster that made it (#1639). Keep that association
-					// even though the raw entry is dropped.
+					// backup list, but which storage view listed a snapshot
+					// is evidence only this connection has: it helps
+					// attribute a snapshot whose VMID exists on several
+					// clusters (#1639). Keep that association, tagged with
+					// the storage it came from, even though the raw entry is
+					// dropped - a listing only proves this connection can see
+					// the snapshot, so consumers need the storage to tell an
+					// exclusive view from a shared or synced datastore.
 					if confirmationType := pbsGuestConfirmationType(backupType); confirmationType != "" && content.VMID > 0 && content.CTime > 0 {
 						pbsConfirmations = append(pbsConfirmations, models.PBSGuestConfirmation{
+							Storage:    storage.Storage,
 							BackupType: confirmationType,
 							VMID:       content.VMID,
 							Time:       content.CTime,
@@ -378,6 +382,15 @@ func (m *Monitor) pollStorageBackupsWithNodes(ctx context.Context, instanceName
 			Msg("Preserving previous storage backup data due to partial failures")
 	}
 
+	pbsConfirmations, preservedConfirmationStorages := preserveFailedPBSGuestConfirmations(
+		m.state.PBSGuestConfirmationsForInstance(instanceName), storagePreserveNeeded, pbsConfirmations)
+	if len(preservedConfirmationStorages) > 0 {
+		log.Warn().
+			Str("instance", instanceName).
+			Strs("storages", preservedConfirmationStorages).
+			Msg("Preserving previous PBS guest confirmation evidence due to partial failures")
+	}
+
 	// Decide whether to keep existing backups when every query failed
 	if shouldPreserveBackups(len(nodes), hadSuccessfulNode, storagesWithBackup, contentSuccess) {
 		if len(nodes) > 0 && !hadSuccessfulNode {
@@ -558,6 +571,46 @@ func storageNamesForNode(readState unifiedresources.ReadState, instanceName, nod
 	return storages
 }
 
+// preserveFailedPBSGuestConfirmations carries forward the PBS guest
+// confirmation evidence of storages whose content query failed this cycle,
+// mirroring preserveFailedStorageBackups. Publishing the partial (possibly
+// empty) set instead would evict attribution evidence on a transient poll
+// failure and flip collision-VMID guests between cycles (#1639).
+func preserveFailedPBSGuestConfirmations(previous []models.PBSGuestConfirmation, storagesToPreserve map[string]struct{}, current []models.PBSGuestConfirmation) ([]models.PBSGuestConfirmation, []string) {
+	if len(storagesToPreserve) == 0 || len(previous) == 0 {
+		return current, nil
+	}
+
+	existing := make(map[models.PBSGuestConfirmation]struct{}, len(current))
+	for _, confirmation := range current {
+		existing[confirmation] = struct{}{}
+	}
+
+	preserved := make(map[string]struct{})
+	for _, confirmation := range previous {
+		if _, ok := storagesToPreserve[confirmation.Storage]; !ok {
+			continue
+		}
+		if _, duplicate := existing[confirmation]; duplicate {
+			continue
+		}
+		current = append(current, confirmation)
+		existing[confirmation] = struct{}{}
+		preserved[confirmation.Storage] = struct{}{}
+	}
+
+	if len(preserved) == 0 {
+		return current, nil
+	}
+
+	storages := make([]string, 0, len(preserved))
+	for storage := range preserved {
+		storages = append(storages, storage)
+	}
+	sort.Strings(storages)
+	return current, storages
+}
+
 func preserveFailedStorageBackups(instanceName string, snapshot models.StateSnapshot, storagesToPreserve map[string]struct{}, current []models.StorageBackup) ([]models.StorageBackup, []string) {
 	if len(storagesToPreserve) == 0 {
 		return current, nil

internal/monitoring/monitor_backups_dir_storage_test.go

@@ -2,6 +2,7 @@ package monitoring
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -69,6 +70,85 @@ func TestPollStorageBackups_KeepsDirStorageNamedLikePBS(t *testing.T) {
 	}
 }
 
+// pbsPartialFailureClient serves two pbs-type storages and can be told to
+// fail one storage's content query, standing in for a partial poll failure.
+type pbsPartialFailureClient struct {
+	mockPVEClientExtra
+	snapshotTime time.Time
+	failStorage  string
+}
+
+func (c *pbsPartialFailureClient) GetStorage(ctx context.Context, node string) ([]pveapi.Storage, error) {
+	return []pveapi.Storage{
+		{Storage: "pbs-one", Content: "backup", Type: "pbs", Enabled: 1, Active: 1},
+		{Storage: "pbs-two", Content: "backup", Type: "pbs", Enabled: 1, Active: 1},
+	}, nil
+}
+
+func (c *pbsPartialFailureClient) GetStorageContent(ctx context.Context, node, storage string) ([]pveapi.StorageContent, error) {
+	if storage == c.failStorage {
+		return nil, fmt.Errorf("500 internal error")
+	}
+	vmid := 173
+	if storage == "pbs-two" {
+		vmid = 174
+	}
+	return []pveapi.StorageContent{{
+		Volid:   fmt.Sprintf("%s:backup/vm/%d/2026-07-27T01:00:00Z", storage, vmid),
+		VMID:    vmid,
+		Size:    2048,
+		CTime:   c.snapshotTime.Unix(),
+		Content: "backup",
+		Format:  "pbs-vm",
+	}}, nil
+}
+
+// TestPollStorageBackups_Issue1639PreservesConfirmationsOnPartialFailure
+// verifies that a storage whose content query fails keeps its previous PBS
+// guest confirmations instead of having them evicted by the partial set.
+// Losing them flips collision-VMID attribution from cycle to cycle.
+func TestPollStorageBackups_Issue1639PreservesConfirmationsOnPartialFailure(t *testing.T) {
+	snapshotTime := time.Date(2026, 7, 27, 1, 0, 0, 0, time.UTC)
+
+	m := &Monitor{
+		state: models.NewState(),
+		config: &config.Config{
+			PBSInstances: []config.PBSInstance{{Name: "pbs-1", Host: "https://pbs.example:8007"}},
+		},
+	}
+	nodes := []pveapi.Node{{Node: "node-a", Status: "online"}}
+	online := map[string]string{"node-a": "online"}
+
+	// Healthy cycle: both storages list one snapshot each.
+	m.pollStorageBackupsWithNodes(context.Background(), "cluster-a",
+		&pbsPartialFailureClient{snapshotTime: snapshotTime}, nodes, online)
+
+	healthy := m.state.PBSGuestConfirmationsForInstance("cluster-a")
+	if len(healthy) != 2 {
+		t.Fatalf("expected both storages to contribute confirmations, got %+v", healthy)
+	}
+
+	// Degraded cycle: pbs-one's content query fails. Its evidence must
+	// survive, and pbs-two's must be refreshed as usual.
+	m.pollStorageBackupsWithNodes(context.Background(), "cluster-a",
+		&pbsPartialFailureClient{snapshotTime: snapshotTime, failStorage: "pbs-one"}, nodes, online)
+
+	degraded := m.state.PBSGuestConfirmationsForInstance("cluster-a")
+	if len(degraded) != 2 {
+		t.Fatalf("partial failure evicted confirmation evidence, got %+v", degraded)
+	}
+	byStorage := map[string]models.PBSGuestConfirmation{}
+	for _, confirmation := range degraded {
+		byStorage[confirmation.Storage] = confirmation
+	}
+	if got, ok := byStorage["pbs-one"]; !ok || got.VMID != 173 {
+		t.Fatalf("pbs-one confirmation not preserved across the failed query, got %+v", degraded)
+	}
+	if got, ok := byStorage["pbs-two"]; !ok || got.VMID != 174 {
+		t.Fatalf("pbs-two confirmation missing after successful query, got %+v", degraded)
+	}
+}
+
 // pbsCollisionStorageClient serves one pbs-type storage listing a single
 // snapshot for VMID 173, standing in for cluster-a's own PBS storage view.
 type pbsCollisionStorageClient struct {

internal/monitoring/monitor_backups_readstate_test.go

@@ -969,7 +969,7 @@ func TestRetirePVEInstanceRuntimeClearsPBSGuestConfirmations(t *testing.T) {
 		{ID: "a-173", VMID: "173", BackupType: "vm", BackupTime: backupTime, Instance: "pbs-1", Datastore: "backups"},
 	})
 	state.UpdatePBSGuestConfirmationsForInstance("cluster-a", []models.PBSGuestConfirmation{
-		{BackupType: "vm", VMID: 173, Time: backupTime.Unix()},
+		{Storage: "pbs-store", BackupType: "vm", VMID: 173, Time: backupTime.Unix()},
 	})
 
 	state.SyncGuestBackupTimes()

internal/proxmoxidentity/backup_identity.go

@@ -113,12 +113,53 @@ const (
 // with a per-cluster token and often a per-cluster datastore, so evidence
 // learned from unambiguous guests can attribute root-namespace,
 // comment-less snapshots whose VMID exists on more than one cluster (#1639).
+//
+// Evidence is only ever positive: a cluster becomes visible to the learner
+// by having a snapshot attributed to it. A source key mapping to exactly one
+// visible cluster therefore proves nothing on its own, because a cluster
+// with no attributable snapshot at all is indistinguishable from a cluster
+// that does not use that source. Callers must declare every cluster that
+// could have authored a backup on a PBS instance with RegisterCandidate, and
+// resolution stays inconclusive until each of them has been observed.
 type PBSSourceLearner struct {
 	instancesByKey map[string]map[string]struct{}
+	// candidatesByPBS lists, per PBS instance, the PVE connections that own
+	// a guest one of that instance's backups could belong to.
+	candidatesByPBS map[string]map[string]struct{}
+	// observed is every PVE connection that has had at least one snapshot
+	// positively attributed to it, from any PBS instance.
+	observed map[string]struct{}
 }
 
 func NewPBSSourceLearner() *PBSSourceLearner {
-	return &PBSSourceLearner{instancesByKey: make(map[string]map[string]struct{})}
+	return &PBSSourceLearner{
+		instancesByKey:  make(map[string]map[string]struct{}),
+		candidatesByPBS: make(map[string]map[string]struct{}),
+		observed:        make(map[string]struct{}),
+	}
+}
+
+// RegisterCandidate declares that pveInstance owns a guest that a backup on
+// pbsInstance could belong to. Resolve refuses to be decisive for that PBS
+// instance until every registered candidate has been observed, so a cluster
+// that contributes no attributable snapshot cannot be silently written out of
+// the candidate set and have its backups handed to a cluster that shares its
+// owner token or datastore (#1639).
+func (l *PBSSourceLearner) RegisterCandidate(pbsInstance, pveInstance string) {
+	if l == nil {
+		return
+	}
+	pveInstance = strings.TrimSpace(pveInstance)
+	if pveInstance == "" {
+		return
+	}
+	key := strings.TrimSpace(pbsInstance)
+	set, ok := l.candidatesByPBS[key]
+	if !ok {
+		set = make(map[string]struct{})
+		l.candidatesByPBS[key] = set
+	}
+	set[pveInstance] = struct{}{}
 }
 
 func pbsSourceKeys(pbsInstance, datastore, owner string) []string {
@@ -149,6 +190,7 @@ func (l *PBSSourceLearner) Observe(pbsInstance, datastore, owner, pveInstance st
 	if pveInstance == "" {
 		return
 	}
+	l.observed[pveInstance] = struct{}{}
 	for _, key := range pbsSourceKeys(pbsInstance, datastore, owner) {
 		set, ok := l.instancesByKey[key]
 		if !ok {
@@ -159,6 +201,31 @@ func (l *PBSSourceLearner) Observe(pbsInstance, datastore, owner, pveInstance st
 	}
 }
 
+// candidatesAllObserved reports whether every PVE connection that could have
+// authored a backup on this PBS instance has had at least one snapshot
+// positively attributed to it. Until that holds, a source key mapping to a
+// single connection is not evidence the source is exclusive to it — the
+// unobserved connection may be pushing through the very same owner token or
+// datastore, and its snapshots would be handed to the observed one (#1639).
+//
+// Observation is not scoped to the PBS instance on purpose: a cluster seen
+// submitting to a different PBS instance is visible evidence about where its
+// backups land, which is exactly what distinguishes clusters that each own a
+// PBS server. A cluster with no attributed snapshot anywhere is invisible and
+// blocks resolution.
+func (l *PBSSourceLearner) candidatesAllObserved(pbsInstance string) bool {
+	candidates := l.candidatesByPBS[strings.TrimSpace(pbsInstance)]
+	if len(candidates) == 0 {
+		return false
+	}
+	for candidate := range candidates {
+		if _, ok := l.observed[candidate]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
 // Resolve reports the single PVE connection the backup's source evidence
 // identifies. Evidence is consulted strongest-first; a source component that
 // maps to several connections is not a discriminator and defers to the next
@@ -166,10 +233,18 @@ func (l *PBSSourceLearner) Observe(pbsInstance, datastore, owner, pveInstance st
 // an unfamiliar owner or datastore means the backup may belong to a cluster
 // we have no evidence for, and guessing from weaker components would
 // attribute it to the wrong cluster.
+//
+// Resolution is inconclusive for the whole PBS instance while any registered
+// candidate cluster is still unobserved, because a singleton mapping then
+// only reflects which clusters happened to be attributable, not that the
+// source is exclusive to one of them.
 func (l *PBSSourceLearner) Resolve(pbsInstance, datastore, owner string) (string, bool) {
 	if l == nil {
 		return "", false
 	}
+	if !l.candidatesAllObserved(pbsInstance) {
+		return "", false
+	}
 	for _, key := range pbsSourceKeys(pbsInstance, datastore, owner) {
 		set, ok := l.instancesByKey[key]
 		if !ok {

internal/proxmoxidentity/backup_identity_test.go

@@ -86,6 +86,8 @@ func TestLocationLabelsEqualRequiresExactNormalizedMatch(t *testing.T) {
 
 func TestPBSSourceLearnerResolvesOwnerBeforeWeakerComponents(t *testing.T) {
 	l := NewPBSSourceLearner()
+	l.RegisterCandidate("pbs-main", "cluster-a")
+	l.RegisterCandidate("pbs-main", "cluster-b")
 	l.Observe("pbs-main", "backups", "cluster-a@pbs!token", "cluster-a")
 	l.Observe("pbs-main", "backups", "cluster-b@pbs!token", "cluster-b")
 
@@ -101,6 +103,8 @@ func TestPBSSourceLearnerResolvesOwnerBeforeWeakerComponents(t *testing.T) {
 
 func TestPBSSourceLearnerSharedSourceIsNotDecisive(t *testing.T) {
 	l := NewPBSSourceLearner()
+	l.RegisterCandidate("pbs-main", "cluster-a")
+	l.RegisterCandidate("pbs-main", "cluster-b")
 	l.Observe("pbs-main", "backups", "shared@pbs!token", "cluster-a")
 	l.Observe("pbs-main", "backups", "shared@pbs!token", "cluster-b")
 
@@ -111,7 +115,10 @@ func TestPBSSourceLearnerSharedSourceIsNotDecisive(t *testing.T) {
 
 func TestPBSSourceLearnerUnfamiliarComponentStopsResolution(t *testing.T) {
 	l := NewPBSSourceLearner()
+	l.RegisterCandidate("pbs-main", "cluster-a")
+	l.RegisterCandidate("pbs-main", "cluster-b")
 	l.Observe("pbs-main", "backups", "cluster-a@pbs!token", "cluster-a")
+	l.Observe("pbs-main", "store-b", "cluster-b@pbs!token", "cluster-b")
 
 	// The datastore alone would resolve to cluster-a, but an owner token
 	// that was never observed means the backup may come from a cluster we
@@ -125,8 +132,66 @@ func TestPBSSourceLearnerUnfamiliarComponentStopsResolution(t *testing.T) {
 	}
 }
 
+// TestPBSSourceLearnerUnobservedCandidateBlocksResolution covers the
+// asymmetry that made a singleton source set unsafe: clusters only become
+// visible through snapshots that were already attributable, so a cluster with
+// none looks like it does not exist. A source shared with that cluster then
+// maps to a single observed cluster and hands it the other cluster's
+// snapshots (#1639).
+func TestPBSSourceLearnerUnobservedCandidateBlocksResolution(t *testing.T) {
+	l := NewPBSSourceLearner()
+	l.RegisterCandidate("pbs-main", "cluster-a")
+	l.RegisterCandidate("pbs-main", "cluster-b")
+	l.Observe("pbs-main", "backups", "shared@pbs!token", "cluster-a")
+
+	if inst, ok := l.Resolve("pbs-main", "backups", "shared@pbs!token"); ok {
+		t.Fatalf("resolved to %q while cluster-b had no attributed snapshot, want inconclusive", inst)
+	}
+
+	// Once cluster-b is visible with a source of its own, the singleton is
+	// backed by evidence that the token is not shared.
+	l.Observe("pbs-main", "store-b", "cluster-b@pbs!token", "cluster-b")
+	if inst, ok := l.Resolve("pbs-main", "backups", "shared@pbs!token"); !ok || inst != "cluster-a" {
+		t.Fatalf("resolution with full coverage = %q,%v, want cluster-a,true", inst, ok)
+	}
+}
+
+// TestPBSSourceLearnerCandidateVisibleOnOtherPBSInstanceResolves covers the
+// reported topology: each cluster owns a PBS server, so the PBS instance is
+// itself the discriminator. cluster-b never submits to pbs-one, but it is
+// visible on pbs-two, which is what makes pbs-one's singleton meaningful.
+func TestPBSSourceLearnerCandidateVisibleOnOtherPBSInstanceResolves(t *testing.T) {
+	l := NewPBSSourceLearner()
+	l.RegisterCandidate("pbs-one", "cluster-a")
+	l.RegisterCandidate("pbs-one", "cluster-b")
+	l.RegisterCandidate("pbs-two", "cluster-a")
+	l.RegisterCandidate("pbs-two", "cluster-b")
+	l.Observe("pbs-one", "backups", "", "cluster-a")
+	l.Observe("pbs-two", "backups", "", "cluster-b")
+
+	if inst, ok := l.Resolve("pbs-one", "backups", ""); !ok || inst != "cluster-a" {
+		t.Fatalf("pbs-one resolution = %q,%v, want cluster-a,true", inst, ok)
+	}
+	if inst, ok := l.Resolve("pbs-two", "backups", ""); !ok || inst != "cluster-b" {
+		t.Fatalf("pbs-two resolution = %q,%v, want cluster-b,true", inst, ok)
+	}
+}
+
+// TestPBSSourceLearnerWithoutCandidatesIsInert guards the fail-closed
+// default: a caller that never declares the candidate clusters gets no
+// resolution rather than one drawn from whatever it happened to observe.
+func TestPBSSourceLearnerWithoutCandidatesIsInert(t *testing.T) {
+	l := NewPBSSourceLearner()
+	l.Observe("pbs-main", "backups", "cluster-a@pbs!token", "cluster-a")
+
+	if inst, ok := l.Resolve("pbs-main", "backups", "cluster-a@pbs!token"); ok {
+		t.Fatalf("resolved to %q without declared candidates, want inconclusive", inst)
+	}
+}
+
 func TestPBSSourceLearnerNilReceiverIsInert(t *testing.T) {
 	var l *PBSSourceLearner
+	l.RegisterCandidate("pbs-main", "cluster-a")
 	l.Observe("pbs-main", "backups", "owner", "cluster-a")
 	if _, ok := l.Resolve("pbs-main", "backups", "owner"); ok {
 		t.Fatal("nil learner must never be decisive")

internal/recovery/mapper/proxmox/mapper.go

@@ -407,7 +407,20 @@ func FromPBSBackups(backups []models.PBSBackup, candidatesByKey map[string][]Gue
 	// First pass: learn each PBS submission source's cluster from the
 	// backups that are attributable on their own evidence, so the second
 	// pass can resolve collision VMIDs with no evidence of their own (#1639).
+	// Every connection owning a candidate guest is registered first, so a
+	// connection that contributes no attributable backup keeps the learner
+	// inconclusive instead of letting a source it may share resolve to the
+	// connections that happened to be attributable.
 	learner := proxmoxidentity.NewPBSSourceLearner()
+	for _, b := range backups {
+		if strings.TrimSpace(b.ID) == "" {
+			continue
+		}
+		key := strings.ToLower(strings.TrimSpace(b.BackupType)) + ":" + strings.TrimSpace(b.VMID)
+		for _, candidate := range candidatesByKey[key] {
+			learner.RegisterCandidate(b.Instance, candidate.InstanceName)
+		}
+	}
 	for _, b := range backups {
 		if strings.TrimSpace(b.ID) == "" {
 			continue

internal/recovery/mapper/proxmox/mapper_test.go

@@ -551,10 +551,24 @@ func TestFromPBSBackups_DisambiguatesCollisionVMIDBySubmissionSource(t *testing.
 				BackupTypeKey: "vm",
 			},
 		},
+		"vm:150": {
+			{
+				ResourceID:    "vm-dddddddddddddddd",
+				SourceID:      "cluster-b:pve-b1:150",
+				ResourceType:  unifiedresources.ResourceTypeVM,
+				DisplayName:   "db-b",
+				InstanceName:  "cluster-b",
+				NodeName:      "pve-b1",
+				VMID:          150,
+				BackupTypeKey: "vm",
+			},
+		},
 	}
 
 	backups := []models.PBSBackup{
-		// Teaching snapshot: unique VMID attributes cluster-a's owner token.
+		// Teaching snapshots: unique VMIDs attribute each cluster's owner
+		// token, which is also what makes both clusters visible — a cluster
+		// with no attributable backup would keep the learner inconclusive.
 		{
 			ID:         "pbs-a-100",
 			VMID:       "100",
@@ -564,6 +578,15 @@ func TestFromPBSBackups_DisambiguatesCollisionVMIDBySubmissionSource(t *testing.
 			BackupType: "vm",
 			BackupTime: time.Date(2026, 7, 26, 1, 0, 0, 0, time.UTC),
 		},
+		{
+			ID:         "pbs-b-150",
+			VMID:       "150",
+			Instance:   "pbs-main",
+			Datastore:  "store-b",
+			Owner:      "cluster-b@pbs!token",
+			BackupType: "vm",
+			BackupTime: time.Date(2026, 7, 26, 2, 0, 0, 0, time.UTC),
+		},
 		// Collision snapshot: no namespace, no comment, cluster-a's source.
 		{
 			ID:         "pbs-a-173",
@@ -588,8 +611,8 @@ func TestFromPBSBackups_DisambiguatesCollisionVMIDBySubmissionSource(t *testing.
 	}
 
 	result := FromPBSBackups(backups, candidatesByKey)
-	if len(result) != 3 {
-		t.Fatalf("expected 3 points, got %d", len(result))
+	if len(result) != 4 {
+		t.Fatalf("expected 4 points, got %d", len(result))
 	}
 
 	byID := map[string]int{}
@@ -607,3 +630,82 @@ func TestFromPBSBackups_DisambiguatesCollisionVMIDBySubmissionSource(t *testing.
 		t.Fatalf("unattributable snapshot linked to %q, want unlinked", unlinked.SubjectResourceID)
 	}
 }
+
+// Issue #1639 hardening: the learner only ever sees clusters that already had
+// a backup attributed to them, so a cluster with no attributable backup is
+// invisible and a source it shares looks exclusive to the cluster that was
+// visible. The collision snapshot must stay unlinked rather than be handed to
+// the visible cluster.
+func TestFromPBSBackups_InvisibleClusterKeepsCollisionSnapshotUnlinked(t *testing.T) {
+	candidatesByKey := map[string][]GuestCandidate{
+		"vm:173": {
+			{
+				ResourceID:    "vm-aaaaaaaaaaaaaaaa",
+				SourceID:      "cluster-a:pve-a1:173",
+				ResourceType:  unifiedresources.ResourceTypeVM,
+				DisplayName:   "web-a",
+				InstanceName:  "cluster-a",
+				NodeName:      "pve-a1",
+				VMID:          173,
+				BackupTypeKey: "vm",
+			},
+			{
+				ResourceID:    "vm-bbbbbbbbbbbbbbbb",
+				SourceID:      "cluster-b:pve-b1:173",
+				ResourceType:  unifiedresources.ResourceTypeVM,
+				DisplayName:   "web-b",
+				InstanceName:  "cluster-b",
+				NodeName:      "pve-b1",
+				VMID:          173,
+				BackupTypeKey: "vm",
+			},
+		},
+		"vm:100": {
+			{
+				ResourceID:    "vm-cccccccccccccccc",
+				SourceID:      "cluster-a:pve-a1:100",
+				ResourceType:  unifiedresources.ResourceTypeVM,
+				DisplayName:   "db-a",
+				InstanceName:  "cluster-a",
+				NodeName:      "pve-a1",
+				VMID:          100,
+				BackupTypeKey: "vm",
+			},
+		},
+	}
+
+	backups := []models.PBSBackup{
+		{
+			ID:         "pbs-a-100",
+			VMID:       "100",
+			Instance:   "pbs-main",
+			Datastore:  "backups",
+			Owner:      "shared@pbs!token",
+			BackupType: "vm",
+			BackupTime: time.Date(2026, 7, 26, 1, 0, 0, 0, time.UTC),
+		},
+		// Authored by cluster-b, which owns no attributable guest here.
+		{
+			ID:         "pbs-b-173",
+			VMID:       "173",
+			Instance:   "pbs-main",
+			Datastore:  "backups",
+			Owner:      "shared@pbs!token",
+			BackupType: "vm",
+			BackupTime: time.Date(2026, 7, 27, 1, 0, 0, 0, time.UTC),
+		},
+	}
+
+	result := FromPBSBackups(backups, candidatesByKey)
+	if len(result) != 2 {
+		t.Fatalf("expected 2 points, got %d", len(result))
+	}
+	for _, point := range result {
+		if point.ID != "pbs-backup:pbs-b-173" {
+			continue
+		}
+		if point.SubjectResourceID != "" {
+			t.Fatalf("collision snapshot linked to %q, want unlinked while cluster-b is invisible", point.SubjectResourceID)
+		}
+	}
+}