Pin each manual-link side from its own identity (#2719)

Stacked on #2715 (its commit is this branch's parent; whichever lands
first, the other still merges).

## Problem

A manual link merges two resources for display, but
`PersistIdentityPins` pinned the merged identity. A link primary was
pinned with the folded resource's machine key, and the store's
one-owner-per-key rule deleted the folded resource's own pin.

For a Proxmox node with a qualified endpoint
(`https://pve1.example.test:8006`) linked to a host agent, the node's
pin records the endpoint, so the next rebuild completed the agent's
machine key from it, minted the agent's machine-derived ID and merged
the two by ID before the link applied. With the node as primary,
succession then rewrote the link into a self-link on the agent's ID; in
either direction, removing the link left the pair merged. Reproduced
through `MonitorAdapter.PopulateSnapshotAndSupplemental` with a
`MemoryStore` on #2715's head.

## Change

- `applyManualLinks` captures each side's own pin before the merge
(`recordLinkOwnPin`, first capture per rebuild wins), and
`PersistIdentityPins` writes those instead of the merged projection,
keeping a pinnable folded side pinned under its own canonical ID.
- Link sides write through a new
`ResourceStore.ReplaceResourceIdentityPins` (both stores). Like an
upsert the row keeps fields the side's own pin leaves empty, but it
drops a strong key another resource holds this rebuild, which is how a
key an earlier release borrowed from the merged identity leaves the row.
- A folded side drops a key a live resource or an earlier folded side
already holds instead of trading it every rebuild, and writes no pin
when that would cost it its machine evidence (a cluster-only row would
be open to cluster-slot succession).
- Proxmox auto-links (LinkedAgentID / identity matcher) merge before any
manual link applies, so their boot-window pins are unchanged
(`TestIssue1753*`, `TestMonitorAdapterRebuildPersistsIdentityPins`
pass).

Known limits, written into the contract: a node row an earlier release
already filled with the agent's key still merges the pair by ID while
the node can complete from it; two live resources sharing a strong key
still alternate, as on main; evidence a later record pass of the same
rebuild merges into an already-joined row reaches neither side's pin.

## Verification

