Merge candidate 20260906T094514Z-core-runtime

Change-source: pulse-maintainer

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

@@ -7861,29 +7861,38 @@ symmetric bridges would miss removal of filtering from one inventory.
 This proof does not establish safety for arbitrarily renamed Docker bridges
 or repair persisted links, and is not reporter or installed-release validation.
 
-### Manual host/node association safety boundary (6 September 2026)
+### Durable host/node association intent (6 September 2026)
 
-`TestManualHostLinkSurvivesUnmatchedReportsAndProviderRefresh` protects an
-operator-created association between `nas.example` and an otherwise unmatched
-`pve` node through repeated reports, `UpdateNodesForInstance`, and explicit
-unlink. Reapplying the rejected stale-link cleanup plus provider-name matcher
-change makes the first report fail this regression. Neither change is included
-with this test.
+The earlier restart reproduction showed that the link API only changed memory,
+and that provider-name cleanup erased unmatched manual links. The replacement
+uses internal `NodeLinkSource` provenance in the host continuity journal:
+`manual`, `unlinked`, `automatic`, or empty (legacy/unknown). This is not a new
+API or UI surface.
 
-Restart remains an independently reproduced gap, not a supported invariant:
-using `newHostRemovalLifecycleMonitor` with a temporary data directory, submit
-`hostRemovalLifecycleReport("machine", "machine", "agent", "nas.example",
-"linux", now)`, populate the unmatched `pve-node`, and call `LinkHostAgent`.
-Reconstructing the monitor from the same directory immediately yields a host
-without the link. Sending another report before reconstruction persists the
-link, but refreshing the provider on the reconstructed monitor and submitting
-the next report still returns an empty `LinkedNodeID`.
+Link and unlink take the host lifecycle write lock and commit all affected
+journal entries while holding the state lock, before publishing visible state.
+A failed write (or unavailable store) returns an error without changing either
+direction of the link. Reassigning an occupied node writes the displaced host's
+explicit unlink in the same transaction. Identity, report ordering and removal
+metadata are preserved.
 
-`LinkHostAgent` currently changes state only. `HostContinuityEntry` stores the
-target ID but no manual/automatic provenance. Legacy manual and automatic links
-can therefore have identical persisted representations. A repair must not
-classify every unmarked legacy link as automatic. Durable operator intent,
-failed-write handling, provider identity changes, and explicit unlink across
-restart need coverage before enabling destructive stale-link cleanup. This is
-maintenance evidence, not demand for a new user-visible linking surface or a
-claim of resolution of issue #1930.
+Manual links are pinned to the operator-selected provider ID, including through
+unmatched reports and restart before another report. Provider identity replacement
+does not authorise redirection: intent stays dormant if that ID disappears and
+reattaches if it returns. A persisted reservation also prevents another host's
+automatic match from taking the node before its owner reconnects. Replacing a
+dormant owner explicitly persists that owner's unlink too. Explicit unlink persists and suppresses automatic node
+reassociation even when a subsequent report matches. An explicit link can select
+a replacement ID.
+
+Only known automatic associations are re-evaluated destructively against provider
+names and network evidence. Obsolete reverse links are removed with host updates.
+An unmarked persisted link is not assumed automatic: report ingestion retains it,
+so this is deliberately **not** a blanket repair of existing v6.4.1 associations.
+Legacy provider reconciliation otherwise retains its previous behaviour.
+
+`host_manual_link_regression_test.go` covers immediate restart, unlink with
+positive matching evidence, provider replacement/return, write failure and
+automatic versus unknown-provenance cleanup. State and config tests cover atomic
+replacement and preservation of lifecycle evidence. These are synthetic local
+proofs, not reporter confirmation or installed-release resolution of #1930.

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

@@ -39,6 +39,54 @@ ambiguous-name tests remain required. These checks prevent a reproduced
 synthetic misassociation, not all private-address collisions or unknown
 custom bridge names, and do not prove the cause of #1930.
 
+### Host/node associations retain durable operator intent
+
+Monitoring owns the provenance used to distinguish an operator selection from
+an automatic association. The host continuity journal records `manual`,
+`unlinked` or `automatic`; an absent value is legacy/unknown, not evidence that
+the link was automatic. This internal provenance does not add a public API or
+frontend field.
+
+The existing link/unlink APIs hold the host lifecycle write lock and persist
+the complete affected intent transaction under the state lock before publishing
+either direction of the association. Missing storage or a failed journal write
+must return an error without changing visible links or the journal's in-memory
+entries. Replacing a node's owner persists the displaced owner's explicit unlink
+too, including an owner which has not yet reconnected after restart. Link writes
+preserve identity, report watermarks, credential and removal evidence.
+
+A manual selection remains pinned to the selected provider ID through unmatched
+reports, provider refresh and restart before another report. A replacement
+provider ID must not inherit that selection by name: intent remains dormant
+until the original ID returns or the operator explicitly selects another ID.
+The persisted reservation prevents another host's automatic match taking that
+node before its owner reconnects. Explicit unlink survives restart and suppresses
+automatic node reassociation even when the next report supplies matching evidence.
+
+Automatic name matching uses the provider's node name, never the display name
+merged from an already-linked agent. Known automatic associations may be
+re-evaluated and cleared; provenance-bearing host updates remove obsolete
+reverse links without clearing another agent's link. Unmarked legacy forward
+links are retained during report ingestion, and unmarked one-way reverse links
+must survive host updates because SMART fallback also consumes them. Legacy
+provider reconciliation otherwise keeps its existing behaviour. This deliberately
+does not claim to repair every persisted v6.4.1 association or resolve #1930.
+
+Verification:
+- `internal/monitoring/host_manual_link_regression_test.go` covers immediate
+  restart, explicit unlink with positive matching evidence, provider-ID
+  replacement/return, failed writes, automatic versus unknown provenance,
+  dormant reservations and concurrent reports/provider refresh.
+- `internal/config/host_continuity_test.go` pins journal preservation and
+  replacement of a dormant owner.
+- `internal/models/state_additional_test.go` pins transaction rollback,
+  obsolete reverse-link cleanup and retention of unknown legacy reverse links.
+- `TestMaybePollPhysicalDisksAsync_AgentFallbackWhenDiskQueryFails` in
+  `internal/monitoring/monitor_pve_disk_fallback_test.go` remains required:
+  association cleanup must not remove the one-way link used by SMART fallback.
+
+### Physical-disk observation cadence
+
 Physical disk inventory has an independent collector schedule. The PVE poller
 carries its default five-minute or configured interval with each disk record,
 while keeping the last successful observation timestamp on retained records.

internal/config/host_continuity.go

@@ -28,6 +28,7 @@ type HostContinuityEntry struct {
 	DeniedTokenIDs    []string  `json:"deniedTokenIds,omitempty"`
 	AgentVersion      string    `json:"agentVersion,omitempty"`
 	Platform          string    `json:"platform,omitempty"`
+	NodeLinkSource    string    `json:"nodeLinkSource,omitempty"`
 	LinkedNodeID      string    `json:"linkedNodeId,omitempty"`
 	LinkedVMID        string    `json:"linkedVmId,omitempty"`
 	LinkedContainerID string    `json:"linkedContainerId,omitempty"`
@@ -401,3 +402,56 @@ func uniqueTrimmedStrings(values ...string) []string {
 	}
 	return out
 }
+
+// SetNodeLinkIntents writes all affected links as one journal transaction.
+// It preserves report watermarks, credentials and removal tombstones.
+func (s *HostContinuityStore) SetNodeLinkIntents(links []HostContinuityEntry) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.loadErr != nil {
+		return s.loadErr
+	}
+	previous := make(map[string]HostContinuityEntry, len(s.entries))
+	for id, entry := range s.entries {
+		previous[id] = entry
+	}
+	for _, link := range links {
+		// An operator can replace an owner which has not reported since restart.
+		// Persist that owner's unlink too, so it cannot reclaim the selection.
+		if link.NodeLinkSource == "manual" && link.LinkedNodeID != "" {
+			for id, prior := range s.entries {
+				if id != link.HostID && prior.LinkedNodeID == link.LinkedNodeID && prior.RemovedAt.IsZero() {
+					prior.LinkedNodeID, prior.NodeLinkSource = "", "unlinked"
+					s.entries[id] = prior
+				}
+			}
+		}
+		entry, ok := s.entries[link.HostID]
+		if !ok {
+			entry = link
+		}
+		entry.LinkedNodeID = link.LinkedNodeID
+		entry.LinkedVMID, entry.LinkedContainerID = "", ""
+		entry.NodeLinkSource = link.NodeLinkSource
+		s.entries[link.HostID] = entry
+	}
+	if err := s.save(); err != nil {
+		s.entries = previous
+		return err
+	}
+	return nil
+}
+
+// NodeLinkReservedByOther protects an operator-selected node even before its
+// owner reconnects after restart. Removal releases the reservation.
+func (s *HostContinuityStore) NodeLinkReservedByOther(hostID, nodeID string) bool {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	for id, entry := range s.entries {
+		if id != hostID && entry.NodeLinkSource == "manual" &&
+			entry.LinkedNodeID == nodeID && entry.RemovedAt.IsZero() {
+			return true
+		}
+	}
+	return false
+}

internal/config/host_continuity_test.go

@@ -260,3 +260,51 @@ func TestHostContinuityStoreReportsLoadFailure(t *testing.T) {
 		t.Fatal("LoadError = nil, want continuity read failure")
 	}
 }
+
+func TestHostContinuityNodeLinkIntentsPreserveJournal(t *testing.T) {
+	dir := t.TempDir()
+	s := NewHostContinuityStore(dir, nil)
+	entry := HostContinuityEntry{HostID: "a", LinkedNodeID: "n", NodeLinkSource: "manual",
+		TokenID: "synthetic-token", DeniedTokenIDs: []string{"synthetic-denied"},
+		ReportStreamID: "stream", ReportSequence: 42, RemovedAt: time.Now().UTC()}
+	if err := s.Upsert(entry); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.SetNodeLinkIntents([]HostContinuityEntry{
+		{HostID: "a", NodeLinkSource: "unlinked"},
+		{HostID: "b", LinkedNodeID: "n", NodeLinkSource: "manual"},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	reloaded := NewHostContinuityStore(dir, nil)
+	a, _ := reloaded.Get("a")
+	b, _ := reloaded.Get("b")
+	if a.NodeLinkSource != "unlinked" || a.LinkedNodeID != "" || b.NodeLinkSource != "manual" || b.LinkedNodeID != "n" {
+		t.Fatal("replacement intent lost across reload")
+	}
+	if a.TokenID != entry.TokenID || len(a.DeniedTokenIDs) != 1 || a.ReportSequence != 42 || !a.RemovedAt.Equal(entry.RemovedAt) {
+		t.Fatal("link update changed lifecycle evidence")
+	}
+}
+
+func TestHostContinuityNodeLinkReplacesDormantOwner(t *testing.T) {
+	dir := t.TempDir()
+	s := NewHostContinuityStore(dir, nil)
+	if err := s.SetNodeLinkIntents([]HostContinuityEntry{{HostID: "a", LinkedNodeID: "n", NodeLinkSource: "manual"}}); err != nil {
+		t.Fatal(err)
+	}
+	if !s.NodeLinkReservedByOther("b", "n") || s.NodeLinkReservedByOther("a", "n") {
+		t.Fatal("reservation ownership incorrect")
+	}
+	if err := s.SetNodeLinkIntents([]HostContinuityEntry{{HostID: "b", LinkedNodeID: "n", NodeLinkSource: "manual"}}); err != nil {
+		t.Fatal(err)
+	}
+	s = NewHostContinuityStore(dir, nil)
+	a, _ := s.Get("a")
+	if a.LinkedNodeID != "" || a.NodeLinkSource != "unlinked" {
+		t.Fatal("dormant owner can reclaim replacement link")
+	}
+	if !s.NodeLinkReservedByOther("a", "n") || s.NodeLinkReservedByOther("b", "n") {
+		t.Fatal("replacement reservation incorrect")
+	}
+}

internal/models/models.go

@@ -363,6 +363,7 @@ type Host struct {
 	DiskWriteRate float64 `json:"diskWriteRate,omitempty"`
 
 	// Linking: When this host agent is running on a known PVE node/VM/container
+	NodeLinkSource    string `json:"-"`                           // manual, unlinked, automatic; empty means legacy/unknown
 	LinkedNodeID      string `json:"linkedNodeId,omitempty"`      // ID of the PVE node this agent is running on
 	LinkedVMID        string `json:"linkedVmId,omitempty"`        // ID of the VM this agent is running inside
 	LinkedContainerID string `json:"linkedContainerId,omitempty"` // ID of the container this agent is running inside
@@ -3859,6 +3860,21 @@ func preferNodeForMerge(existing Node, candidate Node) Node {
 }
 
 func reconcileHostNodeLinksLocked(hosts []Host, nodes []Node) {
+	// Explicit operator intent is pinned to the selected provider ID, never
+	// transferred by hostname to a replacement provider identity.
+	for _, host := range hosts {
+		if host.NodeLinkSource != "manual" && host.NodeLinkSource != "unlinked" {
+			continue
+		}
+		for i := range nodes {
+			if nodes[i].LinkedAgentID == host.ID {
+				nodes[i].LinkedAgentID = ""
+			}
+			if host.NodeLinkSource == "manual" && nodes[i].ID == host.LinkedNodeID {
+				nodes[i].LinkedAgentID = host.ID
+			}
+		}
+	}
 	linkedNodeByHostID := make(map[string]string)
 	multipleNodeLinksByHostID := make(map[string]struct{})
 	for _, node := range nodes {
@@ -3880,6 +3896,9 @@ func reconcileHostNodeLinksLocked(hosts []Host, nodes []Node) {
 			continue
 		}
 
+		if hosts[i].NodeLinkSource == "manual" || hosts[i].NodeLinkSource == "unlinked" {
+			continue
+		}
 		nodeID, hasLinkedNode := linkedNodeByHostID[hostID]
 		_, ambiguous := multipleNodeLinksByHostID[hostID]
 		switch {
@@ -5315,6 +5334,13 @@ func (s *State) UpsertHost(host Host) {
 	defer s.mu.Unlock()
 
 	host = cloneHost(host)
+	for i, node := range s.Nodes {
+		// Legacy one-way links have unknown intent; an unmarked host update
+		// cannot authorise their removal (SMART fallback also uses these links).
+		if host.NodeLinkSource != "" && host.ID != "" && node.LinkedAgentID == host.ID && node.ID != host.LinkedNodeID {
+			s.Nodes[i].LinkedAgentID = ""
+		}
+	}
 
 	updated := false
 	for i, existing := range s.Hosts {
@@ -6077,3 +6103,39 @@ func (s *State) UpdatePollStats(pollDuration float64, uptime int64, wsClients in
 	s.Stats.Uptime = uptime
 	s.Stats.WebSocketClients = wsClients
 }
+
+// SetHostNodeLinkIntent commits operator intent only after durable storage
+// accepts the complete set of changed hosts. persist must not call State.
+func (s *State) SetHostNodeLinkIntent(hostID, nodeID string, persist func([]Host) error) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	next := &State{Hosts: cloneHosts(s.Hosts), Nodes: append([]Node(nil), s.Nodes...)}
+	if nodeID != "" {
+		if err := next.LinkHostAgentToNode(hostID, nodeID); err != nil {
+			return err
+		}
+	} else if !next.UnlinkHostAgent(hostID) {
+		return fmt.Errorf("host not found or not linked to a node")
+	}
+	var changed []Host
+	for i := range next.Hosts {
+		h := &next.Hosts[i]
+		if h.ID == hostID {
+			h.NodeLinkSource = "manual"
+			if nodeID == "" {
+				h.NodeLinkSource = "unlinked"
+			}
+			changed = append(changed, *h)
+		} else if h.LinkedNodeID != s.Hosts[i].LinkedNodeID {
+			h.NodeLinkSource = "unlinked"
+			changed = append(changed, *h)
+		}
+	}
+	if persist != nil {
+		if err := persist(changed); err != nil {
+			return err
+		}
+	}
+	s.Hosts, s.Nodes, s.LastUpdate = next.Hosts, next.Nodes, next.LastUpdate
+	return nil
+}

internal/models/state_additional_test.go

@@ -685,3 +685,56 @@ func TestStateContainers(t *testing.T) {
 		t.Fatalf("expected false for missing container")
 	}
 }
+
+func TestSetHostNodeLinkIntentTransaction(t *testing.T) {
+	s := &State{
+		Hosts: []Host{{ID: "a", LinkedNodeID: "n", NodeLinkSource: "manual"}, {ID: "b"}},
+		Nodes: []Node{{ID: "n", LinkedAgentID: "a"}},
+	}
+	failure := errors.New("disk unavailable")
+	err := s.SetHostNodeLinkIntent("b", "n", func(changed []Host) error {
+		if len(changed) != 2 {
+			t.Fatalf("changed hosts = %d", len(changed))
+		}
+		if changed[0].NodeLinkSource != "unlinked" || changed[1].NodeLinkSource != "manual" {
+			t.Fatalf("replacement intent not explicit: %+v", changed)
+		}
+		return failure
+	})
+	if !errors.Is(err, failure) {
+		t.Fatalf("error = %v", err)
+	}
+	if s.Hosts[0].LinkedNodeID != "n" || s.Hosts[1].LinkedNodeID != "" || s.Nodes[0].LinkedAgentID != "a" {
+		t.Fatal("failed persistence changed visible state")
+	}
+	if err := s.SetHostNodeLinkIntent("b", "n", func([]Host) error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if s.Hosts[0].NodeLinkSource != "unlinked" || s.Hosts[1].NodeLinkSource != "manual" || s.Nodes[0].LinkedAgentID != "b" {
+		t.Fatal("replacement did not commit both intents")
+	}
+	called := false
+	if err := s.SetHostNodeLinkIntent("b", "missing", func([]Host) error { called = true; return nil }); err == nil || called {
+		t.Fatal("invalid target reached persistence")
+	}
+}
+
+func TestUpsertHostClearsOnlyObsoleteReverseLinks(t *testing.T) {
+	s := &State{Nodes: []Node{
+		{ID: "old", LinkedAgentID: "agent"},
+		{ID: "current", LinkedAgentID: "agent"},
+		{ID: "other", LinkedAgentID: "other-agent"},
+	}}
+	s.UpsertHost(Host{ID: "agent", LinkedNodeID: "current", NodeLinkSource: "automatic"})
+	if s.Nodes[0].LinkedAgentID != "" || s.Nodes[1].LinkedAgentID != "agent" || s.Nodes[2].LinkedAgentID != "other-agent" {
+		t.Fatalf("incorrect reverse-link cleanup: %+v", s.Nodes)
+	}
+}
+
+func TestUpsertHostPreservesUnknownLegacyReverseLink(t *testing.T) {
+	s := &State{Nodes: []Node{{ID: "legacy-node", LinkedAgentID: "legacy-agent"}}}
+	s.UpsertHost(Host{ID: "legacy-agent", Hostname: "legacy"})
+	if s.Nodes[0].LinkedAgentID != "legacy-agent" {
+		t.Fatal("unmarked host update erased unknown legacy reverse link")
+	}
+}

internal/monitoring/host_manual_link_regression_test.go

@@ -1,6 +1,8 @@
 package monitoring
 
 import (
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"os"
 	"testing"
 	"time"
 
@@ -11,6 +13,7 @@ import (
 // repairing old automatic associations must not erase a manual association.
 func TestManualHostLinkSurvivesUnmatchedReportsAndProviderRefresh(t *testing.T) {
 	monitor := issue1654Monitor()
+	monitor.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil)
 	report := issue1654Report(time.Now().UTC())
 	report.Host.Hostname = "nas.example"
 	report.Host.ReportIP = ""
@@ -52,3 +55,250 @@ func TestManualHostLinkSurvivesUnmatchedReportsAndProviderRefresh(t *testing.T)
 		t.Fatalf("explicit unlink retained node link %q", got)
 	}
 }
+
+// A different provider ID is not permission to redirect an operator's intent.
+// Keep the original selection dormant until that ID returns or explicit unlink.
+func TestManualHostLinkDurableIntent(t *testing.T) {
+	dir := t.TempDir()
+	newMonitor := func() *Monitor {
+		m := issue1654Monitor()
+		m.hostContinuityStore = config.NewHostContinuityStore(dir, nil)
+		return m
+	}
+	m := newMonitor()
+	report := issue1654Report(time.Now().UTC())
+	report.Host.Hostname, report.Host.ReportIP, report.Network = "nas.example", "", nil
+	host, err := m.ApplyHostReport(report, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	node := models.Node{ID: "selected", Name: "pve", Instance: "cluster"}
+	m.state.UpdateNodesForInstance("cluster", []models.Node{node})
+	if err := m.LinkHostAgent(host.ID, node.ID); err != nil {
+		t.Fatal(err)
+	}
+	// Restart immediately, with no intervening report to accidentally save intent.
+	m = newMonitor()
+	for i := 0; i < 3; i++ {
+		provider := node
+		if i == 1 {
+			provider.ID = "replacement"
+		}
+		m.state.UpdateNodesForInstance("cluster", []models.Node{provider})
+		report.Timestamp = report.Timestamp.Add(time.Second)
+		host, err = m.ApplyHostReport(report, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		m.state.UpdateNodesForInstance("cluster", []models.Node{provider})
+		if got := m.state.GetHosts()[0]; got.LinkedNodeID != node.ID || got.NodeLinkSource != "manual" {
+			t.Fatalf("step %d lost intent: %+v", i, got)
+		}
+		want := host.ID
+		if i == 1 {
+			want = ""
+		}
+		if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != want {
+			t.Fatalf("step %d reverse = %q, want %q", i, got, want)
+		}
+	}
+	if err := m.UnlinkHostAgent(host.ID); err != nil {
+		t.Fatal(err)
+	}
+	m = newMonitor()
+	// Positive matching evidence must not undo an explicit unlink.
+	report.Host.Hostname = node.Name
+	for i := 0; i < 2; i++ {
+		m.state.UpdateNodesForInstance("cluster", []models.Node{node})
+		report.Timestamp = report.Timestamp.Add(time.Second)
+		host, err = m.ApplyHostReport(report, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if host.LinkedNodeID != "" || host.NodeLinkSource != "unlinked" {
+			t.Fatalf("explicit unlink undone: %+v", host)
+		}
+		if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != "" {
+			t.Fatalf("reverse = %q", got)
+		}
+	}
+}
+
+func TestHostLinkProvenanceCleanupAcrossRestart(t *testing.T) {
+	for _, source := range []string{"automatic", ""} {
+		t.Run("source="+source, func(t *testing.T) {
+			dir := t.TempDir()
+			m := issue1654Monitor()
+			m.hostContinuityStore = config.NewHostContinuityStore(dir, nil)
+			report := issue1654Report(time.Now().UTC())
+			report.Host.Hostname, report.Host.ReportIP, report.Network = "nas.example", "", nil
+			host, err := m.ApplyHostReport(report, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			node := models.Node{ID: "pve", Name: "pve", Instance: "cluster", LinkedAgentID: host.ID}
+			host.LinkedNodeID, host.NodeLinkSource = node.ID, source
+			m.state.UpsertHost(host)
+			m.state.UpdateNodesForInstance("cluster", []models.Node{node})
+			entry, _ := m.hostContinuityStore.Get(host.ID)
+			entry.LinkedNodeID, entry.NodeLinkSource = node.ID, source
+			if err := m.hostContinuityStore.Upsert(entry); err != nil {
+				t.Fatal(err)
+			}
+			for i := 0; i < 2; i++ {
+				if i == 1 {
+					m = issue1654Monitor()
+					m.hostContinuityStore = config.NewHostContinuityStore(dir, nil)
+					m.state.UpdateNodesForInstance("cluster", []models.Node{node})
+				}
+				report.Timestamp = report.Timestamp.Add(time.Second)
+				host, err = m.ApplyHostReport(report, nil)
+				if err != nil {
+					t.Fatal(err)
+				}
+				want := ""
+				if source == "" {
+					want = node.ID
+				}
+				if host.LinkedNodeID != want {
+					t.Fatalf("step %d link = %q, want %q", i, host.LinkedNodeID, want)
+				}
+				reverse := ""
+				if want != "" {
+					reverse = host.ID
+				}
+				if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != reverse {
+					t.Fatalf("reverse = %q, want %q", got, reverse)
+				}
+			}
+		})
+	}
+}
+
+func TestHostLinkIntentPersistenceFailure(t *testing.T) {
+	dir := t.TempDir()
+	m := issue1654Monitor()
+	m.hostContinuityStore = config.NewHostContinuityStore(dir, nil)
+	report := issue1654Report(time.Now().UTC())
+	host, err := m.ApplyHostReport(report, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	node := models.Node{ID: "selected", Name: "pve", Instance: "cluster"}
+	m.state.UpdateNodesForInstance("cluster", []models.Node{node})
+	// Replace the directory with a file after the store has loaded successfully.
+	if err := os.RemoveAll(dir); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(dir, []byte("not a directory"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.LinkHostAgent(host.ID, node.ID); err == nil {
+		t.Fatal("link acknowledged without persistence")
+	}
+	if got := m.state.GetHosts()[0].LinkedNodeID; got != "" {
+		t.Fatalf("failed link changed state: %q", got)
+	}
+	entry, _ := m.hostContinuityStore.Get(host.ID)
+	if entry.LinkedNodeID != "" {
+		t.Fatal("failed link changed journal memory")
+	}
+	if err := os.Remove(dir); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Mkdir(dir, 0700); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.LinkHostAgent(host.ID, node.ID); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.RemoveAll(dir); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(dir, []byte("not a directory"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.UnlinkHostAgent(host.ID); err == nil {
+		t.Fatal("unlink acknowledged without persistence")
+	}
+	if got := m.state.GetHosts()[0].LinkedNodeID; got != node.ID {
+		t.Fatalf("failed unlink changed state: %q", got)
+	}
+	if got := m.state.GetSnapshot().Nodes[0].LinkedAgentID; got != host.ID {
+		t.Fatal("failed unlink changed reverse link")
+	}
+	entry, _ = m.hostContinuityStore.Get(host.ID)
+	if entry.LinkedNodeID != node.ID || entry.NodeLinkSource != "manual" {
+		t.Fatal("failed unlink changed journal memory")
+	}
+}
+
+func TestManualHostLinkConcurrentReportsAndRefresh(t *testing.T) {
+	m := issue1654Monitor()
+	m.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil)
+	report := issue1654Report(time.Now().UTC())
+	report.Host.Hostname, report.Host.ReportIP, report.Network = "nas.example", "", nil
+	host, err := m.ApplyHostReport(report, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	node := models.Node{ID: "selected", Name: "pve", Instance: "cluster"}
+	m.state.UpdateNodesForInstance("cluster", []models.Node{node})
+	if err := m.LinkHostAgent(host.ID, node.ID); err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() {
+		for i := 0; i < 20; i++ {
+			report.Timestamp = report.Timestamp.Add(time.Second)
+			if _, err := m.ApplyHostReport(report, nil); err != nil {
+				done <- err
+				return
+			}
+			m.state.UpdateNodesForInstance("cluster", []models.Node{node})
+		}
+		done <- nil
+	}()
+	for i := 0; i < 20; i++ {
+		if err := m.UnlinkHostAgent(host.ID); err != nil {
+			t.Error(err)
+			break
+		}
+		if err := m.LinkHostAgent(host.ID, node.ID); err != nil {
+			t.Error(err)
+			break
+		}
+	}
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	entry, _ := m.hostContinuityStore.Get(host.ID)
+	live := m.state.GetHosts()[0]
+	if live.NodeLinkSource != "manual" || live.LinkedNodeID != node.ID ||
+		entry.NodeLinkSource != "manual" || entry.LinkedNodeID != node.ID ||
+		m.state.GetSnapshot().Nodes[0].LinkedAgentID != host.ID {
+		t.Fatal("concurrent reports or provider refresh overwrote committed intent")
+	}
+}
+
+func TestManualHostLinkReservesNodeBeforeOwnerReconnects(t *testing.T) {
+	dir := t.TempDir()
+	m := issue1654Monitor()
+	m.hostContinuityStore = config.NewHostContinuityStore(dir, nil)
+	if err := m.hostContinuityStore.SetNodeLinkIntents([]config.HostContinuityEntry{{
+		HostID: "dormant-owner", LinkedNodeID: "selected", NodeLinkSource: "manual",
+	}}); err != nil {
+		t.Fatal(err)
+	}
+	m.hostContinuityStore = config.NewHostContinuityStore(dir, nil)
+	m.state.UpdateNodesForInstance("cluster", []models.Node{{ID: "selected", Name: "pve", Instance: "cluster"}})
+	report := issue1654Report(time.Now().UTC())
+	report.Host.Hostname = "pve"
+	host, err := m.ApplyHostReport(report, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if host.LinkedNodeID != "" || m.state.GetSnapshot().Nodes[0].LinkedAgentID != "" {
+		t.Fatal("automatic match stole operator reservation")
+	}
+}

internal/monitoring/monitor_additional_test.go

@@ -64,7 +64,7 @@ func TestMonitorGetDockerHostsNilReceiver(t *testing.T) {
 }
 
 func TestMonitorLinkHostAgent(t *testing.T) {
-	monitor := &Monitor{state: models.NewState()}
+	monitor := &Monitor{state: models.NewState(), hostContinuityStore: config.NewHostContinuityStore(t.TempDir(), nil)}
 
 	if err := monitor.LinkHostAgent("", "node-1"); err == nil {
 		t.Fatalf("expected error on empty host ID")

internal/monitoring/monitor_agents.go

@@ -974,6 +974,8 @@ func hostAgentIdentifiersMatch(left, right string) bool {
 // This is used when auto-linking can't disambiguate (e.g., multiple nodes with hostname "pve").
 // After linking, the host agent's temperature/sensor data will appear on the correct node.
 func (m *Monitor) LinkHostAgent(hostID, nodeID string) error {
+	m.hostAgentLifecycleMu.Lock()
+	defer m.hostAgentLifecycleMu.Unlock()
 	hostID = strings.TrimSpace(hostID)
 	nodeID = strings.TrimSpace(nodeID)
 	if hostID == "" {
@@ -983,7 +985,7 @@ func (m *Monitor) LinkHostAgent(hostID, nodeID string) error {
 		return fmt.Errorf("node id is required")
 	}
 
-	if err := m.state.LinkHostAgentToNode(hostID, nodeID); err != nil {
+	if err := m.state.SetHostNodeLinkIntent(hostID, nodeID, m.persistNodeLinkIntents); err != nil {
 		return fmt.Errorf("link host agent %q to node %q: %w", hostID, nodeID, err)
 	}
 
@@ -999,13 +1001,15 @@ func (m *Monitor) LinkHostAgent(hostID, nodeID string) error {
 // The agent will continue to report but will appear in the Managed Agents table
 // instead of being merged with the PVE node in the Dashboard.
 func (m *Monitor) UnlinkHostAgent(hostID string) error {
+	m.hostAgentLifecycleMu.Lock()
+	defer m.hostAgentLifecycleMu.Unlock()
 	hostID = strings.TrimSpace(hostID)
 	if hostID == "" {
 		return fmt.Errorf("host id is required")
 	}
 
-	if !m.state.UnlinkHostAgent(hostID) {
-		return fmt.Errorf("host not found or not linked to a node")
+	if err := m.state.SetHostNodeLinkIntent(hostID, "", m.persistNodeLinkIntents); err != nil {
+		return err
 	}
 
 	log.Info().
@@ -1664,6 +1668,7 @@ func (m *Monitor) persistHostContinuity(host models.Host, report agentshost.Repo
 		TokenID:              strings.TrimSpace(host.TokenID),
 		AgentVersion:         strings.TrimSpace(host.AgentVersion),
 		Platform:             strings.TrimSpace(host.Platform),
+		NodeLinkSource:       host.NodeLinkSource,
 		LinkedNodeID:         strings.TrimSpace(host.LinkedNodeID),
 		LinkedVMID:           strings.TrimSpace(host.LinkedVMID),
 		LinkedContainerID:    strings.TrimSpace(host.LinkedContainerID),
@@ -1782,6 +1787,7 @@ func hostFromContinuityEntry(entry config.HostContinuityEntry) models.Host {
 		Platform:          platformsupport.NormalizeAgentReportedPlatform(entry.Platform),
 		IsLegacy:          entry.IsLegacy,
 		IntervalSeconds:   entry.IntervalSeconds,
+		NodeLinkSource:    entry.NodeLinkSource,
 		LinkedNodeID:      strings.TrimSpace(entry.LinkedNodeID),
 		LinkedVMID:        strings.TrimSpace(entry.LinkedVMID),
 		LinkedContainerID: strings.TrimSpace(entry.LinkedContainerID),
@@ -3561,6 +3567,25 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 		report.Host.ReportIP,
 		report.Network,
 	)
+	// Unknown legacy provenance is not permission to erase an operator link.
+	// Only associations created with explicit automatic provenance are cleaned.
+	prior, exists := m.hostByID(host.ID)
+	if !exists && m.hostContinuityStore != nil {
+		if entry, ok := m.hostContinuityStore.Get(host.ID); ok {
+			prior = hostFromContinuityEntry(entry)
+		}
+	}
+	host.NodeLinkSource = "automatic"
+	if prior.NodeLinkSource == "manual" || prior.NodeLinkSource == "unlinked" ||
+		(prior.NodeLinkSource == "" && prior.LinkedNodeID != "") {
+		host.NodeLinkSource = prior.NodeLinkSource
+		linkedNodeID = prior.LinkedNodeID
+		linkedVMID, linkedContainerID = "", ""
+	}
+	if linkedNodeID != "" && host.NodeLinkSource != "manual" && m.hostContinuityStore != nil &&
+		m.hostContinuityStore.NodeLinkReservedByOther(host.ID, linkedNodeID) {
+		linkedNodeID = ""
+	}
 	if linkedNodeID != "" {
 		host.LinkedNodeID = linkedNodeID
 		log.Debug().
@@ -4578,7 +4603,8 @@ func (m *Monitor) findLinkedProxmoxEntityWithHints(
 	// Check PVE nodes first - but detect ambiguity when multiple nodes match
 	var matchingNodes []linkedEntityMatch
 	for _, node := range nodes {
-		if matchHostname(node.Name()) {
+		// Merged display names can come from the linked agent, not the provider.
+		if matchHostname(node.NodeName()) {
 			matchingNodes = append(matchingNodes, linkedEntityMatch{
 				id:       node.SourceID(),
 				instance: node.Instance(),
@@ -5144,4 +5170,19 @@ func sharedSystemAlertCorrelationForHost(host models.Host, nodes []models.Node)
 	)
 }
 
-// sortContent sorts comma-separated content values for consistent display
+// Called with the lifecycle write lock and state lock held. No report or
+// provider refresh can interleave the journal commit and visible link change.
+func (m *Monitor) persistNodeLinkIntents(hosts []models.Host) error {
+	if m.hostContinuityStore == nil {
+		return fmt.Errorf("host continuity storage unavailable")
+	}
+	entries := make([]config.HostContinuityEntry, 0, len(hosts))
+	for _, h := range hosts {
+		entries = append(entries, config.HostContinuityEntry{
+			HostID: h.ID, Hostname: h.Hostname, MachineID: h.MachineID,
+			TokenID: h.TokenID, LastSeen: h.LastSeen, LinkedNodeID: h.LinkedNodeID,
+			NodeLinkSource: h.NodeLinkSource,
+		})
+	}
+	return m.hostContinuityStore.SetNodeLinkIntents(entries)
+}

internal/monitoring/monitor_full_coverage_test.go

@@ -536,6 +536,7 @@ func TestMonitor_HostAgentManagement(t *testing.T) {
 	m.nodePendingUpdatesCache = make(map[string]pendingUpdatesCache)
 
 	// Test UnlinkHostAgent
+	m.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil)
 	err := m.UnlinkHostAgent("host1")
 	if err != nil {
 		t.Errorf("UnlinkHostAgent failed: %v", err)