Keep a remediation lock when a reporter gains an identity key

A host agent or Docker host that reported no machine ID, DMI UUID or
cluster slot is keyed by its source-specific ID and holds no identity pin,
because a pin needs a strong key. When a later report carried one the
resource minted a new ID, no pin named the old one, and pin-driven
succession declared nothing. The operator's rows, a remediation lock
included, stayed on an ID the registry no longer lists, while the dispatch
gates read the row of the listed ID. Pulse could then act on a host the
operator had told it never to touch.

The retired ID is recomputable from the report itself, as a Proxmox
guest's retired IDs are. ingestRecord now notes the source-specific ID of
each host agent and Docker host reporter that lands on another ID with a
key, and the snapshot and record ingests hand the batch to
applyRecordSuccessions, which skips an ID still observed. The stores
already carry the row, or its lock onto an existing row, once per
predecessor. Proxmox nodes are left out on purpose: they borrow their
linked agent's key, and a link or split can move that borrow either way.

The other uncovered change named in the contract, a physical disk re-keyed
in place to a machine-scoped ID, is not a way around the lock today: the
planner plans only capabilities a resource advertises and no producer gives
a physical disk one. TestPhysicalDiskResourcesAdvertiseNoActionCapability
pins that for the Proxmox, agent, Unraid and TrueNAS producers, and the
contract now says so, including that the disk's other operator settings are
not carried.

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

@@ -3630,11 +3630,62 @@ resolve the ID (a member absent from the current generation) stays on that ID
 until then. The store cannot see which links the registry declines to fold
 (availability-owned endpoints, a node and agent pair a newer split separated),
 so those share too, which can lock a host because a linked probe was retired.
-Re-recording an existing link shares again. Canonical ID changes that reach
-neither a link nor `ApplyCanonicalIDSuccessions` are not covered here: a
-physical disk re-keyed in place to a machine-scoped ID
-(`rekeyPhysicalDiskLocked` moves no operator state), and a host whose earlier
-ID never held an identity pin, which no pin-driven succession can name. Proof:
+Re-recording an existing link shares again.
+
+A host or Docker reporter that had no machine ID, DMI UUID or cluster slot is
+keyed by its source-specific ID (unless a link or match folds it into another
+resource) and held no identity pin (a pin needs one of those keys), so when a
+report first carries a key, its own or one its pin completes, no pin names the
+ID it leaves. The registry declares that ID itself, as it does a Proxmox guest's
+retired IDs, because the report recomputes it: `noteReporterEraLocked` notes
+the reporter's source-specific ID against the resource its report landed on,
+and the snapshot or record ingest hands the batch to `applyRecordSuccessions`
+(`takeReporterEras`), which skips an ID still observed live or folded into a
+link. Only `SourceAgent` and `SourceDocker` reporters do this, and only in an
+ingest that may write (a registry with a store, not saved-host continuity):
+they own their identity, whereas a Proxmox node borrows its linked agent's key
+and a link or split can move that borrow either way, so a borrowed key is not
+an era. Two kinds of source-specific ID are left out. One that holds an
+identity pin belongs to pin succession, which decides whether another key is
+the same host and lets a different machine mint fresh, so a keyless-era ID that
+gained a cluster-only pin on the way is carried only if that succession names
+it, which it does not for a later machine key. One an operator split names is
+the candidate of a source kept apart from another resource, and a succession
+would rewrite that pair into an exclusion of the resource from itself in a
+generation where the other side is absent, erasing the split, so the
+declaration leaves that reporter's rows alone (a pin-driven succession can
+still re-key such an ID when the split source holds the shared machine key,
+which this does not change). Reporters that gain a key declare
+their era on every rebuild and the stores re-key a predecessor once, to the
+successor it first had (the record is keyed by the predecessor), so a lock the
+operator set on the keyless ID is carried onto the keyed one (the whole row
+when the keyed ID has none). The first rebuild after an upgrade declares the
+era of every keyed reporter that passes those checks, in the batches of its
+ingest passes, and also restores a lock or row an earlier keyless-to-keyed
+change left behind, even where the operator has edited the keyed row since. An ID held under a report ID the reporter no longer sends is
+not recomputable. Like every succession, the move happens while the previous
+generation is still being served, so when the keyed ID had no row an action
+planned against the old ID in that window finds no row there, and a link the
+succession rewrites takes effect in the next generation, because a registry
+loads its links when it is built (the lock is shared across the pair at once).
+
+Canonical ID changes that reach neither a link nor `ApplyCanonicalIDSuccessions`
+carry nothing: a physical disk re-keyed in place to a machine-scoped ID
+(`rekeyPhysicalDiskLocked` moves no operator state, and the disk takes the
+unscoped ID again when the other machine's copy goes away, finding the old row
+as it was). For the lock that is a boundary, not a gap, because the planner
+plans only a capability the resource advertises and no producer gives a
+physical disk one, so no dispatch can reach a disk's ID. A producer that adds
+one must first make the lock follow the disk;
+`TestPhysicalDiskResourcesAdvertiseNoActionCapability` fails when one of the
+registry's producers does (a Proxmox disk, an agent's SMART disk, an Unraid
+array disk, a record-fed disk), and `TestProviderDiskRecordsAdvertiseNoActionCapability`
+in `internal/truenas/provider_test.go` when the TrueNAS provider's records do.
+The disk's other operator settings (retirement, maintenance, notes) are
+orphaned by the same re-key and are not carried. Proof:
+`TestReporterGainingAStrongKeyKeepsItsRemediationLock`,
+`TestReporterEraIsDeclaredOnlyForAKeyTheReporterGains`,
+`TestPhysicalDiskResourcesAdvertiseNoActionCapability`,
 `TestResourceRegistry_ManualLinkFoldKeepsRemediationLock`,
 `TestResourceRegistry_ManualLinkFoldLockLeavesSurvivorSettings`,
 `TestLinkedComponentSharesRemediationLockThroughLinksAndUnlinks`,