- New tests in `registry_merge_policy_test.go` (standalone and clustered
node, both link directions, three rebuilds with no steady-state pin
writes, then unlink separates the pair; shared-key, kept-field,
hostname-guard, machine-loss and borrowed-key cases) and `store_test.go`
(replace vs upsert on memory and SQLite).
- Each new test fails against #2715's runtime files via `go test
-overlay`; each guard rule fails its own test when removed.
- `internal/unifiedresources` full package, `internal/monitoring`
link/identity tests, `internal/api/resourceapi`, `go build ./...` and
`go vet` pass.
- Contract: `unified-resources.md` clause 25 and "Operator links reach
record-ingested resources".

docs/release-control/v6/internal/subsystems/unified-resources.md

@@ -2767,7 +2767,7 @@ application resource-provider or WebSocket lifecycle.
     (`canonical_id` ↔ machine ID / DMI UUID / cluster / hostname) in the
     `resource_identities` table (`ResourceStore.UpsertResourceIdentityPins`
     / `ListResourceIdentityPins`, written diff-aware after monitor-adapter
-    rebuilds) and completes weak incoming identities from those pins
+    rebuilds; manual-link sides write through `ReplaceResourceIdentityPins`) and completes weak incoming identities from those pins
     before matching and ID derivation
     (`internal/unifiedresources/canonical_id_pins.go`). Persisted primary
     hostnames preserve their normalized dotted form. The pin index resolves
@@ -2783,7 +2783,15 @@ application resource-provider or WebSocket lifecycle.
     must not mint host canonical IDs from snapshot-content-dependent
     identity subsets without consulting the pins, and pin writes stay on
     the durable store-backed registry (ephemeral per-request registries
-    consult, never write). Pinned hostnames preserve the full dotted name
+    consult, never write). A pin records the identity a resource derives
+    from its own sources rather than a manual link's merged projection:
+    each side of a link writes the pin captured before the link merged
+    them, keeping its row's other fields unless another resource holds that
+    key, and a pinnable folded side keeps its pin under its own canonical
+    ID, less any strong key a live resource or an earlier folded side
+    holds. Capture follows pin completion, so a key an earlier release
+    borrowed into a side's row can still reach the capture (see "Operator
+    links reach record-ingested resources"). Pinned hostnames preserve the full dotted name
     (`NormalizeFullHostname`): distinct machines that share a short
     hostname (`cloud.rnd-lax1` vs `cloud.gce-or1`) must keep distinct
     pins, distinct pin-index buckets, and distinct presentation host rows.
@@ -6646,11 +6654,50 @@ separate.
   system's facet, and the system's pools re-parent to the merged row.
 - The registry remembers each resource a manual link folded into its
   primary. A folded resource is still observed, so neither pin-driven nor
-  record-declared succession may succeed its ID. An agent-type primary
-  carries the folded resource's machine keys and is pinned with them, and
-  succeeding the folded ID re-keyed the link onto the primary itself, which
-  split the pair on the next rebuild. That already broke snapshot links whose
-  primary is a machine-keyless agent-type row, such as a Proxmox node.
+  record-declared succession may succeed its ID: that re-keyed the link onto
+  the primary itself, which split the pair on the next rebuild, and already
+  broke snapshot links whose primary is a machine-keyless agent-type row,
+  such as a Proxmox node.
+- Each side of a link is pinned only from its own identity, captured when
+  the link first joins it in a rebuild (`recordLinkOwnPin`). A side with no
+  pinnable identity (not agent-shaped, or no strong key) writes no pin, and
+  a pinnable folded side keeps its pin under its own canonical ID. The merged
+  primary used to be pinned with both sides' keys, and the store's
+  one-owner-per-key rule then deleted the folded side's pin. A Proxmox node's
+  pin records its qualified endpoint as the hostname, so the node's next
+  rebuild completed the linked agent's machine key from that merged pin,
+  minted the agent's machine-derived ID and merged with the agent by ID
+  before the link applied. With the node as primary, succession then
+  rewrote the link into a self-link on the agent's ID; in either direction,
+  removing the link left the pair merged.
+- A link side rewrites its whole row (`ReplaceResourceIdentityPins`)
+  instead of upserting it. Like an upsert it keeps the stored fields its
+  own pin leaves empty, so evidence from an earlier observation survives a
+  weaker one, but it never keeps a machine, DMI or cluster-and-hostname key
+  another resource holds this rebuild, including a cluster key a kept
+  hostname would complete. That is how a key an earlier release borrowed
+  into the row from the merged identity leaves it; a key a resource reports
+  now also beats one a link side's row only kept, as the store already
+  hands a moved key to the new pin. The store gives each strong key one
+  owner and deletes any other row holding a key a written pin carries, so a
+  folded side drops each own key held by a live non-link resource's
+  upserted row (an upsert cannot drop a field), a live link side's own pin,
+  or a folded side earlier in canonical-ID order. It writes no pin when no
+  strong key is left, or when it would lose the machine or DMI key it
+  reports: a cluster-only row reads as a machine-keyless host's, which
+  cluster-slot succession lets a later machine in that slot absorb. A row kept for a side that writes no pin goes
+  when another written pin carries one of its keys, such as the folded
+  agent's own pin carrying the machine key an earlier release borrowed into
+  a Proxmox node's row. That hand-back only reaches a row the node cannot
+  complete from: while the node's qualified endpoint, or its cluster and
+  name, still resolve to that row, completion supplies the borrowed key
+  before any link applies, so the pair merges by ID and succession rewrites
+  the link into a self-link, as it did before this change. A link side whose
+  row loses a borrowed machine key becomes a machine-keyless row, open to
+  the cluster-slot succession any machine-keyless host gets. Two live resources that share a strong key
+  still take it from each other on alternate rebuilds, as before. Evidence
+  a later record pass of the same rebuild merges into an already-joined row
+  reaches neither side's pin.
 - Fold history is registry-local: resource seeding and overlay clones do not
   carry it. A reference to the folded resource's canonical ID, such as an
   availability check configured against an agent later linked into its
@@ -6665,6 +6712,25 @@ resources API seeds, lists the same rows
 pins the node case (`TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence`),
 record-declared succession (`TestRecordDeclaredSuccessionSparesALinkFoldedResource`)
 and both payloads (`TestManualLinkKeepsProviderPayloadsThePrimaryLacks`).
+It also pins each side's own pin for a qualified-endpoint node linked to an
+agent, standalone and clustered, in both directions, through three rebuilds
+with no steady-state pin writes and a separate pair once the link is removed
+(`TestManualLinkPinsEachSideFromItsOwnIdentity`), an unclaimed stored field
+kept without a rewrite every rebuild
+(`TestManualLinkSideKeepsAnUnclaimedStoredField`), a shared
+cluster-and-hostname key kept by the live primary while the folded side keeps
+its machine key, and handed to the folded side when the primary's row only
+kept it (`TestManualLinkFoldedSideNeverTakesALiveResourcesKey`), a kept
+hostname refused where it would complete another resource's cluster key
+(`TestManualLinkSideKeepsNoHostnameThatCompletesAHeldClusterKey`), no pin for a
+folded side whose machine key another host holds
+(`TestManualLinkFoldedSideWritesNoPinWithoutItsMachineKey`), a folded side
+re-pinned after the primary's new pin deletes its widened row
+(`TestManualLinkFoldedSideKeepsItsPinWhenAPrimaryClaimsItsWiderRow`), and the
+agent's own pin taking back the machine key an earlier release borrowed into
+a node's row (`TestManualLinkReturnsABorrowedMachineKeyToItsAgent`).
+`store_test.go` (`TestResourceIdentityPinReplaceClearsFieldsUpsertKeeps`) pins
+replace against upsert on both stores.
 `monitor_adapter_read_state_test.go`
 (`TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds`) pins the
 thresholds on the rebuild, supplemental and overlay paths, and

internal/unifiedresources/canonical_id_pins.go

@@ -3,6 +3,7 @@ package unifiedresources
 import (
 	"log"
 	"net"
+	"slices"
 	"strings"
 )
 
@@ -326,12 +327,106 @@ func (index *identityPinIndex) successionsFor(pin ResourceIdentityPin) []Canonic
 	return successions
 }
 
+// upsertedOnto returns the row UpsertResourceIdentityPins stores when pin is
+// written over existing: an empty field keeps the stored value.
+func (pin ResourceIdentityPin) upsertedOnto(existing ResourceIdentityPin) ResourceIdentityPin {
+	if pin.MachineID == "" {
+		pin.MachineID = existing.MachineID
+	}
+	if pin.DMIUUID == "" {
+		pin.DMIUUID = existing.DMIUUID
+	}
+	if pin.ClusterName == "" {
+		pin.ClusterName = existing.ClusterName
+	}
+	if pin.Hostname == "" {
+		pin.Hostname = existing.Hostname
+	}
+	return pin
+}
+
+// keptOver returns the row a manual-link side writes over existing: like an
+// upsert it keeps the stored fields its own pin leaves empty, so evidence
+// from earlier observations survives a weaker one, but it never keeps a
+// strong key another resource holds. Earlier releases pinned a link's
+// primary with the merged identity, so such a key may be the other side's,
+// and a key a resource reports now beats one a row only kept, as the store
+// already hands a moved key to the new pin.
+func (pin ResourceIdentityPin) keptOver(existing ResourceIdentityPin, held map[string]string) ResourceIdentityPin {
+	if pin.Hostname == "" && (pin.ClusterName == "" || !heldByOther(held, clusterPinKey(pin.ClusterName, existing.Hostname), pin.CanonicalID)) {
+		pin.Hostname = existing.Hostname
+	}
+	if pin.MachineID == "" && existing.MachineID != "" && !heldByOther(held, machinePinKey(existing.MachineID), pin.CanonicalID) {
+		pin.MachineID = existing.MachineID
+	}
+	if pin.DMIUUID == "" && existing.DMIUUID != "" && !heldByOther(held, dmiPinKey(existing.DMIUUID), pin.CanonicalID) {
+		pin.DMIUUID = existing.DMIUUID
+	}
+	if pin.ClusterName == "" && existing.ClusterName != "" && !heldByOther(held, clusterPinKey(existing.ClusterName, pin.Hostname), pin.CanonicalID) {
+		pin.ClusterName = existing.ClusterName
+	}
+	return pin
+}
+
+// withoutKeysHeldElsewhere drops the strong keys another resource holds.
+func (pin ResourceIdentityPin) withoutKeysHeldElsewhere(held map[string]string) ResourceIdentityPin {
+	if pin.MachineID != "" && heldByOther(held, machinePinKey(pin.MachineID), pin.CanonicalID) {
+		pin.MachineID = ""
+	}
+	if pin.DMIUUID != "" && heldByOther(held, dmiPinKey(pin.DMIUUID), pin.CanonicalID) {
+		pin.DMIUUID = ""
+	}
+	if pin.ClusterName != "" && heldByOther(held, clusterPinKey(pin.ClusterName, pin.Hostname), pin.CanonicalID) {
+		pin.ClusterName = ""
+	}
+	return pin
+}
+
+func heldByOther(held map[string]string, key, canonicalID string) bool {
+	owner, taken := held[key]
+	return taken && owner != canonicalID
+}
+
+// ownerKeys lists the strong keys the store lets only one pin hold, matching
+// the conflict rule in UpsertResourceIdentityPins.
+func (pin ResourceIdentityPin) ownerKeys() []string {
+	var keys []string
+	if pin.MachineID != "" {
+		keys = append(keys, machinePinKey(pin.MachineID))
+	}
+	if pin.DMIUUID != "" {
+		keys = append(keys, dmiPinKey(pin.DMIUUID))
+	}
+	if pin.ClusterName != "" {
+		keys = append(keys, clusterPinKey(pin.ClusterName, pin.Hostname))
+	}
+	return keys
+}
+
+func machinePinKey(machineID string) string { return "machine:" + machineID }
+
+func dmiPinKey(dmiUUID string) string { return "dmi:" + dmiUUID }
+
+func clusterPinKey(clusterName, hostname string) string {
+	return "cluster:" + clusterName + "\x00" + hostname
+}
+
 // PersistIdentityPins writes the identity pins for the registry's current
 // host resources into the resource store. Only new or changed pins are
 // written, so steady-state rebuild ticks cost no writes. Call this after a
 // rebuild on the durable store-backed registry; ephemeral per-request
 // registries consult pins but do not write them.
 //
+// Each resource is pinned from its own sources. Both sides of a manual link
+// are pinned from the identity captured before the link merged them
+// (recordLinkOwnPin), and a pinnable folded side keeps its pin under its own
+// canonical ID: a link is display intent, so the merged identity must not
+// lend one machine's keys to the other's pin, or a later rebuild would
+// complete those keys, mint the same canonical ID for both sides and keep
+// them merged after the link is gone. A link side's row is rewritten whole
+// (see keptOver), because an upsert would keep a key an earlier release
+// borrowed into it from the merged identity.
+//
 // When a new pin supersedes an earlier era's pin for the same physical host
 // (see successionsFor), the store re-keys the host's operator-owned rows to
 // the new canonical ID before the pin write, so operator intent
@@ -347,31 +442,93 @@ func (rr *ResourceRegistry) PersistIdentityPins() {
 	}
 
 	rr.mu.RLock()
-	var pins []ResourceIdentityPin
-	var successions []CanonicalIDSuccession
-	for _, resource := range rr.resources {
+	// held maps each strong key to the resource that holds it this rebuild:
+	// a live resource's row (an upsert only ever widens it), each link
+	// side's own pin, and then the keys link sides keep from their rows.
+	held := make(map[string]string)
+	hold := func(pin ResourceIdentityPin) {
+		for _, key := range pin.ownerKeys() {
+			if _, taken := held[key]; !taken {
+				held[key] = pin.CanonicalID
+			}
+		}
+	}
+	var pins, linkPins []ResourceIdentityPin
+	for id, resource := range rr.resources {
+		if own, linked := rr.linkOwnPins[id]; linked {
+			if own != nil {
+				hold(*own)
+				linkPins = append(linkPins, *own)
+			}
+			continue
+		}
 		pin, ok := identityPinForResource(resource)
 		if !ok {
 			continue
 		}
-		if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known && existing == pin {
-			continue
+		effective := pin
+		if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known {
+			effective = pin.upsertedOnto(existing)
 		}
+		hold(effective)
 		pins = append(pins, pin)
-		for _, succession := range rr.identityPins.successionsFor(pin) {
-			// A primary that absorbed a linked resource also carries its
-			// machine keys. That resource is still observed, and succeeding
-			// it would re-key the link onto the primary itself, splitting
-			// the pair again on the next rebuild.
-			if rr.canonicalIDObservedLocked(succession.OldCanonicalID) {
-				continue
-			}
-			successions = append(successions, succession)
+	}
+	folded := make([]string, 0, len(rr.linkOwnPins))
+	for id, own := range rr.linkOwnPins {
+		if _, live := rr.resources[id]; !live && own != nil {
+			folded = append(folded, id)
 		}
 	}
+	slices.Sort(folded)
+	for _, id := range folded {
+		// The store gives each strong key one owner. A folded side drops a
+		// key a live resource, or an earlier folded side, already holds
+		// rather than taking it back on every other rebuild. A live non-link
+		// row is upserted, which cannot drop a field, so its kept fields
+		// hold their keys here. A side that would lose its machine evidence
+		// writes no pin: a cluster-only row reads as a machine-keyless
+		// host's, which cluster-slot succession lets a later machine absorb.
+		own := *rr.linkOwnPins[id]
+		pin := own.withoutKeysHeldElsewhere(held)
+		if !pin.hasStrongKey() || (pin.MachineID == "" && pin.DMIUUID == "" && (own.MachineID != "" || own.DMIUUID != "")) {
+			continue
+		}
+		hold(pin)
+		linkPins = append(linkPins, pin)
+	}
+	slices.SortFunc(linkPins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) })
+	for i, pin := range linkPins {
+		if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known {
+			linkPins[i] = pin.keptOver(existing, held)
+			hold(linkPins[i])
+		}
+	}
+
+	var successions []CanonicalIDSuccession
+	changed := func(batch []ResourceIdentityPin) []ResourceIdentityPin {
+		var out []ResourceIdentityPin
+		for _, pin := range batch {
+			if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known && existing == pin {
+				continue
+			}
+			out = append(out, pin)
+			for _, succession := range rr.identityPins.successionsFor(pin) {
+				// A resource a manual link folded into its primary is still
+				// observed. Succeeding it would re-key the link onto the
+				// primary itself, splitting the pair on the next rebuild.
+				if rr.canonicalIDObservedLocked(succession.OldCanonicalID) {
+					continue
+				}
+				successions = append(successions, succession)
+			}
+		}
+		return out
+	}
+	pins = changed(pins)
+	linkPins = changed(linkPins)
 	rr.mu.RUnlock()
 
-	if len(pins) == 0 {
+	if len(pins) == 0 && len(linkPins) == 0 {
 		return
 	}
 	if len(successions) > 0 {
@@ -381,9 +538,20 @@ func (rr *ResourceRegistry) PersistIdentityPins() {
 			}
 		}
 	}
-	if err := rr.store.UpsertResourceIdentityPins(pins); err != nil {
-		log.Printf("unifiedresources: failed to persist identity pins: %v", err)
-		return
+	// Upserts go first: one that deletes a link side's old row, which still
+	// held a key the side has now dropped, is then followed by the side's
+	// replace, which writes the row again.
+	if len(pins) > 0 {
+		if err := rr.store.UpsertResourceIdentityPins(pins); err != nil {
+			log.Printf("unifiedresources: failed to persist identity pins: %v", err)
+			return
+		}
+	}
+	if len(linkPins) > 0 {
+		if err := rr.store.ReplaceResourceIdentityPins(linkPins); err != nil {
+			log.Printf("unifiedresources: failed to persist manual-link identity pins: %v", err)
+			return
+		}
 	}
 	refreshed, err := rr.store.ListResourceIdentityPins()
 	if err != nil {

internal/unifiedresources/registry.go

@@ -99,6 +99,12 @@ type ResourceRegistry struct {
 	// primary. They are still observed, so identity succession treats them
 	// as live.
 	linkMergedIDs map[string]struct{}
+	// linkOwnPins holds the identity pin each side of a manual link derives
+	// from its own sources, captured before the link merged them. A nil
+	// value records a side with no pinnable identity. PersistIdentityPins
+	// writes these instead of the merged projection, so a link never lends
+	// one machine's identity keys to another resource's durable pin.
+	linkOwnPins map[string]*ResourceIdentityPin
 	// supersededIndex maps record-declared retired canonical IDs to the live
 	// resource that superseded them, so references persisted under a retired
 	// ID (availability links, API reads) keep resolving. An empty value marks
@@ -4848,6 +4854,8 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura
 			primaryID, otherID = otherID, primaryID
 		}
 
+		rr.recordLinkOwnPin(primaryID, primary)
+		rr.recordLinkOwnPin(otherID, other)
 		rr.mergeResourceData(primary, other, thresholds)
 		delete(rr.resources, otherID)
 		if rr.linkMergedIDs == nil {
@@ -4858,6 +4866,24 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura
 	}
 }
 
+// recordLinkOwnPin captures the identity pin a manual-link side derives from
+// its own sources, before the link merge projects the other side's identity
+// onto the primary. The first capture in a registry generation wins, so a
+// primary that absorbs several links keeps the pin it had before any of them.
+func (rr *ResourceRegistry) recordLinkOwnPin(id string, resource *Resource) {
+	if _, recorded := rr.linkOwnPins[id]; recorded {
+		return
+	}
+	if rr.linkOwnPins == nil {
+		rr.linkOwnPins = make(map[string]*ResourceIdentityPin)
+	}
+	if pin, ok := identityPinForResource(resource); ok {
+		rr.linkOwnPins[id] = &pin
+		return
+	}
+	rr.linkOwnPins[id] = nil
+}
+
 func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource, thresholds map[DataSource]time.Duration) {
 	if other == nil || primary == nil {
 		return

internal/unifiedresources/registry_merge_policy_test.go

@@ -763,6 +763,463 @@ func TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence(t *testing.T)
 	}
 }
 
+// pinCountingStore counts identity pin writes, so tests can tell a
+// steady-state rebuild (no writes) from one that rewrites a row every tick.
+type pinCountingStore struct {
+	*MemoryStore
+	pinWrites int
+}
+
+func (s *pinCountingStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error {
+	s.pinWrites++
+	return s.MemoryStore.UpsertResourceIdentityPins(pins)
+}
+
+func (s *pinCountingStore) ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error {
+	s.pinWrites++
+	return s.MemoryStore.ReplaceResourceIdentityPins(pins)
+}
+
+// manualLinkPinSnapshot is a Proxmox node and an unrelated host agent. With a
+// cluster name the node is a cluster member whose instance names the cluster,
+// so its identity carries the cluster.
+func manualLinkPinSnapshot(now time.Time, clusterName string) models.StateSnapshot {
+	instance := "lab"
+	if clusterName != "" {
+		instance = clusterName
+	}
+	return models.StateSnapshot{
+		Nodes: []models.Node{{
+			ID: instance + "-pve1", Name: "pve1", Instance: instance, ClusterName: clusterName, IsClusterMember: clusterName != "",
+			Host: "https://pve1.example.test:8006", Status: "online", LastSeen: now,
+		}},
+		Hosts: []models.Host{{
+			ID: "host-box", Hostname: "box-x", MachineID: "machine-box", Status: "online", LastSeen: now,
+		}},
+		LastUpdate: now,
+	}
+}
+
+// A manual link joins two resources for display; it does not make them one
+// machine, so each side keeps pinning its own identity. A Proxmox node whose
+// configured endpoint is a qualified name pins that name. Pinned with the
+// linked agent's machine key instead, the node's next rebuild completed that
+// key, minted the agent's machine-derived ID and merged with the agent before
+// the link applied, and removing the link could not separate them.
+func TestManualLinkPinsEachSideFromItsOwnIdentity(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		clusterName string
+		nodePrimary bool
+	}{
+		{name: "standalone node primary", nodePrimary: true},
+		{name: "standalone agent primary"},
+		{name: "cluster node primary", clusterName: "lab-cluster", nodePrimary: true},
+		{name: "cluster agent primary", clusterName: "lab-cluster"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			now := time.Now().UTC()
+			snapshot := manualLinkPinSnapshot(now, tc.clusterName)
+			store := &pinCountingStore{MemoryStore: NewMemoryStore()}
+			adapter := NewMonitorAdapter(NewRegistry(store))
+			rebuild := func(step int) {
+				next := snapshot
+				next.LastUpdate = now.Add(time.Duration(step) * time.Second)
+				adapter.PopulateSnapshotAndSupplemental(next, nil)
+			}
+			rebuild(0)
+			var nodeID, agentID string
+			for _, resource := range adapter.GetAll() {
+				if resource.Proxmox != nil {
+					nodeID = resource.ID
+				} else if resource.Agent != nil {
+					agentID = resource.ID
+				}
+			}
+			if nodeID == "" || agentID != MachineIdentityCanonicalID(ResourceTypeAgent, "machine-box") || nodeID == agentID {
+				t.Fatalf("unlinked estate = %v, want a node and the machine-keyed agent", resourceIDs(adapter.GetAll()))
+			}
+			assertOwnPins := func(step string) {
+				t.Helper()
+				pins, err := store.ListResourceIdentityPins()
+				if err != nil {
+					t.Fatalf("%s: list identity pins: %v", step, err)
+				}
+				agentPinned, nodePinned := false, false
+				for _, pin := range pins {
+					switch pin.CanonicalID {
+					case agentID:
+						if pin.MachineID != "machine-box" || pin.Hostname != "box-x" || pin.ClusterName != "" {
+							t.Fatalf("%s: agent pin = %+v, want its own machine key and hostname", step, pin)
+						}
+						agentPinned = true
+					case nodeID:
+						if pin.MachineID != "" || pin.ClusterName != tc.clusterName || pin.Hostname != "pve1.example.test" {
+							t.Fatalf("%s: node pin = %+v, want only its own cluster and endpoint", step, pin)
+						}
+						nodePinned = true
+					default:
+						t.Fatalf("%s: unexpected pin %+v", step, pin)
+					}
+				}
+				if !agentPinned {
+					t.Fatalf("%s: pins = %+v, want the agent's own machine-keyed pin", step, pins)
+				}
+				if tc.clusterName != "" && !nodePinned {
+					t.Fatalf("%s: pins = %+v, want the cluster node's own pin", step, pins)
+				}
+			}
+			assertOwnPins("unlinked")
+
+			primaryID, otherID := agentID, nodeID
+			if tc.nodePrimary {
+				primaryID, otherID = nodeID, agentID
+			}
+			if err := store.AddLink(ResourceLink{ResourceA: primaryID, ResourceB: otherID, PrimaryID: primaryID}); err != nil {
+				t.Fatalf("add link: %v", err)
+			}
+			for step := 1; step <= 3; step++ {
+				writesBefore := store.pinWrites
+				rebuild(step)
+				if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{primaryID}) {
+					t.Fatalf("rebuild %d: resources = %v, want only the linked primary %s", step, got, primaryID)
+				}
+				links, err := store.GetLinks()
+				if err != nil {
+					t.Fatalf("get links: %v", err)
+				}
+				if len(links) != 1 || links[0].ResourceA != primaryID || links[0].ResourceB != otherID {
+					t.Fatalf("rebuild %d: links = %+v, want %s -> %s kept", step, links, primaryID, otherID)
+				}
+				assertOwnPins("linked rebuild")
+				if step > 1 && store.pinWrites != writesBefore {
+					t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", step, store.pinWrites-writesBefore)
+				}
+			}
+
+			// Removing the link separates the pair on the next rebuild:
+			// nothing the link lent either side may outlive it.
+			store.mu.Lock()
+			store.links = nil
+			store.mu.Unlock()
+			rebuild(4)
+			if got, want := resourceIDs(adapter.GetAll()), resourceIDs([]Resource{{ID: nodeID}, {ID: agentID}}); !slices.Equal(got, want) {
+				t.Fatalf("after unlink: resources = %v, want %v", got, want)
+			}
+			for _, resource := range adapter.GetAll() {
+				if resource.ID == nodeID && (resource.Agent != nil || resource.Identity.MachineID != "") {
+					t.Fatalf("after unlink: node = %+v, want no agent payload or borrowed machine key", resource)
+				}
+			}
+			assertOwnPins("after unlink")
+		})
+	}
+}
+
+// Earlier releases pinned a link's primary with the merged identity, so an
+// agent primary's row can still hold the linked node's cluster. No resource
+// holds that cluster with the agent's hostname, so the row keeps it, as an
+// upsert keeps any stored field, and is not rewritten on every rebuild.
+func TestManualLinkSideKeepsAnUnclaimedStoredField(t *testing.T) {
+	now := time.Now().UTC()
+	snapshot := manualLinkPinSnapshot(now, "lab-cluster")
+	snapshot.Nodes[0].Host = "https://192.0.2.10:8006"
+	agentID := MachineIdentityCanonicalID(ResourceTypeAgent, "machine-box")
+	store := &pinCountingStore{MemoryStore: NewMemoryStore()}
+	widened := ResourceIdentityPin{
+		CanonicalID: agentID, ResourceType: ResourceTypeAgent,
+		MachineID: "machine-box", ClusterName: "lab-cluster", Hostname: "box-x",
+	}
+	if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{widened}); err != nil {
+		t.Fatalf("seed widened pin: %v", err)
+	}
+	unlinked := NewMonitorAdapter(NewRegistry(nil))
+	unlinked.PopulateSnapshotAndSupplemental(snapshot, nil)
+	var nodeID string
+	for _, resource := range unlinked.GetAll() {
+		if resource.Proxmox != nil {
+			nodeID = resource.ID
+		}
+	}
+	if nodeID == "" || nodeID == agentID {
+		t.Fatalf("unlinked estate = %v, want a node beside agent %s", resourceIDs(unlinked.GetAll()), agentID)
+	}
+	if err := store.AddLink(ResourceLink{ResourceA: agentID, ResourceB: nodeID, PrimaryID: agentID}); err != nil {
+		t.Fatalf("add link: %v", err)
+	}
+
+	adapter := NewMonitorAdapter(NewRegistry(store))
+	for step := 0; step < 3; step++ {
+		writesBefore := store.pinWrites
+		next := snapshot
+		next.LastUpdate = now.Add(time.Duration(step) * time.Second)
+		adapter.PopulateSnapshotAndSupplemental(next, nil)
+		if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{agentID}) {
+			t.Fatalf("rebuild %d: resources = %v, want only the linked agent %s", step, got, agentID)
+		}
+		if step > 0 && store.pinWrites != writesBefore {
+			t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", step, store.pinWrites-writesBefore)
+		}
+		pins, err := store.ListResourceIdentityPins()
+		if err != nil {
+			t.Fatalf("list identity pins: %v", err)
+		}
+		for _, pin := range pins {
+			if pin.CanonicalID == agentID && pin != widened {
+				t.Fatalf("rebuild %d: agent pin = %+v, want the stored row kept %+v", step, pin, widened)
+			}
+		}
+	}
+}
+
+// The store gives each strong key one owner. Two hosts can keep separate
+// machine IDs yet share a pool name and hostname (an old and a replacement
+// enrollment); linked, the live primary keeps that key and the folded side
+// keeps only its machine key, rather than the two taking the pool from each
+// other every rebuild. A key the primary's row only kept from an earlier
+// release's merged pin goes to the folded side that reports it.
+func TestManualLinkFoldedSideNeverTakesALiveResourcesKey(t *testing.T) {
+	newPin := ResourceIdentityPin{CanonicalID: "agent-new", ResourceType: ResourceTypeAgent, MachineID: "machine-new", ClusterName: "pool-a", Hostname: "xcp1"}
+	oldPin := ResourceIdentityPin{CanonicalID: "agent-old", ResourceType: ResourceTypeAgent, MachineID: "machine-old", ClusterName: "pool-a", Hostname: "xcp1"}
+	ownNewPin, ownOldPin := newPin, oldPin
+	ownNewPin.ClusterName, ownOldPin.ClusterName = "", ""
+	for _, tc := range []struct {
+		name        string
+		primaryPool string
+		seed        []ResourceIdentityPin
+		want        []ResourceIdentityPin
+	}{
+		{name: "both report the pool", primaryPool: "pool-a", want: []ResourceIdentityPin{newPin, ownOldPin}},
+		{name: "primary row borrowed the pool", seed: []ResourceIdentityPin{newPin}, want: []ResourceIdentityPin{ownNewPin, oldPin}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			now := time.Now().UTC()
+			host := func(id, machineID, pool string) Resource {
+				return Resource{
+					ID: id, Type: ResourceTypeAgent, Name: "xcp1", Status: StatusOnline, LastSeen: now,
+					Sources:  []DataSource{SourceAgent},
+					Agent:    &AgentData{AgentID: id, Hostname: "xcp1"},
+					Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: []string{"xcp1"}},
+				}
+			}
+			store := &pinCountingStore{MemoryStore: NewMemoryStore()}
+			if err := store.UpsertResourceIdentityPins(tc.seed); err != nil {
+				t.Fatalf("seed pins: %v", err)
+			}
+			if err := store.AddLink(ResourceLink{ResourceA: "agent-new", ResourceB: "agent-old", PrimaryID: "agent-new"}); err != nil {
+				t.Fatalf("add link: %v", err)
+			}
+			for rebuild := 1; rebuild <= 3; rebuild++ {
+				writesBefore := store.pinWrites
+				registry := NewRegistry(store)
+				registry.IngestResources([]Resource{host("agent-new", "machine-new", tc.primaryPool), host("agent-old", "machine-old", "pool-a")})
+				registry.PersistIdentityPins()
+				if rebuild > 1 && store.pinWrites != writesBefore {
+					t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", rebuild, store.pinWrites-writesBefore)
+				}
+				pins, err := store.ListResourceIdentityPins()
+				if err != nil {
+					t.Fatalf("list identity pins: %v", err)
+				}
+				slices.SortFunc(pins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) })
+				if !slices.Equal(pins, tc.want) {
+					t.Fatalf("rebuild %d: pins = %+v, want %+v", rebuild, pins, tc.want)
+				}
+			}
+		})
+	}
+}
+
+// Two hosts can share a cloned machine ID while an exclusion keeps them
+// apart. A folded side whose machine key the other host holds writes no pin
+// rather than a cluster-only one: cluster-slot succession treats a
+// machine-keyless row as a host a later machine in that slot may absorb.
+func TestManualLinkFoldedSideWritesNoPinWithoutItsMachineKey(t *testing.T) {
+	now := time.Now().UTC()
+	host := func(id, machineID, pool, hostname string) Resource {
+		return Resource{
+			ID: id, Type: ResourceTypeAgent, Name: hostname, Status: StatusOnline, LastSeen: now,
+			Sources:  []DataSource{SourceAgent},
+			Agent:    &AgentData{AgentID: id, Hostname: hostname},
+			Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: []string{hostname}},
+		}
+	}
+	store := NewMemoryStore()
+	if err := store.AddLink(ResourceLink{ResourceA: "agent-primary", ResourceB: "agent-clone", PrimaryID: "agent-primary"}); err != nil {
+		t.Fatalf("add link: %v", err)
+	}
+	registry := NewRegistry(store)
+	registry.IngestResources([]Resource{
+		host("agent-original", "machine-cloned", "", "swarm-a"),
+		host("agent-clone", "machine-cloned", "swarm", "swarm-b"),
+		host("agent-primary", "machine-primary", "", "primary"),
+	})
+	registry.PersistIdentityPins()
+	pins, err := store.ListResourceIdentityPins()
+	if err != nil {
+		t.Fatalf("list identity pins: %v", err)
+	}
+	for _, pin := range pins {
+		if pin.CanonicalID == "agent-clone" {
+			t.Fatalf("folded clone pinned as %+v, want no pin without its machine key", pin)
+		}
+	}
+	if len(pins) != 2 {
+		t.Fatalf("pins = %+v, want the original's and the primary's", pins)
+	}
+}
+
+// A link side's row keeps a stored hostname its own pin lacks, but never
+// where that hostname would complete a cluster key another resource holds:
+// the store would delete that resource's whole row.
+func TestManualLinkSideKeepsNoHostnameThatCompletesAHeldClusterKey(t *testing.T) {
+	now := time.Now().UTC()
+	host := func(id, machineID, pool string, hostnames ...string) Resource {
+		return Resource{
+			ID: id, Type: ResourceTypeAgent, Name: id, Status: StatusOnline, LastSeen: now,
+			Sources:  []DataSource{SourceAgent},
+			Agent:    &AgentData{AgentID: id},
+			Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: hostnames},
+		}
+	}
+	store := &pinCountingStore{MemoryStore: NewMemoryStore()}
+	if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
+		CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", ClusterName: "pool-d", Hostname: "h1",
+	}}); err != nil {
+		t.Fatalf("seed pin: %v", err)
+	}
+	if err := store.AddLink(ResourceLink{ResourceA: "agent-a", ResourceB: "agent-f", PrimaryID: "agent-a"}); err != nil {
+		t.Fatalf("add link: %v", err)
+	}
+	want := []ResourceIdentityPin{
+		{CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", ClusterName: "pool-c"},
+		{CanonicalID: "agent-b", ResourceType: ResourceTypeAgent, MachineID: "machine-b", ClusterName: "pool-c", Hostname: "h1"},
+		{CanonicalID: "agent-f", ResourceType: ResourceTypeAgent, MachineID: "machine-f", Hostname: "f1"},
+	}
+	for rebuild := 1; rebuild <= 3; rebuild++ {
+		writesBefore := store.pinWrites
+		registry := NewRegistry(store)
+		registry.IngestResources([]Resource{
+			host("agent-a", "machine-a", "pool-c"),
+			host("agent-b", "machine-b", "pool-c", "h1"),
+			host("agent-f", "machine-f", "", "f1"),
+		})
+		registry.PersistIdentityPins()
+		if rebuild > 1 && store.pinWrites != writesBefore {
+			t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", rebuild, store.pinWrites-writesBefore)
+		}
+		pins, err := store.ListResourceIdentityPins()
+		if err != nil {
+			t.Fatalf("list identity pins: %v", err)
+		}
+		slices.SortFunc(pins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) })
+		if !slices.Equal(pins, want) {
+			t.Fatalf("rebuild %d: pins = %+v, want %+v", rebuild, pins, want)
+		}
+	}
+}
+
+// A folded side's own pin can be narrower than a row an earlier release
+// widened with a pool its primary now reports. The primary's new pin claims
+// that pool and hostname, so the store deletes the folded side's whole row;
+// the folded side's own pin is written after it, so the side stays pinned.
+func TestManualLinkFoldedSideKeepsItsPinWhenAPrimaryClaimsItsWiderRow(t *testing.T) {
+	now := time.Now().UTC()
+	host := func(id, machineID, pool string) Resource {
+		return Resource{
+			ID: id, Type: ResourceTypeAgent, Name: "xcp1", Status: StatusOnline, LastSeen: now,
+			Sources:  []DataSource{SourceAgent},
+			Agent:    &AgentData{AgentID: id, Hostname: "xcp1"},
+			Identity: ResourceIdentity{MachineID: machineID, ClusterName: pool, Hostnames: []string{"xcp1"}},
+		}
+	}
+	store := &pinCountingStore{MemoryStore: NewMemoryStore()}
+	if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
+		CanonicalID: "agent-old", ResourceType: ResourceTypeAgent,
+		MachineID: "machine-old", ClusterName: "pool-a", Hostname: "xcp1",
+	}}); err != nil {
+		t.Fatalf("seed widened pin: %v", err)
+	}
+	if err := store.AddLink(ResourceLink{ResourceA: "agent-new", ResourceB: "agent-old", PrimaryID: "agent-new"}); err != nil {
+		t.Fatalf("add link: %v", err)
+	}
+	want := []ResourceIdentityPin{
+		{CanonicalID: "agent-new", ResourceType: ResourceTypeAgent, MachineID: "machine-new", ClusterName: "pool-a", Hostname: "xcp1"},
+		{CanonicalID: "agent-old", ResourceType: ResourceTypeAgent, MachineID: "machine-old", Hostname: "xcp1"},
+	}
+	for rebuild := 1; rebuild <= 3; rebuild++ {
+		writesBefore := store.pinWrites
+		registry := NewRegistry(store)
+		registry.IngestResources([]Resource{host("agent-new", "machine-new", "pool-a"), host("agent-old", "machine-old", "")})
+		registry.PersistIdentityPins()
+		if rebuild > 1 && store.pinWrites != writesBefore {
+			t.Fatalf("rebuild %d: wrote identity pins %d time(s), want none in steady state", rebuild, store.pinWrites-writesBefore)
+		}
+		pins, err := store.ListResourceIdentityPins()
+		if err != nil {
+			t.Fatalf("list identity pins: %v", err)
+		}
+		slices.SortFunc(pins, func(a, b ResourceIdentityPin) int { return strings.Compare(a.CanonicalID, b.CanonicalID) })
+		if !slices.Equal(pins, want) {
+			t.Fatalf("rebuild %d: pins = %+v, want %+v", rebuild, pins, want)
+		}
+	}
+}
+
+// An earlier release pinned a linked node primary with the agent's machine
+// key and dropped the agent's own pin. The node has nothing of its own to
+// pin, so the agent's own pin takes the key back on the next rebuild.
+func TestManualLinkReturnsABorrowedMachineKeyToItsAgent(t *testing.T) {
+	now := time.Now().UTC()
+	snapshot := manualLinkPinSnapshot(now, "")
+	snapshot.Nodes[0].Host = "https://192.0.2.10:8006"
+	unlinked := NewMonitorAdapter(NewRegistry(nil))
+	unlinked.PopulateSnapshotAndSupplemental(snapshot, nil)
+	var nodeID string
+	for _, resource := range unlinked.GetAll() {
+		if resource.Proxmox != nil {
+			nodeID = resource.ID
+		}
+	}
+	agentID := MachineIdentityCanonicalID(ResourceTypeAgent, "machine-box")
+	if nodeID == "" || nodeID == agentID {
+		t.Fatalf("unlinked estate = %v, want a node beside agent %s", resourceIDs(unlinked.GetAll()), agentID)
+	}
+
+	store := &pinCountingStore{MemoryStore: NewMemoryStore()}
+	if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
+		CanonicalID: nodeID, ResourceType: ResourceTypeAgent, MachineID: "machine-box", Hostname: "pve1",
+	}}); err != nil {
+		t.Fatalf("seed borrowed pin: %v", err)
+	}
+	if err := store.AddLink(ResourceLink{ResourceA: nodeID, ResourceB: agentID, PrimaryID: nodeID}); err != nil {
+		t.Fatalf("add link: %v", err)
+	}
+	adapter := NewMonitorAdapter(NewRegistry(store))
+	for step := 0; step < 2; step++ {
+		next := snapshot
+		next.LastUpdate = now.Add(time.Duration(step) * time.Second)
+		adapter.PopulateSnapshotAndSupplemental(next, nil)
+		if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{nodeID}) {
+			t.Fatalf("rebuild %d: resources = %v, want only the linked node %s", step, got, nodeID)
+		}
+		pins, err := store.ListResourceIdentityPins()
+		if err != nil {
+			t.Fatalf("list identity pins: %v", err)
+		}
+		if len(pins) != 1 || pins[0].CanonicalID != agentID || pins[0].MachineID != "machine-box" || pins[0].Hostname != "box-x" {
+			t.Fatalf("rebuild %d: pins = %+v, want only the agent's own pin", step, pins)
+		}
+	}
+	links, err := store.GetLinks()
+	if err != nil {
+		t.Fatalf("get links: %v", err)
+	}
+	if len(links) != 1 || links[0].ResourceA != nodeID || links[0].ResourceB != agentID {
+		t.Fatalf("links = %+v, want %s -> %s kept", links, nodeID, agentID)
+	}
+}
+
 // A record may name an older canonical ID as superseded on every ingest. When
 // that ID belongs to a resource a manual link folded into the record's own
 // resource, it is still observed: succeeding it on a later refresh would

internal/unifiedresources/store.go

@@ -29,6 +29,10 @@ type ResourceStore interface {
 	// Identity pins keep canonical IDs for merged-source hosts stable across
 	// restarts. See ResourceIdentityPin.
 	UpsertResourceIdentityPins(pins []ResourceIdentityPin) error
+	// ReplaceResourceIdentityPins writes each pin as its row's whole
+	// content, clearing fields the pin leaves empty, where an upsert keeps
+	// them. Strong-key conflicts resolve as for an upsert.
+	ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error
 	ListResourceIdentityPins() ([]ResourceIdentityPin, error)
 	RecordChange(change ResourceChange) error
 	ResourceHistoryIDs(resourceID string) ([]string, error)
@@ -1521,9 +1525,39 @@ func (s *SQLiteResourceStore) Close() error {
 }
 
 func (s *SQLiteResourceStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error {
+	return s.writeResourceIdentityPins(pins, false)
+}
+
+func (s *SQLiteResourceStore) ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error {
+	return s.writeResourceIdentityPins(pins, true)
+}
+
+// writeResourceIdentityPins upserts pins. An upsert keeps a stored field the
+// pin leaves empty; a replace writes the pin as the whole row.
+func (s *SQLiteResourceStore) writeResourceIdentityPins(pins []ResourceIdentityPin, replace bool) error {
 	if len(pins) == 0 {
 		return nil
 	}
+	write := `INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname)
+			VALUES (?, ?, ?, ?, ?, ?)
+			ON CONFLICT(canonical_id) DO UPDATE SET
+				resource_type = excluded.resource_type,
+				machine_id = COALESCE(excluded.machine_id, resource_identities.machine_id),
+				dmi_uuid = COALESCE(excluded.dmi_uuid, resource_identities.dmi_uuid),
+				cluster_name = COALESCE(NULLIF(excluded.cluster_name, ''), resource_identities.cluster_name),
+				primary_hostname = COALESCE(NULLIF(excluded.primary_hostname, ''), resource_identities.primary_hostname),
+				updated_at = CURRENT_TIMESTAMP`
+	if replace {
+		write = `INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname)
+			VALUES (?, ?, ?, ?, ?, ?)
+			ON CONFLICT(canonical_id) DO UPDATE SET
+				resource_type = excluded.resource_type,
+				machine_id = excluded.machine_id,
+				dmi_uuid = excluded.dmi_uuid,
+				cluster_name = excluded.cluster_name,
+				primary_hostname = excluded.primary_hostname,
+				updated_at = CURRENT_TIMESTAMP`
+	}
 
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -1560,15 +1594,7 @@ func (s *SQLiteResourceStore) UpsertResourceIdentityPins(pins []ResourceIdentity
 		); err != nil {
 			return fmt.Errorf("clear conflicting identity pins for %q: %w", pin.CanonicalID, err)
 		}
-		if _, err := tx.Exec(`INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname)
-			VALUES (?, ?, ?, ?, ?, ?)
-			ON CONFLICT(canonical_id) DO UPDATE SET
-				resource_type = excluded.resource_type,
-				machine_id = COALESCE(excluded.machine_id, resource_identities.machine_id),
-				dmi_uuid = COALESCE(excluded.dmi_uuid, resource_identities.dmi_uuid),
-				cluster_name = COALESCE(NULLIF(excluded.cluster_name, ''), resource_identities.cluster_name),
-				primary_hostname = COALESCE(NULLIF(excluded.primary_hostname, ''), resource_identities.primary_hostname),
-				updated_at = CURRENT_TIMESTAMP`,
+		if _, err := tx.Exec(write,
 			pin.CanonicalID,
 			string(pin.ResourceType),
 			nullIfEmptyArg(pin.MachineID),
@@ -3413,6 +3439,14 @@ func NewMemoryStore() *MemoryStore {
 }
 
 func (m *MemoryStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error {
+	return m.writeResourceIdentityPins(pins, false)
+}
+
+func (m *MemoryStore) ReplaceResourceIdentityPins(pins []ResourceIdentityPin) error {
+	return m.writeResourceIdentityPins(pins, true)
+}
+
+func (m *MemoryStore) writeResourceIdentityPins(pins []ResourceIdentityPin, replace bool) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	for _, pin := range pins {
@@ -3432,7 +3466,7 @@ func (m *MemoryStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) err
 				delete(m.identityPins, canonicalID)
 			}
 		}
-		if existing, ok := m.identityPins[pin.CanonicalID]; ok {
+		if existing, ok := m.identityPins[pin.CanonicalID]; ok && !replace {
 			if pin.MachineID == "" {
 				pin.MachineID = existing.MachineID
 			}

internal/unifiedresources/store_test.go

@@ -2864,6 +2864,66 @@ func TestSQLiteResourceStore_ResourceOperatorState_RoundTrips(t *testing.T) {
 	}
 }
 
+// An upsert keeps a stored field the pin leaves empty; a replace writes the
+// pin as the whole row. Both give a strong key to the pin that carries it.
+func TestResourceIdentityPinReplaceClearsFieldsUpsertKeeps(t *testing.T) {
+	sqliteStore, err := NewSQLiteResourceStore(t.TempDir(), "")
+	if err != nil {
+		t.Fatalf("open sqlite store: %v", err)
+	}
+	defer sqliteStore.Close()
+
+	wide := ResourceIdentityPin{CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", DMIUUID: "dmi-a", ClusterName: "pool-a", Hostname: "host-a"}
+	narrow := ResourceIdentityPin{CanonicalID: "agent-a", ResourceType: ResourceTypeAgent, MachineID: "machine-a", Hostname: "host-a.lan"}
+	rival := ResourceIdentityPin{CanonicalID: "agent-b", ResourceType: ResourceTypeAgent, MachineID: "machine-b", ClusterName: "pool-a", Hostname: "host-a"}
+	for name, store := range map[string]ResourceStore{"memory": NewMemoryStore(), "sqlite": sqliteStore} {
+		t.Run(name, func(t *testing.T) {
+			pinFor := func(id string) (ResourceIdentityPin, bool) {
+				t.Helper()
+				pins, err := store.ListResourceIdentityPins()
+				if err != nil {
+					t.Fatalf("list pins: %v", err)
+				}
+				for _, pin := range pins {
+					if pin.CanonicalID == id {
+						return pin, true
+					}
+				}
+				return ResourceIdentityPin{}, false
+			}
+			if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{wide}); err != nil {
+				t.Fatalf("upsert wide pin: %v", err)
+			}
+			if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{narrow}); err != nil {
+				t.Fatalf("upsert narrow pin: %v", err)
+			}
+			kept := wide
+			kept.Hostname = narrow.Hostname
+			if got, _ := pinFor("agent-a"); got != kept {
+				t.Fatalf("after upsert: pin = %+v, want stored fields kept %+v", got, kept)
+			}
+			if err := store.ReplaceResourceIdentityPins([]ResourceIdentityPin{narrow}); err != nil {
+				t.Fatalf("replace narrow pin: %v", err)
+			}
+			if got, _ := pinFor("agent-a"); got != narrow {
+				t.Fatalf("after replace: pin = %+v, want exactly %+v", got, narrow)
+			}
+			if err := store.ReplaceResourceIdentityPins([]ResourceIdentityPin{wide}); err != nil {
+				t.Fatalf("replace wide pin: %v", err)
+			}
+			if err := store.ReplaceResourceIdentityPins([]ResourceIdentityPin{rival}); err != nil {
+				t.Fatalf("replace rival pin: %v", err)
+			}
+			if _, ok := pinFor("agent-a"); ok {
+				t.Fatalf("a replace claiming pool-a/host-a must delete the row holding that key")
+			}
+			if got, _ := pinFor("agent-b"); got != rival {
+				t.Fatalf("rival pin = %+v, want %+v", got, rival)
+			}
+		})
+	}
+}
+
 func TestResourceChangeReadsMergeCanonicalIDEras(t *testing.T) {
 	const machineID = "7d465a78-test-machine-id"
 	steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID)