Keep Proxmox guest publication owned by its live connection

Stage both enumeration routes' guest readings, snapshots, alerts, backup eligibility and History until their complete cycle is admitted. Retirement, runtime replacement and transport changes cancel obsolete reads and serialize state withdrawal against admitted effects without holding the monitor mutex through callbacks.

Retain authoritative inventory after detail-budget expiry, independent agents and accepted cache safety; cancellation is not command completion. Add deterministic late-reply and publication-order controls.

Change-source: pulse-maintainer

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

@@ -9735,3 +9735,20 @@ The accepted upstream `diagnostics_token_visibility_test.go` controls and API
 reference route checks cover caller filtering and the documented authority.
 Final-source checks must retain this upstream behavior beside the reviewed
 reporting/help repairs; source proof is not installed credential recovery.
+
+### PVE guest-cycle publication and independent agents (12 October 2026)
+
+Ending a PVE connection/runtime cancels its in-flight guest collection and rejects
+late guest inventory, diagnostics, alerts, backup-template eligibility and
+History. Collection ownership is separate from consumed guest-command ownership:
+cancellation never proves a guest command stopped or clears a shared uncertainty
+fence. Transport replacement alone does not invalidate accepted OS-command
+safety or renew an existing cache's age. Independent Pulse-agent reports and
+stored metric observations remain intact when a guest cycle is withdrawn.
+
+`TestGuestPollOwnershipRuntimeDoesNotWithdrawIndependentAgentEvidence` in
+`internal/monitoring/monitor_host_agents_test.go` checks those independent
+sources through a runtime transition. Production collection/late-reply and
+publication-order controls are in `guest_poll_ownership_test.go`; diagnostics
+and both History stores are covered in `diagnostic_snapshots_test.go`. These
+controls establish no native guest restart, thaw or prior-workload restoration.

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

@@ -7283,3 +7283,34 @@ and never queue behind a warm ingest; a cold read still builds its first registr
 consume-once replacement-store controls, refused and skipped drains, read-provider
 re-entry, and duplicate/loss assertions, alongside the original initialization,
 tenant withdrawal, saved first-alert policy and joined-shutdown controls.
+
+### Guest-cycle publication ownership (12 October 2026)
+
+The production guest cycle stages diagnostic snapshots, alert evaluations,
+backup-template eligibility and History until both VM/container collections are
+ready. It admits publication only while its registered PVE client and cycle
+remain current. Retirement, transport replacement and a new runtime cancel
+outstanding guest cycles and order their state changes against any already
+admitted publication; late inventories/details cannot recreate or erase the
+replacement's guest readings. An interrupted inventory does not grant fallback.
+One cycle per live binding owns collection until it completes, including when
+its caller is cancelled. Another connection can still advance independently.
+
+Deep-enrichment expiry is distinct from parent-cycle cancellation: a complete
+authoritative inventory still publishes within the reserved parent tail. An
+accepted collection retains existing failed-node/grace rules, empty-inventory
+handling and metric/source/age policy. Focused builder and collector entry points
+retain their existing immediate effects. No guest operation lock, consumed-read
+uncertainty, timeout, permission, TLS policy or automatic retry is relaxed.
+Client replacement is not a guest reboot and does not erase accepted cache age
+or OS-command safety state. Remote collection holds no publication lock; effects
+run without the monitor state mutex held. The guest publication gate does not
+claim ownership of node, storage, backup-job or native Docker-command collection.
+
+`guest_poll_ownership_test.go`, `diagnostic_snapshots_test.go`,
+`monitor_host_agents_test.go`, `proxmox_guest_snapshot_coherence_test.go` and
+`proxmox_large_cluster_poll_budget_test.go` cover late replies on both enumeration
+routes, cancellation/retirement/runtime/client changes, complete staged readings,
+History and breach continuity, independent agents, single-flight/recovery,
+retirement ordering and bounded-detail inventory. These are controlled source
+proofs, not native restart, independent thaw, installed or release acceptance.

internal/monitoring/diagnostic_snapshots_test.go