@@ -3642,6 +3693,7 @@ ID never held an identity pin, which no pin-driven succession can name. Proof:
 `TestSQLiteStoreRestoresLocksAcrossRecordedSuccessionChains` in
 `internal/unifiedresources/registry_test.go`, and
 `TestCanonicalIDSuccessionCarriesRemediationLockOntoExistingSuccessorRow`,
+`TestCanonicalIDSuccessionRekeysAPredecessorOnce`,
 `TestCanonicalIDSuccessionLockReachesLinkedSurvivor` and
 `TestCanonicalIDSuccessionKeepsLinkedMembersTogether` in
 `internal/unifiedresources/canonical_id_succession_test.go`.

internal/truenas/provider_test.go

@@ -132,6 +132,33 @@ func (s *controllableStubFetcher) SystemMetricHistory(context.Context, time.Dura
 	return &copied, nil
 }
 
+// An operator's remediation lock is not carried across a physical disk's
+// in-place re-key in the unified registry, which is safe only while nothing can
+// dispatch to a disk: the planner plans only a capability the resource
+// advertises. This pins that premise for the records this provider produces
+// (the registry side is TestPhysicalDiskResourcesAdvertiseNoActionCapability).
+func TestProviderDiskRecordsAdvertiseNoActionCapability(t *testing.T) {
+	previous := IsFeatureEnabled()
+	SetFeatureEnabled(true)
+	t.Cleanup(func() {
+		SetFeatureEnabled(previous)
+	})
+
+	disks := 0
+	for _, record := range NewProvider(DefaultFixtures()).Records() {
+		if record.Resource.Type != unifiedresources.ResourceTypePhysicalDisk {
+			continue
+		}
+		disks++
+		if len(record.Resource.Capabilities) != 0 {
+			t.Errorf("disk record %s advertises capabilities %+v; make the remediation lock follow a disk through its in-place re-key first", record.SourceID, record.Resource.Capabilities)
+		}
+	}
+	if disks == 0 {
+		t.Fatal("the fixture produced no disk records, so this pins nothing")
+	}
+}
+
 func TestFixtureFetcherReturnsSnapshotCopy(t *testing.T) {
 	fixtures := DefaultFixtures()
 	fetcher := &FixtureFetcher{Snapshot: fixtures}

internal/unifiedresources/canonical_id_succession.go

@@ -59,7 +59,11 @@ func (s *SQLiteResourceStore) ApplyCanonicalIDSuccessions(successions []Canonica
 		if oldID == "" || newID == "" || oldID == newID {
 			continue
 		}
-		if successor, done := recorded[oldID]; done && successor == newID {
+		// The record is keyed by the predecessor, so it re-keys once, to the
+		// successor it first had (as MemoryStore does). A later, different
+		// successor could only be refused by the insert below, after a write
+		// transaction and a pin-cache invalidation on every rebuild.
+		if _, done := recorded[oldID]; done {
 			continue
 		}
 		pending = append(pending, CanonicalIDSuccession{OldCanonicalID: oldID, NewCanonicalID: newID})

internal/unifiedresources/canonical_id_succession_test.go

@@ -230,6 +230,46 @@ func TestCanonicalIDSuccessionCarriesRemediationLockOntoExistingSuccessorRow(t *
 	}
 }
 
+// A succession is recorded by its predecessor: the predecessor re-keys once, to
+// the successor it first had, and a later declaration naming another successor
+// (a reporter whose key changed again re-declares every rebuild) changes
+// nothing, in either store.
+func TestCanonicalIDSuccessionRekeysAPredecessorOnce(t *testing.T) {
+	stores := map[string]func(t *testing.T) ResourceStore{
+		"memory": func(t *testing.T) ResourceStore { return NewMemoryStore() },
+		"sqlite": func(t *testing.T) ResourceStore {
+			store, err := NewSQLiteResourceStore(t.TempDir(), "default")
+			if err != nil {
+				t.Fatalf("NewSQLiteResourceStore: %v", err)
+			}
+			t.Cleanup(func() { _ = store.Close() })
+			return store
+		},
+	}
+	for name, newStore := range stores {
+		t.Run(name, func(t *testing.T) {
+			store := newStore(t)
+			if err := store.SetResourceOperatorState(ResourceOperatorState{CanonicalID: "once-old", NeverAutoRemediate: true, SetAt: time.Now().UTC(), SetBy: "operator"}); err != nil {
+				t.Fatalf("lock: %v", err)
+			}
+			succeeder := store.(canonicalIDSuccessor)
+			for _, successor := range []string{"once-first", "once-second", "once-first"} {
+				if err := succeeder.ApplyCanonicalIDSuccessions([]CanonicalIDSuccession{{OldCanonicalID: "once-old", NewCanonicalID: successor}}); err != nil {
+					t.Fatalf("ApplyCanonicalIDSuccessions(%s): %v", successor, err)
+				}
+			}
+			if state, found, err := store.GetResourceOperatorState("once-first"); err != nil || !found || !state.BlocksRemediation() {
+				t.Fatalf("the first successor lost the lock: found=%v err=%v state=%+v", found, err, state)
+			}
+			for _, id := range []string{"once-old", "once-second"} {
+				if state, found, err := store.GetResourceOperatorState(id); err != nil || found {
+					t.Fatalf("%s holds a row after the predecessor re-keyed once: found=%v err=%v state=%+v", id, found, err, state)
+				}
+			}
+		})
+	}
+}
+
 // The registry keeps one member of a manual link, and a succession re-keys the
 // link onto the successor. The successor's lock (whether it kept its own row
 // or took the predecessor's whole row) has to reach the member that survives.

internal/unifiedresources/registry.go

@@ -145,6 +145,17 @@ type ResourceRegistry struct {
 	// resource per machine ID, so it cannot see same-serial disks on several
 	// machines. Entries are validated on read, never removed eagerly.
 	physicalDisksByHardware map[string]map[string]struct{}
+	// reporterEras collects, while an ingest pass runs, the source-specific ID
+	// each host or Docker reporter held before an identity key re-keyed its
+	// resource (noteReporterEraLocked). The pass epilogue takes them
+	// (takeReporterEras) and declares them as successions.
+	reporterEras []CanonicalIDSuccession
+	// exclusionIDs holds every ID an entry of exclusions names, derived from
+	// it on first use (exclusionNamesLocked), so noteReporterEraLocked asks one
+	// lookup per reporter instead of scanning the splits. exclusions is fixed
+	// before the first ingest, whether it was read from the store or carried
+	// from the previous generation (carryOverridesFrom).
+	exclusionIDs map[string]struct{}
 
 	// Cached typed view indexes. Invalidated on ingest, rebuilt lazily on
 	// first access. Protected by mu — callers hold RLock to read, and the
@@ -423,7 +434,7 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
 	rr.mu.Lock()
 	rr.agentNodeScanIndex = nil
 	rr.mu.Unlock()
-	rr.applyRecordSuccessions(guestSuccessions)
+	rr.applyRecordSuccessions(append(guestSuccessions, rr.takeReporterEras()...))
 	for _, storage := range snapshot.Storage {
 		rr.ingestStorage(storage)
 	}
@@ -765,7 +776,7 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec
 			})
 		}
 	}