@@ -1,6 +1,7 @@
 package monitoring
 
 import (
+	"context"
 	"encoding/json"
 	"strings"
 	"testing"
@@ -1395,3 +1396,30 @@ func TestGuestMemorySampleCapacityContract(t *testing.T) {
 	t.Run("selection", testGuestMemorySampleCapacitySelection)
 	t.Run("cache-binding", testGuestMemorySampleCapacityCacheBinding)
 }
+
+func TestGuestPollOwnershipStagesAllReadingEffects(t *testing.T) {
+	c := &guestPollOwnershipClient{cpu: .95}
+	m := guestPollOwnershipMonitor(t, c)
+	ctx, owner := m.beginPVEGuestPoll(context.Background(), "site-a", c)
+	if owner == nil {
+		t.Fatal("registered cycle not admitted")
+	}
+	defer m.finishPVEGuestPoll("site-a", owner)
+	resources := c.resources()
+	vm, ok := m.handleClusterVMResource(ctx, "site-a", resources[0], "site-a:node:105", c, nil, nil)
+	if !ok {
+		t.Fatal("VM builder lost inventory")
+	}
+	ct, ok := m.handleClusterContainerResource(ctx, "site-a", resources[1], "site-a:node:106", c, nil, nil)
+	if !ok {
+		t.Fatal("container builder lost inventory")
+	}
+	stagePVEGuestEffect(ctx, func() { m.recordGuestMetrics([]models.VM{vm}, []models.Container{ct}, time.Time{}) })
+	if len(m.GetDiagnosticSnapshots().Guests) != 0 || len(m.metricsHistory.GetGuestMetrics(vm.ID, "cpu", time.Hour)) != 0 || len(m.state.GetSnapshot().VMs) != 0 {
+		t.Fatal("a partial builder published before the complete guest generation")
+	}
+	m.publishPVEGuests(ctx, "site-a", []models.VM{vm}, []models.Container{ct})
+	if len(m.GetDiagnosticSnapshots().Guests) != 2 || len(m.state.GetSnapshot().VMs) != 1 || len(m.metricsHistory.GetGuestMetrics(vm.ID, "cpu", time.Hour)) != 1 || len(guestHistoryStoredPoints(t, m, "vm", vm.ID, "cpu")) != 1 {
+		t.Fatal("accepted generation lost diagnostics or in-memory/persistent History")
+	}
+}

internal/monitoring/guest_poll_ownership_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -25,6 +26,7 @@ type guestPollOwnershipClient struct {
 	release chan struct{}
 	block   string
 	once    sync.Once
+	calls   atomic.Int32
 }
 
 func (c *guestPollOwnershipClient) wait(ctx context.Context, phase string) {
@@ -46,6 +48,7 @@ func (c *guestPollOwnershipClient) resources() []proxmox.ClusterResource {
 }
 
 func (c *guestPollOwnershipClient) GetClusterResources(ctx context.Context, _ string) ([]proxmox.ClusterResource, error) {
+	c.calls.Add(1)
 	if c.route == "node" {
 		return nil, fmt.Errorf("cluster/resources unavailable")
 	}
@@ -53,6 +56,93 @@ func (c *guestPollOwnershipClient) GetClusterResources(ctx context.Context, _ st
 	return c.resources(), nil
 }
 
+func TestGuestPollOwnershipSingleFlightAndOrdinaryRecovery(t *testing.T) {
+	for _, canceled := range []bool{false, true} {
+		t.Run(fmt.Sprint(canceled), func(t *testing.T) {
+			c := &guestPollOwnershipClient{cpu: .1, block: "inventory", entered: make(chan context.Context, 1), release: make(chan struct{})}
+			m := guestPollOwnershipMonitor(t, c)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			var once sync.Once
+			unblock := func() { once.Do(func() { close(c.release) }) }
+			defer unblock()
+			done := make(chan error, 1)
+			go func() { done <- runGuestOwnershipPoll(m, ctx, c) }()
+			select {
+			case <-c.entered:
+			case <-time.After(5 * time.Second):
+				t.Fatal("first inventory did not start")
+			}
+			if canceled {
+				cancel()
+			}
+			// A repeat cannot issue even an inventory GET while this binding's
+			// first cycle still owns the outstanding read.
+			if err := runGuestOwnershipPoll(m, context.Background(), c); err != nil {
+				t.Fatal(err)
+			}
+			if c.calls.Load() != 1 {
+				t.Error("overlapping/canceled cycles stacked inventory reads")
+			}
+			unblock()
+			waitGuestOwnership(t, done)
+			if err := runGuestOwnershipPoll(m, context.Background(), c); err != nil {
+				t.Fatal(err)
+			}
+			got := m.state.GetSnapshot()
+			if c.calls.Load() != 2 || len(got.VMs) != 1 || len(got.Containers) != 1 || len(m.pveGuestPolls) != 0 {
+				t.Fatal("ordinary completed-cycle recovery lost inventory or leaked ownership")
+			}
+		})
+	}
+}
+
+func TestGuestPollOwnershipRetirementWaitsForAdmittedPublication(t *testing.T) {
+	c := &guestPollOwnershipClient{}
+	m := guestPollOwnershipMonitor(t, c)
+	ctx, owner := m.beginPVEGuestPoll(context.Background(), "site-a", c)
+	if owner == nil {
+		t.Fatal("registered cycle not admitted")
+	}
+	defer m.finishPVEGuestPoll("site-a", owner)
+	entered, release := make(chan struct{}), make(chan struct{})
+	var once sync.Once
+	unblock := func() { once.Do(func() { close(release) }) }
+	defer unblock()
+	stagePVEGuestEffect(ctx, func() {
+		// Effects may read the monitor; publication must not hold m.mu.
+		if _, ok := m.getPVEClient("site-a"); !ok {
+			t.Error("publication lost its connection before its admitted effects")
+		}
+		close(entered)
+		<-release
+		m.recordGuestSnapshot("site-a", "qemu", "node", 105, GuestMemorySnapshot{Name: "accepted"})
+	})
+	done := make(chan error, 1)
+	go func() {
+		m.publishPVEGuests(ctx, "site-a", []models.VM{{ID: "site-a:node:105", Instance: "site-a", Node: "node", VMID: 105}}, nil)
+		done <- nil
+	}()
+	select {
+	case <-entered:
+	case <-time.After(5 * time.Second):
+		t.Fatal("publication effect did not start or could not re-enter monitor")
+	}
+	retired := make(chan error, 1)
+	go func() { m.retirePVEInstanceRuntime("site-a"); retired <- nil }()
+	select {
+	case <-retired:
+		t.Error("retirement completed midway through an admitted guest batch")
+	case <-time.After(30 * time.Millisecond):
+	}
+	unblock()
+	waitGuestOwnership(t, done)
+	waitGuestOwnership(t, retired)
+	if len(m.state.GetSnapshot().VMs) != 0 || len(m.GetDiagnosticSnapshots().Guests) != 0 {
+		t.Error("publication resurrected retired guest state/diagnostics")
+	}
+}
+
 func (c *guestPollOwnershipClient) GetVMs(ctx context.Context, _ string) ([]proxmox.VM, error) {
 	c.wait(ctx, "inventory")
 	var rows []proxmox.VM

internal/monitoring/guest_poll_publication.go

@@ -0,0 +1,124 @@
+package monitoring
+
+import (
+	"context"
+	"sync"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+)
+
+// A guest cycle collects outside publication locks. Its effects are staged
+// until both guest kinds are ready and the registered connection still owns
+// this cycle. Child enrichment deadlines do not invalidate an authoritative
+// inventory; cancellation of the parent cycle does.
+type pveGuestPoll struct {
+	ctx         context.Context
+	cancel      context.CancelFunc
+	stopRuntime func() bool
+	client      PVEClientInterface
+	mu          sync.Mutex
+	effects     []func()
+}
+
+type pveGuestPollContextKey struct{}
+
+func (m *Monitor) beginPVEGuestPoll(ctx context.Context, instance string, client PVEClientInterface) (context.Context, *pveGuestPoll) {
+	m.pveGuestPublicationMu.RLock()
+	defer m.pveGuestPublicationMu.RUnlock()
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if ctx.Err() != nil || client == nil || m.pveClients[instance] != client || (m.runtimeCtx != nil && m.runtimeCtx.Err() != nil) {
+		return ctx, nil
+	}
+	if previous := m.pveGuestPolls[instance]; previous != nil {
+		if previous.client == client {
+			// Cancellation is not completion. Do not stack another inventory or
+			// release consumed-command fences while its owner is still running.
+			return ctx, nil
+		}
+		previous.cancel()
+	}
+	ownedCtx, cancel := context.WithCancel(ctx)
+	owner := &pveGuestPoll{ctx: ownedCtx, cancel: cancel, client: client}
+	if m.runtimeCtx != nil {
+		owner.stopRuntime = context.AfterFunc(m.runtimeCtx, cancel)
+	}
+	if m.pveGuestPolls == nil {
+		m.pveGuestPolls = make(map[string]*pveGuestPoll)
+	}
+	m.pveGuestPolls[instance] = owner
+	return context.WithValue(ownedCtx, pveGuestPollContextKey{}, owner), owner
+}
+
+func (m *Monitor) finishPVEGuestPoll(instance string, owner *pveGuestPoll) {
+	owner.cancel()
+	if owner.stopRuntime != nil {
+		owner.stopRuntime()
+	}
+	m.mu.Lock()
+	if m.pveGuestPolls[instance] == owner {
+		delete(m.pveGuestPolls, instance)
+	}
+	m.mu.Unlock()
+}
+
+// Focused builder/collector callers retain their existing immediate effects.
+// Only the production connection cycle installs a batch, inherited by all its
+// concurrent builders. A completed partial builder is not a published cycle.
+func stagePVEGuestEffect(ctx context.Context, effect func()) {
+	if owner, _ := ctx.Value(pveGuestPollContextKey{}).(*pveGuestPoll); owner != nil {
+		owner.mu.Lock()
+		owner.effects = append(owner.effects, effect)
+		owner.mu.Unlock()
+		return
+	}
+	effect()
+}
+
+func (m *Monitor) publishPVEGuests(ctx context.Context, instance string, vms []models.VM, containers []models.Container) {
+	owner, _ := ctx.Value(pveGuestPollContextKey{}).(*pveGuestPoll)
+	if owner == nil {
+		m.state.UpdateGuestsForInstance(instance, vms, containers)
+		return
+	}
+	// Retirement/client/runtime changes take the writer side and therefore
+	// cannot clear or replace the connection midway through an admitted batch.
+	// m.mu is released before effects (including alert callbacks) run.
+	m.pveGuestPublicationMu.RLock()
+	defer m.pveGuestPublicationMu.RUnlock()
+	m.mu.RLock()
+	current := owner.ctx.Err() == nil && m.pveGuestPolls[instance] == owner && m.pveClients[instance] == owner.client && (m.runtimeCtx == nil || m.runtimeCtx.Err() == nil)
+	m.mu.RUnlock()
+	if !current {
+		return
+	}
+	m.state.UpdateGuestsForInstance(instance, vms, containers)
+	owner.mu.Lock()
+	effects := owner.effects
+	owner.effects = nil
+	owner.mu.Unlock()
+	for _, effect := range effects {
+		effect()
+	}
+}
+
+// Called with m.mu and the publication writer held. Replacing a transport is
+// not a guest reboot: accepted cache ages and OS-command safety stay intact.
+func (m *Monitor) cancelPVEGuestPollLocked(instance string) {
+	if owner := m.pveGuestPolls[instance]; owner != nil {
+		owner.cancel()
+		delete(m.pveGuestPolls, instance)
+	}
+}
+
+func (m *Monitor) replacePVEClient(instance string, client PVEClientInterface) {
+	m.pveGuestPublicationMu.Lock()
+	defer m.pveGuestPublicationMu.Unlock()
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.cancelPVEGuestPollLocked(instance)
+	if m.pveClients == nil {
+		m.pveClients = make(map[string]PVEClientInterface)
+	}
+	m.pveClients[instance] = client
+}

internal/monitoring/monitor.go

@@ -1306,6 +1306,8 @@ type Monitor struct {
 	mockUnifiedViewFixtureAt   time.Time // fixture freshness the published view was built from
 	mockUnifiedViewValid       bool
 	pveClients                 map[string]PVEClientInterface
+	pveGuestPublicationMu      sync.RWMutex             // Orders guest publication with connection/runtime replacement and retirement.
+	pveGuestPolls              map[string]*pveGuestPoll // Guarded by m.mu; canceled cycles retain ownership until completion unless the binding ends.
 	pbsClients                 map[string]*pbs.Client
 	pmgClients                 map[string]*pmg.Client
 	availabilityStatuses       map[string]AvailabilityProbeStatus
@@ -1504,8 +1506,13 @@ type Monitor struct {
 }
 
 func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) {
+	m.pveGuestPublicationMu.Lock()
+	defer m.pveGuestPublicationMu.Unlock()
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	for name := range m.pveGuestPolls {
+		m.cancelPVEGuestPollLocked(name)
+	}
 	// A new runtime invalidates detached replication reads even when the old
 	// caller has not yet canceled its parent context.
 	for name, poll := range m.pveReplicationPolls {

internal/monitoring/monitor_client_init.go

@@ -44,7 +44,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) {
 				endpoints,
 				endpointFingerprints,
 			)
-			m.pveClients[pve.Name] = clusterClient
+			m.replacePVEClient(pve.Name, clusterClient)
 			log.Info().
 				Str("instance", pve.Name).
 				Str("cluster", pve.ClusterName).
@@ -73,7 +73,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) {
 			m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, false)
 			continue
 		}
-		m.pveClients[pve.Name] = client
+		m.replacePVEClient(pve.Name, client)
 		log.Info().Str("instance", pve.Name).Msg("PVE client created successfully")
 		// Set initial connection health to true
 		m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, true)

internal/monitoring/monitor_client_reconnect.go

@@ -116,9 +116,7 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) {
 				clientConfig.Timeout = connectionTimeout
 				clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints)
 
-				m.mu.Lock()
-				m.pveClients[pve.Name] = clusterClient
-				m.mu.Unlock()
+				m.replacePVEClient(pve.Name, clusterClient)
 				m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, true)
 
 				log.Info().
@@ -140,9 +138,7 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) {
 				continue
 			}
 
-			m.mu.Lock()
-			m.pveClients[pve.Name] = client
-			m.mu.Unlock()
+			m.replacePVEClient(pve.Name, client)
 			m.setProviderConnectionHealth(InstanceTypePVE, pve.Name, true)
 
 			log.Info().

internal/monitoring/monitor_cluster_dedupe.go

@@ -76,6 +76,8 @@ func (m *Monitor) retirePVEInstanceRuntime(instanceName string) {
 	if m == nil || instanceName == "" {
 		return
 	}
+	m.pveGuestPublicationMu.Lock()
+	defer m.pveGuestPublicationMu.Unlock()
 	m.retireGuestMemoryCache(instanceName)
 
 	if m.taskQueue != nil {
@@ -88,6 +90,7 @@ func (m *Monitor) retirePVEInstanceRuntime(instanceName string) {
 	key := schedulerKey(InstanceTypePVE, instanceName)
 
 	m.mu.Lock()
+	m.cancelPVEGuestPollLocked(instanceName)
 	if poll := m.pveReplicationPolls[instanceName]; poll != nil {
 		poll.cancel()
 		delete(m.pveReplicationPolls, instanceName)

internal/monitoring/monitor_host_agents_test.go

@@ -31,6 +31,33 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
 )
 
+func TestGuestPollOwnershipRuntimeDoesNotWithdrawIndependentAgentEvidence(t *testing.T) {
+	c := &guestPollOwnershipClient{cpu: .1}
+	m := guestPollOwnershipMonitor(t, c)
+	m.state.UpsertHost(models.Host{ID: "independent", Hostname: "agent", Status: "online", LastSeen: time.Now(), Memory: models.Memory{Total: 1000, Used: 300, Free: 700, Usage: 30}})
+	hosts := m.state.GetSnapshot().Hosts
+	key := guestMetadataCacheKey("site-a", "node", 105)
+	metadata := guestMetadataCacheEntry{osName: "OpenBSD", osInfoFailureCount: guestAgentOSInfoFailureThreshold, osInfoSkip: true, fetchedAt: time.Now().Add(-time.Minute)}
+	m.guestMetadataCache[key] = metadata
+	memKey := guestMemoryCacheKey("site-a", "node", 105)
+	memory := agentMemCacheEntry{info: memoryLifecycleSample(4096), fetchedAt: time.Now().Add(-time.Minute)}
+	m.vmAgentMemCache = map[string]agentMemCacheEntry{memKey: memory}
+	ctx, owner := m.beginPVEGuestPoll(context.Background(), "site-a", c)
+	if owner == nil {
+		t.Fatal("registered cycle not admitted")
+	}
+	defer m.finishPVEGuestPoll("site-a", owner)
+	stagePVEGuestEffect(ctx, func() { m.recordGuestSnapshot("site-a", "qemu", "node", 105, GuestMemorySnapshot{Name: "obsolete"}) })
+	m.setRuntimeContext(context.Background(), nil)
+	m.publishPVEGuests(ctx, "site-a", nil, nil)
+	if owner.ctx.Err() == nil || len(m.GetDiagnosticSnapshots().Guests) != 0 || !reflect.DeepEqual(m.state.GetSnapshot().Hosts, hosts) {
+		t.Fatal("runtime withdrawal admitted obsolete guest evidence or withdrew independent agent memory")
+	}
+	if !reflect.DeepEqual(m.guestMetadataCache[key], metadata) || m.vmAgentMemCache[memKey] != memory {
+		t.Fatal("runtime withdrawal erased OS-command safety or renewed accepted cache age")
+	}
+}
+
 func TestNewMonitorRoutesStartupCustomSensorWarningBeforeStart(t *testing.T) {
 	monitor, err := New(&config.Config{DataPath: t.TempDir()})
 	if err != nil {

internal/monitoring/monitor_polling_containers.go

@@ -298,24 +298,25 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 				}
 
 				// Trigger guest metadata migration if old format exists
-				if m.guestMetadataStore != nil {
-					m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, int(container.VMID))
-				}
+				stagePVEGuestEffect(ctx, func() {
+					if m.guestMetadataStore != nil {
+						m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, int(container.VMID))
+					}
 
-				m.recordGuestSnapshot(instanceName, modelContainer.Type, n.Node, int(container.VMID), GuestMemorySnapshot{
-					Name:           modelContainer.Name,
-					Status:         modelContainer.Status,
-					RetrievedAt:    modelContainer.LastSeen,
-					MemorySource:   memorySource,
-					FallbackReason: guestMemoryFallbackReason(memorySource),
-					Memory:         modelContainer.Memory,
-					Raw:            guestRaw,
+					m.recordGuestSnapshot(instanceName, modelContainer.Type, n.Node, int(container.VMID), GuestMemorySnapshot{
+						Name:           modelContainer.Name,
+						Status:         modelContainer.Status,
+						RetrievedAt:    modelContainer.LastSeen,
+						MemorySource:   memorySource,
+						FallbackReason: guestMemoryFallbackReason(memorySource),
+						Memory:         modelContainer.Memory,
+						Raw:            guestRaw,
+					})
+
+					// Check alerts
+					m.alertManager.CheckGuest(modelContainer, instanceName)
 				})
-
 				nodeContainers = append(nodeContainers, modelContainer)
-
-				// Check alerts
-				m.alertManager.CheckGuest(modelContainer, instanceName)
 			}
 
 			nodeDuration := time.Since(nodeStart)
@@ -355,7 +356,7 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 		}
 	}
 	if failedNodes == 0 && successfulNodes > 0 {
-		m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects)
+		stagePVEGuestEffect(ctx, func() { m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects) })
 	}
 
 	preservedContainers := 0