-	rr.applyRecordSuccessions(successions)
+	rr.applyRecordSuccessions(append(successions, rr.takeReporterEras()...))
 
 	rr.mu.Lock()
 	// Either side of an operator link can arrive as a record: a vSphere or
@@ -3084,6 +3095,9 @@ func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, res
 		identity = rr.completeIdentityFromPins(source, sourceID, resource, identity)
 	}
 	resource.Identity = identity
+	if !onlyMissing && rr.store != nil && resource.Type == ResourceTypeAgent && (source == SourceAgent || source == SourceDocker) {
+		defer func() { rr.noteReporterEraLocked(source, sourceID, identity, ingestedID) }()
+	}
 	resource.Sources = []DataSource{source}
 	sighting := resource.SourceStatus[source]
 	sighting.Status = sourceSightingStatus(resource.LastSeen)
@@ -5686,7 +5700,7 @@ func linkedAgentIDFromResource(resource *Resource) string {
 func (rr *ResourceRegistry) chooseNewID(resourceType ResourceType, identity ResourceIdentity, source DataSource, sourceID string) string {
 	switch resourceType {
 	case ResourceTypeAgent:
-		if identity.MachineID != "" || identity.DMIUUID != "" || identity.ClusterName != "" {
+		if agentKeyedByIdentity(identity) {
 			return rr.canonicalIDFromIdentity(resourceType, identity)
 		}
 	case ResourceTypePhysicalDisk:
@@ -5705,6 +5719,62 @@ func (rr *ResourceRegistry) chooseNewID(resourceType ResourceType, identity Reso
 	return rr.sourceSpecificID(resourceType, source, sourceID)
 }
 
+// agentKeyedByIdentity reports whether an agent's identity carries a key
+// chooseNewID derives its canonical ID from. Without one the agent keeps its
+// source-specific ID, so gaining one is an era change of its ID.
+func agentKeyedByIdentity(identity ResourceIdentity) bool {
+	return identity.MachineID != "" || identity.DMIUUID != "" || identity.ClusterName != ""
+}
+
+// noteReporterEraLocked records that a host or Docker reporter's resource is
+// keyed by a stronger identity than its source-specific ID stands for. A
+// reporter with no machine ID, DMI UUID or cluster is keyed by (source, report
+// ID); when a later report carries a key the resource mints another ID. Such a
+// reporter never held an identity pin (a pin needs a strong key), so no pin
+// names the retired ID, but the report itself does: the source-specific ID is
+// recomputable, as a Proxmox guest's retired IDs are
+// (declareGuestSupersededEras). landedID is the resource the report landed on,
+// which pin succession also follows. A reporter owns its identity, unlike a
+// Proxmox node that borrows its linked agent's, so the key it gains is its own
+// and cannot be lent back.
+//
+// Two things keep a source-specific ID from being a retired era. An ID that
+// holds a pin was a keyed machine's: pin succession decides whether a different
+// key is the same host (successionsFor), and a different machine mints fresh.
+// An ID an operator split names is the candidate of a source kept apart from
+// another resource, and a succession would rewrite the pair into an exclusion
+// of the resource from itself the generation the other side is absent and the
+// reporter lands on that resource's ID, erasing the split. The caller holds
+// rr.mu.
+func (rr *ResourceRegistry) noteReporterEraLocked(source DataSource, sourceID string, identity ResourceIdentity, landedID string) {
+	landedID = CanonicalResourceID(landedID)
+	if landedID == "" || !agentKeyedByIdentity(identity) {
+		return
+	}
+	legacyID := rr.sourceSpecificID(ResourceTypeAgent, source, sourceID)
+	if legacyID == "" || legacyID == landedID {
+		return
+	}
+	if rr.identityPins != nil {
+		if _, pinned := rr.identityPins.byCanonicalID[legacyID]; pinned {
+			return
+		}
+	}
+	if rr.exclusionNamesLocked(legacyID) {
+		return
+	}
+	rr.reporterEras = append(rr.reporterEras, CanonicalIDSuccession{OldCanonicalID: legacyID, NewCanonicalID: landedID})
+}
+
+// takeReporterEras returns the eras the ingest pass noted and clears them.
+func (rr *ResourceRegistry) takeReporterEras() []CanonicalIDSuccession {
+	rr.mu.Lock()
+	defer rr.mu.Unlock()
+	eras := rr.reporterEras
+	rr.reporterEras = nil
+	return eras
+}
+
 // physicalDiskIDForMachineLocked decides which resource a new disk
 // observation joins once findMatch has refused every candidate. A usable
 // serial or WWN identifies a disk only on one machine: dual-ported SAS shelves,
@@ -6879,6 +6949,22 @@ func exclusionKey(a, b string) string {
 	return a + "|" + b
 }
 
+// exclusionNamesLocked reports whether any operator split names the ID on
+// either side. The caller holds rr.mu for writing: the first call derives the
+// set of named IDs.
+func (rr *ResourceRegistry) exclusionNamesLocked(id string) bool {
+	if rr.exclusionIDs == nil {
+		rr.exclusionIDs = make(map[string]struct{}, 2*len(rr.exclusions))
+		for key := range rr.exclusions {
+			a, b, _ := strings.Cut(key, "|")
+			rr.exclusionIDs[a] = struct{}{}
+			rr.exclusionIDs[b] = struct{}{}
+		}
+	}
+	_, named := rr.exclusionIDs[id]
+	return named
+}
+
 // Stable ordering helper for deterministic output.
 func sortResourcesByName(resources []Resource) {
 	if len(resources) < 2 {

internal/unifiedresources/registry_test.go

@@ -8463,6 +8463,418 @@ func TestSQLiteStoreRestoresLocksAcrossRecordedSuccessionChains(t *testing.T) {
 	}
 }
 
+// A host agent that reports no strong key is keyed by its source-specific ID
+// and holds no identity pin (a pin needs a machine ID, a DMI UUID or a cluster
+// slot), so when it starts reporting one the pin-driven succession has no
+// predecessor to name and the operator's rows stayed on the ID the host no
+// longer has. The retired ID is recomputable from the report itself, as a
+// Proxmox guest's is, so the registry declares it. Each rebuild is a fresh
+// registry on the durable store followed by the pin persist, as the monitor
+// runs it.
+func TestReporterGainingAStrongKeyKeepsItsRemediationLock(t *testing.T) {
+	stores := map[string]func(t *testing.T) ResourceStore{
+		"memory": func(t *testing.T) ResourceStore { return NewMemoryStore() },
+		"sqlite": func(t *testing.T) ResourceStore {
+			store, err := NewSQLiteResourceStore(t.TempDir(), "default")
+			if err != nil {
+				t.Fatalf("NewSQLiteResourceStore: %v", err)
+			}
+			t.Cleanup(func() { _ = store.Close() })
+			return store
+		},
+	}
+	locks := map[string]ResourceOperatorState{
+		"never auto-remediate": {NeverAutoRemediate: true},
+		"retired":              {LifecycleState: LifecycleStateRetired},
+	}
+	now := time.Now().UTC()
+	host := models.Host{ID: "agent-standalone-01", Hostname: "standalone-01", Status: "online", LastSeen: now}
+	dockerHost := models.DockerHost{ID: "docker-standalone-02", Hostname: "standalone-02", Status: "online", LastSeen: now}
+	// Both reporters own their identity, so both are keyed by the key they
+	// report: the host agent by /etc/machine-id, the Docker host by the engine's
+	// machine ID, or by the Swarm cluster it joins (a cluster slot).
+	swarmHost := models.DockerHost{ID: "docker-standalone-05", Hostname: "standalone-05", Status: "online", LastSeen: now}
+	reporters := map[string]struct {
+		snapshot func(key string) models.StateSnapshot
+		hostname string
+	}{
+		"host agent": {hostname: host.Hostname, snapshot: func(machineID string) models.StateSnapshot {
+			reporter := host
+			reporter.MachineID = machineID
+			return models.StateSnapshot{Hosts: []models.Host{reporter}}
+		}},
+		"docker host": {hostname: dockerHost.Hostname, snapshot: func(machineID string) models.StateSnapshot {
+			reporter := dockerHost
+			reporter.MachineID = machineID
+			return models.StateSnapshot{DockerHosts: []models.DockerHost{reporter}}
+		}},
+		"docker host joining a swarm": {hostname: swarmHost.Hostname, snapshot: func(key string) models.StateSnapshot {
+			reporter := swarmHost
+			if key != "" {
+				reporter.Swarm = &models.DockerSwarmInfo{ClusterID: "swarm-a", ClusterName: "swarm-a"}
+			}
+			return models.StateSnapshot{DockerHosts: []models.DockerHost{reporter}}
+		}},
+	}
+	rebuild := func(t *testing.T, store ResourceStore, snapshot models.StateSnapshot) string {
+		t.Helper()
+		rr := NewRegistry(store)
+		rr.IngestSnapshot(snapshot)
+		rr.PersistIdentityPins()
+		agents := rr.ListByType(ResourceTypeAgent)
+		if len(agents) != 1 {
+			t.Fatalf("rebuild listed %d agents, want the one reporter", len(agents))
+		}
+		return agents[0].ID
+	}
+	for reporterName, reporter := range reporters {
+		for storeName, newStore := range stores {
+			for lockName, lock := range locks {
+				t.Run(reporterName+"/"+storeName+"/"+lockName, func(t *testing.T) {
+					store := newStore(t)
+					keyless := rebuild(t, store, reporter.snapshot(""))
+					state := lock
+					state.CanonicalID, state.SetAt, state.SetBy = keyless, now.Add(-time.Hour), "operator"
+					state.Note = "set before the reporter had a machine ID"
+					if err := store.SetResourceOperatorState(state); err != nil {
+						t.Fatalf("lock %s: %v", keyless, err)
+					}
+
+					keyed := rebuild(t, store, reporter.snapshot("machine-"+reporter.hostname))
+					if keyed == keyless {
+						t.Fatalf("the reporter kept ID %s after gaining a machine ID; the fixture no longer changes its era", keyless)
+					}
+					got, found, err := store.GetResourceOperatorState(keyed)
+					if err != nil {
+						t.Fatalf("read operator state of %s: %v", keyed, err)
+					}
+					if !found || !got.BlocksRemediation() {
+						t.Fatalf("the reporter's new ID %s lost the lock on %s: found=%v state=%+v", keyed, keyless, found, got)
+					}
+					if got.Note != state.Note {
+						t.Fatalf("the reporter's operator row did not move with it: %+v", got)
+					}
+
+					// The operator owns the lock from here: clearing it must hold
+					// across later rebuilds, which declare the same era again.
+					if err := store.ClearResourceOperatorState(keyed); err != nil {
+						t.Fatalf("operator clears %s: %v", keyed, err)
+					}
+					if again := rebuild(t, store, reporter.snapshot("machine-"+reporter.hostname)); again != keyed {
+						t.Fatalf("a steady-state rebuild moved the reporter from %s to %s", keyed, again)
+					}
+					if restored, found, err := store.GetResourceOperatorState(keyed); err != nil || found {
+						t.Fatalf("a later rebuild restored a row the operator cleared: found=%v err=%v state=%+v", found, err, restored)
+					}
+				})
+			}
+		}
+	}
+}
+
+// The era a reporter leaves is declared only for the reporters that own their
+// identity and only when the report carries a key: a reporter that stays
+// keyless keeps its ID, saved-host continuity (which may only introduce absent
+// resources) moves no operator row, and a Proxmox node that borrows its linked
+// agent's key is not an era of the node's own source-specific ID.
+func TestReporterEraIsDeclaredOnlyForAKeyTheReporterGains(t *testing.T) {
+	stores := map[string]func(t *testing.T) ResourceStore{
+		"memory": func(t *testing.T) ResourceStore { return NewMemoryStore() },
+		"sqlite": func(t *testing.T) ResourceStore {
+			store, err := NewSQLiteResourceStore(t.TempDir(), "default")
+			if err != nil {
+				t.Fatalf("NewSQLiteResourceStore: %v", err)
+			}
+			t.Cleanup(func() { _ = store.Close() })
+			return store
+		},
+	}
+	recorded := func(t *testing.T, store ResourceStore) map[string]string {
+		t.Helper()
+		switch s := store.(type) {
+		case *SQLiteResourceStore:
+			return s.successionMap()
+		case *MemoryStore:
+			return s.canonicalSuccessions
+		}
+		t.Fatalf("%T records no successions", store)
+		return nil
+	}
+	lock := func(t *testing.T, store ResourceStore, id string) {
+		t.Helper()
+		if err := store.SetResourceOperatorState(ResourceOperatorState{CanonicalID: id, NeverAutoRemediate: true, SetAt: time.Now().UTC(), SetBy: "operator"}); err != nil {
+			t.Fatalf("lock %s: %v", id, err)
+		}
+	}
+	locked := func(t *testing.T, store ResourceStore, id string) bool {
+		t.Helper()
+		state, found, err := store.GetResourceOperatorState(id)
+		if err != nil {
+			t.Fatalf("read operator state of %s: %v", id, err)
+		}
+		return found && state.BlocksRemediation()
+	}
+	now := time.Now().UTC()
+	host := func(machineID string) models.Host {
+		return models.Host{ID: "agent-standalone-03", Hostname: "standalone-03", MachineID: machineID, Status: "online", LastSeen: now}
+	}
+	hostEra := SourceSpecificID(ResourceTypeAgent, SourceAgent, "agent-standalone-03")
+
+	for storeName, newStore := range stores {
+		t.Run(storeName+"/a reporter that stays keyless declares nothing", func(t *testing.T) {
+			store := newStore(t)
+			lock(t, store, hostEra)
+			for range 2 {
+				rr := NewRegistry(store)
+				rr.IngestSnapshot(models.StateSnapshot{Hosts: []models.Host{host("")}})
+				rr.PersistIdentityPins()
+				if ids := rr.ListByType(ResourceTypeAgent); len(ids) != 1 || ids[0].ID != hostEra {
+					t.Fatalf("a keyless host is listed as %+v, want its source-specific ID %s", ids, hostEra)
+				}
+			}
+			if got := recorded(t, store); len(got) != 0 {
+				t.Fatalf("a keyless reporter declared successions %v", got)
+			}
+			if !locked(t, store, hostEra) {
+				t.Fatalf("the keyless host lost the lock on its own ID")
+			}
+		})
+
+		// A keyless agent linked to a Proxmox node lands on the node's row. That
+		// is a merge by link, not a key the reporter gained, so its
+		// source-specific ID is not declared retired onto the node's.
+		t.Run(storeName+"/a keyless agent absorbed by a linked node declares nothing", func(t *testing.T) {
+			store := newStore(t)
+			snapshot := sharedSerialDiskSnapshot(now, true, "pve1")
+			snapshot.Hosts[0].MachineID = ""
+			linkedHostEra := SourceSpecificID(ResourceTypeAgent, SourceAgent, snapshot.Hosts[0].ID)
+			lock(t, store, linkedHostEra)
+			rr := NewRegistry(store)
+			rr.IngestSnapshot(snapshot)
+			rr.PersistIdentityPins()
+			if agents := rr.ListByType(ResourceTypeAgent); len(agents) != 1 || agents[0].ID == linkedHostEra || !containsDataSource(agents[0].Sources, SourceAgent) {
+				t.Fatalf("the keyless agent is listed as %+v, want it merged into the node's row", agents)
+			}
+			if got := recorded(t, store); len(got) != 0 {
+				t.Fatalf("a keyless agent absorbed by a node declared successions %v", got)
+			}
+			if !locked(t, store, linkedHostEra) {
+				t.Fatalf("the agent's lock moved off %s", linkedHostEra)
+			}
+		})
+
+		t.Run(storeName+"/saved-host continuity moves no operator row", func(t *testing.T) {
+			store := newStore(t)
+			lock(t, store, hostEra)
+			rr := NewRegistry(store)
+			rr.ingestRecords(SourceAgent, []IngestRecord{HostIngestRecord(host("machine-standalone-03"))}, true, nil)
+			if len(rr.ListByType(ResourceTypeAgent)) != 1 {
+				t.Fatalf("continuity did not introduce the saved host")
+			}
+			if got := recorded(t, store); len(got) != 0 {
+				t.Fatalf("continuity hydration declared successions %v", got)
+			}
+			if !locked(t, store, hostEra) {
+				t.Fatalf("continuity hydration moved the lock off %s", hostEra)
+			}
+		})
+
+		t.Run(storeName+"/record ingest declares the era like a snapshot", func(t *testing.T) {
+			store := newStore(t)
+			lock(t, store, hostEra)
+			rr := NewRegistry(store)
+			rr.IngestRecords(SourceAgent, []IngestRecord{HostIngestRecord(host("machine-standalone-03"))})
+			agents := rr.ListByType(ResourceTypeAgent)
+			if len(agents) != 1 || agents[0].ID == hostEra {
+				t.Fatalf("a keyed host record is listed as %+v, want an ID other than its source-specific %s", agents, hostEra)
+			}
+			if !locked(t, store, agents[0].ID) {
+				t.Fatalf("the record-fed reporter's new ID %s lost the lock on %s", agents[0].ID, hostEra)
+			}
+		})
+
+		// An operator split names the split source by its source-specific ID.
+		// While the other side is present the source lands on that ID and
+		// declares nothing; the generation the other side is absent it lands on
+		// the ID it was split from, and declaring its source-specific ID retired
+		// would rewrite the pair into an exclusion of the resource from itself,
+		// so the split would be gone when the other side returned. The pins
+		// decide whether pin succession also reaches the pair (the split source
+		// and its sibling both pin the shared machine key, and the store keeps
+		// one); this holds the key on the sibling, where only a reporter era
+		// could rewrite the pair.
+		t.Run(storeName+"/a split source keeps its exclusion while its sibling is absent", func(t *testing.T) {
+			store := newStore(t)
+			machineID := "machine-shared-04"
+			agentHost := models.Host{ID: "agent-shared-04", Hostname: "shared-04", MachineID: machineID, Status: "online", LastSeen: now}
+			dockerHost := models.DockerHost{ID: "docker-shared-04", Hostname: "shared-04", MachineID: machineID, Status: "online", LastSeen: now}
+			merged := MachineIdentityCanonicalID(ResourceTypeAgent, machineID)
+			candidate := SourceSpecificID(ResourceTypeAgent, SourceDocker, dockerHost.ID)
+			if err := store.AddExclusion(ResourceExclusion{ResourceA: merged, ResourceB: candidate}); err != nil {
+				t.Fatalf("split the Docker source from %s: %v", merged, err)
+			}
+			agents := func(snapshot models.StateSnapshot) int {
+				t.Helper()
+				rr := NewRegistry(store)
+				rr.IngestSnapshot(snapshot)
+				rr.PersistIdentityPins()
+				return len(rr.ListByType(ResourceTypeAgent))
+			}
+			both := models.StateSnapshot{Hosts: []models.Host{agentHost}, DockerHosts: []models.DockerHost{dockerHost}}
+			if got := agents(both); got != 2 {
+				t.Fatalf("with both reporters present the split left %d agents, want the two kept apart", got)
+			}
+			if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
+				CanonicalID: merged, ResourceType: ResourceTypeAgent, MachineID: machineID, Hostname: "shared-04",
+			}}); err != nil {
+				t.Fatalf("hold the machine key on %s: %v", merged, err)
+			}
+			if pins, err := store.ListResourceIdentityPins(); err != nil || len(pins) != 1 || pins[0].CanonicalID != merged {
+				t.Fatalf("pins after holding the key on %s = %+v (err %v), want that one", merged, pins, err)
+			}
+			if got := agents(models.StateSnapshot{DockerHosts: []models.DockerHost{dockerHost}}); got != 1 {
+				t.Fatalf("with the sibling absent %d agents were listed, want the Docker host alone", got)
+			}
+			if got := recorded(t, store); got[candidate] != "" {
+				t.Fatalf("the split source's ID was declared superseded by %s", got[candidate])
+			}
+			if got := agents(both); got != 2 {
+				t.Fatalf("the split did not survive its sibling's absence: %d agents, want the two kept apart", got)
+			}
+		})
+
+		// A rebuild that cannot read its decisions carries the previous
+		// generation's exclusions (carryOverridesFrom) instead of loading them,
+		// and the split is still the operator's: the guard reads the carried
+		// decisions, not only the ones a store read returned. The control is the
+		// same reporter on a store with no split, which does note its era.
+		t.Run(storeName+"/a rebuild that cannot read its decisions still guards the split", func(t *testing.T) {
+			dockerHost := models.DockerHost{ID: "docker-shared-06", Hostname: "shared-06", MachineID: "machine-shared-06", Status: "online", LastSeen: now}
+			merged := MachineIdentityCanonicalID(ResourceTypeAgent, dockerHost.MachineID)
+			candidate := SourceSpecificID(ResourceTypeAgent, SourceDocker, dockerHost.ID)
+			noted := func(rr *ResourceRegistry) bool {
+				t.Helper()
+				resource, identity := resourceFromDockerHost(dockerHost)
+				rr.ingest(SourceDocker, dockerHost.ID, resource, identity)
+				return slices.Contains(rr.reporterEras, CanonicalIDSuccession{OldCanonicalID: candidate, NewCanonicalID: merged})
+			}
+
+			if !noted(NewRegistry(newStore(t))) {
+				t.Fatalf("control: a reporter with no split did not note %s -> %s", candidate, merged)
+			}
+
+			flaky := &unreadableDecisionStore{ResourceStore: newStore(t)}
+			if err := flaky.AddExclusion(ResourceExclusion{ResourceA: merged, ResourceB: candidate}); err != nil {
+				t.Fatalf("split the Docker source from %s: %v", merged, err)
+			}
+			previous := NewRegistry(flaky)
+			flaky.failAll(true)
+			rebuilt := newRegistryFrom(previous)
+			if !rebuilt.overridesUnreadable || len(rebuilt.exclusions) != 1 {
+				t.Fatalf("the rebuild carried %d exclusions (unreadable=%v), want the previous generation's one", len(rebuilt.exclusions), rebuilt.overridesUnreadable)
+			}
+			if noted(rebuilt) {
+				t.Fatalf("a rebuild holding the carried split noted %s superseded", candidate)
+			}
+		})
+
+		// A pin on the source-specific ID means the ID was a keyed machine's.
+		// Pin succession decides whether another key is the same host, and a
+		// different machine mints fresh instead of taking the rows over.
+		t.Run(storeName+"/a pinned source-specific ID is another machine's", func(t *testing.T) {
+			store := newStore(t)
+			lock(t, store, hostEra)
+			if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
+				CanonicalID: hostEra, ResourceType: ResourceTypeAgent, MachineID: "machine-before-reinstall", Hostname: "standalone-03",
+			}}); err != nil {
+				t.Fatalf("pin %s: %v", hostEra, err)
+			}
+			rr := NewRegistry(store)
+			rr.IngestSnapshot(models.StateSnapshot{Hosts: []models.Host{host("machine-after-reinstall")}})
+			rr.PersistIdentityPins()
+			agents := rr.ListByType(ResourceTypeAgent)
+			if len(agents) != 1 || agents[0].ID == hostEra {
+				t.Fatalf("a different machine under the same report ID is listed as %+v, want a fresh ID", agents)
+			}
+			if locked(t, store, agents[0].ID) {
+				t.Fatalf("the different machine %s inherited the pinned machine's lock", agents[0].ID)
+			}
+			if !locked(t, store, hostEra) || recorded(t, store)[hostEra] != "" {
+				t.Fatalf("the pinned machine's row was succeeded onto another machine")
+			}
+		})
+
+		t.Run(storeName+"/a node borrowing its agent's key is not the node's era", func(t *testing.T) {
+			store := newStore(t)
+			nodeEra := SourceSpecificID(ResourceTypeAgent, SourceProxmox, proxmoxNodeSourceID("pve", "pve1"))
+			lock(t, store, nodeEra)
+			rr := NewRegistry(store)
+			rr.IngestSnapshot(sharedSerialDiskSnapshot(now, true, "pve1"))
+			rr.PersistIdentityPins()
+			if got := recorded(t, store); got[nodeEra] != "" {
+				t.Fatalf("the Proxmox node declared its source-specific ID %s superseded by %s", nodeEra, got[nodeEra])
+			}
+			if !locked(t, store, nodeEra) {
+				t.Fatalf("the node's row moved off %s onto the agent's ID", nodeEra)
+			}
+		})
+	}
+}
+
+// A physical disk keyed by a serial that two machines report is re-keyed in
+// place to a machine-scoped ID (rekeyPhysicalDiskLocked), and that moves no
+// operator-state row: the registry keeps no durable record of a disk's IDs, so
+// the stores cannot be told which row to carry. The remediation lock is not
+// carried across that re-key because it guards nothing there: the planner
+// plans only a capability the resource advertises, and no producer gives a
+// physical disk one, so no dispatch can reach a disk's ID. This pins that
+// premise for the producers this package builds disks from (a Proxmox disk,
+// an agent's SMART disk, an Unraid array disk) and for the registry's merge of
+// a record-fed disk; the TrueNAS provider's own records are pinned beside the
+// producer (internal/truenas TestProviderDiskRecordsAdvertiseNoActionCapability).
+// A change that gives a disk a capability must first make the lock follow the
+// disk through the re-key (see the remediation-lock paragraph of the
+// unified-resources contract), or it lets Pulse act on a disk the operator
+// locked.
+func TestPhysicalDiskResourcesAdvertiseNoActionCapability(t *testing.T) {
+	now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)
+	rr := NewRegistry(nil)
+	snapshot := sharedSerialDiskSnapshot(now, true, "pve1", "pve2")
+	snapshot.Hosts = append(snapshot.Hosts, models.Host{
+		ID: "agent-tower", Hostname: "tower", MachineID: "machine-tower", Status: "online", LastSeen: now,
+		Unraid: &models.HostUnraidStorage{Disks: []models.HostUnraidDisk{
+			{Name: "disk1", Device: "sdc", Role: "data", Status: "online", Serial: "UNRAID-LOCK1", Temperature: 37},
+		}},
+	})
+	rr.IngestSnapshot(snapshot)
+	rr.IngestRecords(SourceTrueNAS, trueNASDiskRecords(now, "a", "archive-a", "tank", "TRUENAS-LOCK1"))
+
+	disks := rr.ListByType(ResourceTypePhysicalDisk)
+	for _, disk := range disks {
+		if len(disk.Capabilities) != 0 {
+			t.Errorf("physical disk %s (%v) advertises capabilities %+v; make the remediation lock follow a disk through its in-place re-key first", disk.ID, disk.Sources, disk.Capabilities)
+		}
+	}
+	// Each producer's disk is in the registry under its serial, so a fixture
+	// that stops producing one cannot leave the check above passing for it.
+	for _, want := range []struct {
+		serial string
+		source DataSource
+	}{
+		{sharedDiskSerial, SourceProxmox},
+		{sharedDiskSerial, SourceAgent},
+		{"UNIQUE-SERIAL-1", SourceProxmox},
+		{"UNRAID-LOCK1", SourceAgent},
+		{"TRUENAS-LOCK1", SourceTrueNAS},
+	} {
+		if !slices.ContainsFunc(disks, func(disk Resource) bool {
+			return disk.PhysicalDisk != nil && disk.PhysicalDisk.Serial == want.serial && containsDataSource(disk.Sources, want.source)
+		}) {
+			t.Errorf("no physical disk with serial %s came from %s, so the fixture no longer covers that producer: %d disks", want.serial, want.source, len(disks))
+		}
+	}
+}
+
 // A succession batch is applied in the order it is declared, and each group of
 // linked IDs it saved follows only the re-keys applied after it. A batch
 // declared against its direction, and one that returns an ID to its start, both