@@ -382,34 +383,36 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 	// Record guest metrics history for running containers (enables sparkline/trends view)
 	if !shouldSkipNativeMockStateMetricWrites() {
 		now := time.Now()
-		for _, ct := range allContainers {
-			if !guestObservedInCycle(ct.LastSeen, startTime) {
-				continue
+		stagePVEGuestEffect(ctx, func() {
+			for _, ct := range allContainers {
+				if !guestObservedInCycle(ct.LastSeen, startTime) {
+					continue
+				}
+				if ct.Status != "running" {
+					continue
+				}
+				diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(
+					ct.DiskRead,
+					ct.DiskWrite,
+					ct.NetworkIn,
+					ct.NetworkOut,
+					ct.IORateValidity,
+				)
+				m.recordGuestMetric(
+					"container",
+					ct.ID,
+					unifiedresources.ProxmoxGuestCPUPercent(ct.CPU),
+					historyMemoryUsage(ct.Memory),
+					historyMemoryUsed(ct.Memory),
+					ct.Disk.Usage,
+					diskRead,
+					diskWrite,
+					networkIn,
+					networkOut,
+					now,
+				)
 			}
-			if ct.Status != "running" {
-				continue
-			}
-			diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(
-				ct.DiskRead,
-				ct.DiskWrite,
-				ct.NetworkIn,
-				ct.NetworkOut,
-				ct.IORateValidity,
-			)
-			m.recordGuestMetric(
-				"container",
-				ct.ID,
-				unifiedresources.ProxmoxGuestCPUPercent(ct.CPU),
-				historyMemoryUsage(ct.Memory),
-				historyMemoryUsed(ct.Memory),
-				ct.Disk.Usage,
-				diskRead,
-				diskWrite,
-				networkIn,
-				networkOut,
-				now,
-			)
-		}
+		})
 	}
 
 	duration := time.Since(startTime)

internal/monitoring/monitor_polling_vm.go

@@ -100,7 +100,7 @@ func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string,
 		}
 	}
 	if failedNodes == 0 && successfulNodes > 0 {
-		m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects)
+		stagePVEGuestEffect(ctx, func() { m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects) })
 	}
 
 	preservedVMs := 0
@@ -122,34 +122,36 @@ func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string,
 
 	if !shouldSkipNativeMockStateMetricWrites() {
 		now := time.Now()
-		for _, vm := range allVMs {
-			if !guestObservedInCycle(vm.LastSeen, startTime) {
-				continue
+		stagePVEGuestEffect(ctx, func() {
+			for _, vm := range allVMs {
+				if !guestObservedInCycle(vm.LastSeen, startTime) {
+					continue
+				}
+				if vm.Status != "running" {
+					continue
+				}
+				diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(
+					vm.DiskRead,
+					vm.DiskWrite,
+					vm.NetworkIn,
+					vm.NetworkOut,
+					vm.IORateValidity,
+				)
+				m.recordGuestMetric(
+					"vm",
+					vm.ID,
+					unifiedresources.ProxmoxGuestCPUPercent(vm.CPU),
+					historyMemoryUsage(vm.Memory),
+					historyMemoryUsed(vm.Memory),
+					vm.Disk.Usage,
+					diskRead,
+					diskWrite,
+					networkIn,
+					networkOut,
+					now,
+				)
 			}
-			if vm.Status != "running" {
-				continue
-			}
-			diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(
-				vm.DiskRead,
-				vm.DiskWrite,
-				vm.NetworkIn,
-				vm.NetworkOut,
-				vm.IORateValidity,
-			)
-			m.recordGuestMetric(
-				"vm",
-				vm.ID,
-				unifiedresources.ProxmoxGuestCPUPercent(vm.CPU),
-				historyMemoryUsage(vm.Memory),
-				historyMemoryUsed(vm.Memory),
-				vm.Disk.Usage,
-				diskRead,
-				diskWrite,
-				networkIn,
-				networkOut,
-				now,
-			)
-		}
+		})
 	}
 
 	duration := time.Since(startTime)

internal/monitoring/monitor_pve.go

@@ -441,8 +441,10 @@ func (m *Monitor) retryPVEPortFallback(ctx context.Context, instanceName string,
 	var pbsInstances []config.PBSInstance
 	var pmgInstances []config.PMGInstance
 
+	m.pveGuestPublicationMu.Lock()
 	m.mu.Lock()
 	instanceCfg.Host = persistHost
+	m.cancelPVEGuestPollLocked(instanceName)
 	m.pveClients[instanceName] = fallbackClient
 
 	// Update in-memory config so subsequent polls build clients against the working port.
@@ -460,6 +462,7 @@ func (m *Monitor) retryPVEPortFallback(ctx context.Context, instanceName string,
 	}
 	persistence = m.persistence
 	m.mu.Unlock()
+	m.pveGuestPublicationMu.Unlock()
 
 	// Persist to disk so restarts keep the working endpoint.
 	if persistence != nil {
@@ -817,6 +820,11 @@ func (m *Monitor) pollGuestsWithFallback(
 		return ctx.Err()
 	default:
 	}
+	ctx, owner := m.beginPVEGuestPoll(ctx, instanceName, client)
+	if owner == nil {
+		return nil // Another cycle or a replaced/ended binding owns guest work.
+	}
+	defer m.finishPVEGuestPoll(instanceName, owner)
 
 	// Always try the efficient cluster/resources endpoint first
 	// This endpoint works on both clustered and standalone nodes
@@ -824,6 +832,9 @@ func (m *Monitor) pollGuestsWithFallback(
 	useClusterEndpoint := m.pollVMsAndContainersEfficient(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodeEffectiveStatus)
 
 	if !useClusterEndpoint {
+		if ctx.Err() != nil {
+			return ctx.Err() // An interrupted inventory is not a fallback grant.
+		}
 		// Fall back to traditional polling only if cluster/resources not available
 		// This should be rare - only for very old Proxmox versions
 		log.Debug().
@@ -851,7 +862,7 @@ func (m *Monitor) pollGuestsWithFallback(
 		if instanceCfg.MonitorContainers {
 			containers = m.collectContainersWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus)
 		}
-		m.state.UpdateGuestsForInstance(instanceName, vms, containers)
+		m.publishPVEGuests(ctx, instanceName, vms, containers)
 	}
 
 	return nil

internal/monitoring/monitor_pve_cluster.go

@@ -682,7 +682,8 @@ func clusterEndpointIdentityChanged(existing, refreshed []config.ClusterEndpoint
 
 // rebuildPVEClusterClient swaps the failover client for a cluster instance so
 // it picks up the refreshed endpoint list. The in-flight poll keeps using the
-// old client; the next poll cycle gets the new one.
+// old client for node work; obsolete guest collections cannot publish. The
+// next poll cycle uses the refreshed transport for guest collection too.
 func (m *Monitor) rebuildPVEClusterClient(instanceName string) {
 	if m.config == nil || m.pveClients == nil {
 		return
@@ -704,9 +705,7 @@ func (m *Monitor) rebuildPVEClusterClient(instanceName string) {
 	clientConfig.Timeout = m.config.ConnectionTimeout
 	clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints)
 
-	m.mu.Lock()
-	m.pveClients[instanceName] = clusterClient
-	m.mu.Unlock()
+	m.replacePVEClient(instanceName, clusterClient)
 
 	log.Info().
 		Str("instance", instanceName).

internal/monitoring/monitor_pve_guest_poll.go

@@ -53,7 +53,7 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam
 		log.Debug().Err(err).Str("instance", instanceName).Msg("cluster/resources not available, falling back to traditional polling")
 		return false
 	}
-	m.updatePVEBackupTemplateSubjectsFromClusterResources(instanceName, resources)
+	stagePVEGuestEffect(ctx, func() { m.updatePVEBackupTemplateSubjectsFromClusterResources(instanceName, resources) })
 	enrichmentCtx, cancelEnrichment := pveGuestEnrichmentContext(ctx)
 	defer cancelEnrichment()
 
@@ -81,9 +81,8 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam
 	// Publish the complete guest generation only after both VM and container
 	// collection/enrichment has finished. Empty authoritative results still
 	// remove genuinely deleted guests.
-	m.state.UpdateGuestsForInstance(instanceName, allVMs, allContainers)
-
-	m.recordGuestMetrics(allVMs, allContainers, cycleStart)
+	stagePVEGuestEffect(ctx, func() { m.recordGuestMetrics(allVMs, allContainers, cycleStart) })
+	m.publishPVEGuests(ctx, instanceName, allVMs, allContainers)
 
 	log.Debug().
 		Str("instance", instanceName).
@@ -387,23 +386,25 @@ func (m *Monitor) handleClusterVMResource(
 	}
 
 	// Trigger guest metadata migration if old format exists
-	if m.guestMetadataStore != nil {
-		m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID)
-	}
+	stagePVEGuestEffect(ctx, func() {
+		if m.guestMetadataStore != nil {
+			m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID)
+		}
 
-	m.recordGuestSnapshot(instanceName, vm.Type, res.Node, res.VMID, GuestMemorySnapshot{
-		Name:           vm.Name,
-		Status:         vm.Status,
-		RetrievedAt:    sampleTime,
-		MemorySource:   memorySource,
-		FallbackReason: guestMemoryFallbackReason(memorySource),
-		Memory:         vm.Memory,
-		Raw:            guestRaw,
-		Notes:          snapshotNotes,
+		m.recordGuestSnapshot(instanceName, vm.Type, res.Node, res.VMID, GuestMemorySnapshot{
+			Name:           vm.Name,
+			Status:         vm.Status,
+			RetrievedAt:    sampleTime,
+			MemorySource:   memorySource,
+			FallbackReason: guestMemoryFallbackReason(memorySource),
+			Memory:         vm.Memory,
+			Raw:            guestRaw,
+			Notes:          snapshotNotes,
+		})
+
+		m.checkGuestAlertsForVM(instanceName, vm)
 	})
 
-	m.checkGuestAlertsForVM(instanceName, vm)
-
 	return vm, true
 }
 
@@ -422,22 +423,24 @@ func (m *Monitor) handleClusterContainerResource(
 	}
 
 	// Trigger guest metadata migration if old format exists
-	if m.guestMetadataStore != nil {
-		m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID)
-	}
+	stagePVEGuestEffect(ctx, func() {
+		if m.guestMetadataStore != nil {
+			m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, res.Node, res.VMID)
+		}
 
-	m.recordGuestSnapshot(instanceName, container.Type, res.Node, res.VMID, GuestMemorySnapshot{
-		Name:           container.Name,
-		Status:         container.Status,
-		RetrievedAt:    sampleTime,
-		MemorySource:   memorySource,
-		FallbackReason: guestMemoryFallbackReason(memorySource),
-		Memory:         container.Memory,
-		Raw:            guestRaw,
+		m.recordGuestSnapshot(instanceName, container.Type, res.Node, res.VMID, GuestMemorySnapshot{
+			Name:           container.Name,
+			Status:         container.Status,
+			RetrievedAt:    sampleTime,
+			MemorySource:   memorySource,
+			FallbackReason: guestMemoryFallbackReason(memorySource),
+			Memory:         container.Memory,
+			Raw:            guestRaw,
+		})
+
+		m.alertManager.CheckGuest(container, instanceName)
 	})
 
-	m.alertManager.CheckGuest(container, instanceName)
-
 	return container, true
 }
 

internal/monitoring/proxmox_guest_snapshot_coherence_test.go

@@ -69,6 +69,7 @@ func TestPollGuestsWithFallbackRetainsOnlyFailedNodeGeneration(t *testing.T) {
 	}
 	nodeStatus := map[string]string{"node-a": "online", "node-b": "online"}
 	cfg := &config.PVEInstance{MonitorVMs: true, MonitorContainers: true}
+	monitor.replacePVEClient("lab", client)
 
 	if err := monitor.pollGuestsWithFallback(context.Background(), "lab", cfg, client, nodes, nodeStatus); err != nil {
 		t.Fatalf("partial poll failed: %v", err)

internal/monitoring/proxmox_large_cluster_poll_budget_test.go

@@ -104,7 +104,10 @@ func TestLargePVEClusterPublishesCompleteGenerationWhenEnrichmentBudgetExpires(t
 		stubPVEClient: &stubPVEClient{},
 		resources:     largeClusterResources(),
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
+	monitor.replacePVEClient("large-cluster", client)
+	// Expire only deep enrichment. The parent still owns the authoritative
+	// inventory and its reserved publication tail, unlike explicit cancellation.
+	ctx, cancel := context.WithTimeout(context.Background(), pvePollTailReserve+40*time.Millisecond)
 	defer cancel()
 
 	started := time.Now()