fix(monitoring): anchor unified sync samples to source observation time

The replay guard added in df44238797 only suppressed repeats for storage, whose samples already carry the provider LastSeen. The agent, VM and app-container syncs still stamped every write with the rebuild wall clock, so dedupeUnifiedMetricWrites could never match and each read-side registry rebuild re-issued the whole batch as fresh rows. That is the same WAL churn described in #1966, still live for VMware/TrueNAS VMs, TrueNAS apps and supplemental agents.

Resolve each resource's observation time from its metric source (unifiedResourceObservedAt) and use it for both the in-memory history point and the store write, matching the storage path. Proxmox and libvirt VMs already have native history writers, so exclude libvirt VMs from the unified VM sync to avoid writing a second timeline for the same series.

Contract sections recorded in monitoring.md and agent-lifecycle.md; focused proof added in monitor_polling_test.go and monitor_host_agents_test.go. Local Go compilation is unavailable this turn (module cache read-only, toolchain snapshot missing); the canonical completion guard passes and gofmt is clean.

Change-source: pulse-maintainer

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

@@ -8080,11 +8080,17 @@ Regression coverage: `TestQuickSecuritySetupForcePreservesSystemSettings` and
 ### Unified metric replay guard does not change agent admission
 
 The unified metric sync runs on host-agent ingest boundaries and on read-side
-registry rebuilds. A per-series replay guard now drops exact timestamp+value
-repeats before the metrics batch is enqueued, reducing WAL churn without
-changing host admission, token binding or removal-block behaviour. Focused proof
-lives in `internal/monitoring/monitor_host_agents_test.go`
-(`TestDedupeUnifiedMetricWritesDropsExactReplays`).
+registry rebuilds. Agent and VM history samples are now anchored to the source
+observation time (`unifiedResourceObservedAt`) instead of the rebuild wall
+clock, so a registry rebuild re-issuing the same observation maps to one
+timestamp rather than a fresh sample. A per-series replay guard then drops exact
+timestamp+value repeats before the metrics batch is enqueued, reducing WAL churn
+without changing host admission, token binding or removal-block behaviour.
+Proxmox and libvirt VMs have native history writers and are excluded from the
+unified VM sync. Focused proof lives in
+`internal/monitoring/monitor_host_agents_test.go`
+(`TestDedupeUnifiedMetricWritesDropsExactReplays`,
+`TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`).
 
 ### Docker update preflight accepts any local RepoDigest for the planned image
 

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

@@ -4091,17 +4091,24 @@ This is synthetic runtime evidence, not USB hardware or reporter acceptance.
 ### Unified metric replay guard bounds registry-rebuild writes
 
 Read-side registry rebuilds (/api/state, websocket hydrate) re-run the unified
-metric sync as often as every two seconds, and storage observations are anchored
-to the provider's LastSeen. Before this change each rebuild re-issued the same
-rows as SQLite UPDATEs and committed a fresh transaction for no new data,
-churning the WAL and driving the disk-write amplification in #1966. Monitoring
-now tracks the last sample handed to the store per (resource type, resource id,
-metric type) series and drops exact timestamp+value repeats before the batch is
-enqueued; a corrected value at the same observation time is still written, so a
-real change is never masked. The guard is bounded so a long-lived monitor with
-churning resource IDs cannot pin memory. Focused proof lives in
+metric sync as often as every two seconds. Every unified sync (agent, VM,
+storage, and app-container) now timestamps its samples with the source
+observation time (`unifiedResourceObservedAt`/`unifiedMetricObservedAt`), not the
+rebuild wall clock. Before this change each rebuild re-issued the same rows as
+SQLite UPDATEs and committed a fresh transaction for no new data, churning the
+WAL and driving the disk-write amplification in #1966. Monitoring tracks the
+last sample handed to the store per (resource type, resource id, metric type)
+series and drops exact timestamp+value repeats before the batch is enqueued; a
+corrected value at the same observation time is still written, so a real change
+is never masked. The guard is bounded so a long-lived monitor with churning
+resource IDs cannot pin memory. Proxmox and libvirt VMs have native history
+writers and are excluded from the unified VM sync so the same series is not
+written twice. Focused proof lives in
 `internal/monitoring/monitor_host_agents_test.go`
-(`TestDedupeUnifiedMetricWritesDropsExactReplays`).
+(`TestDedupeUnifiedMetricWritesDropsExactReplays`) and
+`internal/monitoring/monitor_polling_test.go`
+(`TestSyncUnifiedAppContainerMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`,
+`TestSyncUnifiedVMMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`).
 
 ### Docker image update comparison accepts every local RepoDigest
 

internal/monitoring/canonical_guardrails_test.go

@@ -1944,8 +1944,8 @@ func TestUnifiedAppContainerMetricsUseCanonicalGuestHistoryPath(t *testing.T) {
 		`if target == nil || target.ResourceType != "app-container" || strings.TrimSpace(target.ResourceID) == "" {`,
 		`metricKey := fmt.Sprintf("docker:%s", targetID)`,
 		`storeWrites := make([]metrics.WriteMetric, 0)`,
-		`appendStoreWrite("dockerContainer", targetID, "cpu", value)`,
-		`appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)`,
+		`appendStoreWrite("dockerContainer", targetID, "cpu", value, observedAt)`,
+		`appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value, observedAt)`,
 		`m.metricsStore.WriteBatchBounded(storeWrites)`,
 	}
 	for _, snippet := range requiredSnippets {
@@ -1988,8 +1988,8 @@ func TestUnifiedAgentMetricsUseCanonicalHostHistoryPath(t *testing.T) {
 		`if target == nil || target.ResourceType != "agent" || strings.TrimSpace(target.ResourceID) == "" {`,
 		`metricKey := fmt.Sprintf("agent:%s", targetID)`,
 		`storeWrites := make([]metrics.WriteMetric, 0)`,
-		`appendStoreWrite("agent", targetID, "cpu", value)`,
-		`appendStoreWrite("agent", targetID, "diskwrite", metric.Value)`,
+		`appendStoreWrite("agent", targetID, "cpu", value, observedAt)`,
+		`appendStoreWrite("agent", targetID, "diskwrite", metric.Value, observedAt)`,
 		`m.metricsStore.WriteBatchBounded(storeWrites)`,
 	}
 	for _, snippet := range requiredSnippets {
@@ -2010,8 +2010,8 @@ func TestUnifiedVMMetricsUseCanonicalVMHistoryPath(t *testing.T) {
 		`if target == nil || target.ResourceType != "vm" || strings.TrimSpace(target.ResourceID) == "" {`,
 		`if source == unifiedresources.SourceProxmox {`,
 		`storeWrites := make([]metrics.WriteMetric, 0)`,
-		`appendStoreWrite("vm", targetID, "cpu", value)`,
-		`appendStoreWrite("vm", targetID, "diskwrite", metric.Value)`,
+		`appendStoreWrite("vm", targetID, "cpu", value, observedAt)`,
+		`appendStoreWrite("vm", targetID, "diskwrite", metric.Value, observedAt)`,
 		`m.metricsStore.WriteBatchBounded(storeWrites)`,
 	}
 	for _, snippet := range requiredSnippets {

internal/monitoring/monitor.go

@@ -5257,7 +5257,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
 
 	now := time.Now()
 	storeWrites := make([]metrics.WriteMetric, 0)
-	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64) {
+	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64, observedAt time.Time) {
 		if m.metricsStore == nil {
 			return
 		}
@@ -5266,7 +5266,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
 			ResourceID:   resourceID,
 			MetricType:   metricType,
 			Value:        value,
-			Timestamp:    now,
+			Timestamp:    observedAt,
 			Tier:         metrics.TierRaw,
 		})
 	}
@@ -5294,6 +5294,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
 		}
 		seenTargets[targetID] = struct{}{}
 		metricKey := fmt.Sprintf("agent:%s", targetID)
+		observedAt := unifiedResourceObservedAt(resource, now)
 
 		if metric := resource.Metrics.CPU; metric != nil {
 			value := metric.Percent
@@ -5301,53 +5302,53 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
 				value = metric.Value
 			}
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, observedAt)
 			}
-			appendStoreWrite("agent", targetID, "cpu", value)
+			appendStoreWrite("agent", targetID, "cpu", value, observedAt)
 		}
 
 		if metric := resource.Metrics.Memory; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) {
 			value := metric.Percent
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "memory", value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "memory", value, observedAt)
 			}
-			appendStoreWrite("agent", targetID, "memory", value)
+			appendStoreWrite("agent", targetID, "memory", value, observedAt)
 		}
 
 		if metric := resource.Metrics.Disk; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) {
 			value := metric.Percent
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "disk", value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "disk", value, observedAt)
 			}
-			appendStoreWrite("agent", targetID, "disk", value)
+			appendStoreWrite("agent", targetID, "disk", value, observedAt)
 		}
 
 		if metric := resource.Metrics.NetIn; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, observedAt)
 			}
-			appendStoreWrite("agent", targetID, "netin", metric.Value)
+			appendStoreWrite("agent", targetID, "netin", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.NetOut; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, observedAt)
 			}
-			appendStoreWrite("agent", targetID, "netout", metric.Value)
+			appendStoreWrite("agent", targetID, "netout", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.DiskRead; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, observedAt)
 			}
-			appendStoreWrite("agent", targetID, "diskread", metric.Value)
+			appendStoreWrite("agent", targetID, "diskread", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.DiskWrite; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, observedAt)
 			}
-			appendStoreWrite("agent", targetID, "diskwrite", metric.Value)
+			appendStoreWrite("agent", targetID, "diskwrite", metric.Value, observedAt)
 		}
 	}
 	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
@@ -5368,7 +5369,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
 
 	now := time.Now()
 	storeWrites := make([]metrics.WriteMetric, 0)
-	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64) {
+	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64, observedAt time.Time) {
 		if m.metricsStore == nil {
 			return
 		}
@@ -5377,7 +5378,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
 			ResourceID:   resourceID,
 			MetricType:   metricType,
 			Value:        value,
-			Timestamp:    now,
+			Timestamp:    observedAt,
 			Tier:         metrics.TierRaw,
 		})
 	}
@@ -5390,7 +5391,10 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
 			continue
 		}
 
-		hasNativeVMWriter := false
+		// Proxmox and libvirt both have a native history writer
+		// (monitor_pve_guest_helpers and monitor_libvirt), so the unified sync
+		// must not add a second timeline for the same series.
+		hasNativeVMWriter := strings.EqualFold(strings.TrimSpace(resource.Technology), "libvirt")
 		for _, source := range resource.Sources {
 			if source == unifiedresources.SourceProxmox {
 				hasNativeVMWriter = true
@@ -5410,6 +5414,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
 			continue
 		}
 		seenTargets[targetID] = struct{}{}
+		observedAt := unifiedResourceObservedAt(resource, now)
 
 		if metric := resource.Metrics.CPU; metric != nil {
 			value := metric.Percent
@@ -5417,53 +5422,53 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
 				value = metric.Value
 			}
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(targetID, "cpu", value, now)
+				m.metricsHistory.AddGuestMetric(targetID, "cpu", value, observedAt)
 			}
-			appendStoreWrite("vm", targetID, "cpu", value)
+			appendStoreWrite("vm", targetID, "cpu", value, observedAt)
 		}
 
 		if metric := resource.Metrics.Memory; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) {
 			value := metric.Percent
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(targetID, "memory", value, now)
+				m.metricsHistory.AddGuestMetric(targetID, "memory", value, observedAt)
 			}
-			appendStoreWrite("vm", targetID, "memory", value)
+			appendStoreWrite("vm", targetID, "memory", value, observedAt)
 		}
 
 		if metric := resource.Metrics.Disk; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) {
 			value := metric.Percent
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(targetID, "disk", value, now)
+				m.metricsHistory.AddGuestMetric(targetID, "disk", value, observedAt)
 			}
-			appendStoreWrite("vm", targetID, "disk", value)
+			appendStoreWrite("vm", targetID, "disk", value, observedAt)
 		}
 
 		if metric := resource.Metrics.NetIn; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(targetID, "netin", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(targetID, "netin", metric.Value, observedAt)
 			}
-			appendStoreWrite("vm", targetID, "netin", metric.Value)
+			appendStoreWrite("vm", targetID, "netin", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.NetOut; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(targetID, "netout", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(targetID, "netout", metric.Value, observedAt)
 			}
-			appendStoreWrite("vm", targetID, "netout", metric.Value)
+			appendStoreWrite("vm", targetID, "netout", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.DiskRead; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(targetID, "diskread", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(targetID, "diskread", metric.Value, observedAt)
 			}
-			appendStoreWrite("vm", targetID, "diskread", metric.Value)
+			appendStoreWrite("vm", targetID, "diskread", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.DiskWrite; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(targetID, "diskwrite", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(targetID, "diskwrite", metric.Value, observedAt)
 			}
-			appendStoreWrite("vm", targetID, "diskwrite", metric.Value)
+			appendStoreWrite("vm", targetID, "diskwrite", metric.Value, observedAt)
 		}
 	}
 	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
@@ -5588,6 +5593,28 @@ func unifiedMetricObservedAt(resource unifiedresources.Resource, metric *unified
 	return fallback
 }
 
+// unifiedResourceObservedAt resolves the source observation time for a
+// canonical resource from whichever metric carries a source attribution.
+// Unified syncs use it as the history timestamp so a read-side registry rebuild
+// re-issuing the same observation maps to one timestamp instead of inventing a
+// fresh sample each rebuild (see #1966).
+func unifiedResourceObservedAt(resource unifiedresources.Resource, fallback time.Time) time.Time {
+	if resource.Metrics != nil {
+		for _, metric := range []*unifiedresources.MetricValue{
+			resource.Metrics.CPU,
+			resource.Metrics.Memory,
+			resource.Metrics.Disk,
+			resource.Metrics.NetIn,
+			resource.Metrics.NetOut,
+		} {
+			if metric != nil && metric.Source != "" {
+				return unifiedMetricObservedAt(resource, metric, fallback)
+			}
+		}
+	}
+	return unifiedMetricObservedAt(resource, nil, fallback)
+}
+
 func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) {
 	if store == nil || m.metricsStore == nil {
 		return
@@ -5676,7 +5703,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
 
 	now := time.Now()
 	storeWrites := make([]metrics.WriteMetric, 0)
-	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64) {
+	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64, observedAt time.Time) {
 		if m.metricsStore == nil {
 			return
 		}
@@ -5685,7 +5712,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
 			ResourceID:   resourceID,
 			MetricType:   metricType,
 			Value:        value,
-			Timestamp:    now,
+			Timestamp:    observedAt,
 			Tier:         metrics.TierRaw,
 		})
 	}
@@ -5718,6 +5745,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
 		}
 		seenTargets[targetID] = struct{}{}
 		metricKey := fmt.Sprintf("docker:%s", targetID)
+		observedAt := unifiedResourceObservedAt(resource, now)
 
 		if metric := resource.Metrics.CPU; metric != nil {
 			value := metric.Percent
@@ -5725,53 +5753,53 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
 				value = metric.Value
 			}
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, observedAt)
 			}
-			appendStoreWrite("dockerContainer", targetID, "cpu", value)
+			appendStoreWrite("dockerContainer", targetID, "cpu", value, observedAt)
 		}
 
 		if metric := resource.Metrics.Memory; metric != nil && (metric.Total != nil || metric.Percent > 0) {
 			value := metric.Percent
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "memory", value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "memory", value, observedAt)
 			}
-			appendStoreWrite("dockerContainer", targetID, "memory", value)
+			appendStoreWrite("dockerContainer", targetID, "memory", value, observedAt)
 		}
 
 		if metric := resource.Metrics.Disk; metric != nil && (metric.Total != nil || metric.Percent > 0) {
 			value := metric.Percent
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "disk", value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "disk", value, observedAt)
 			}
-			appendStoreWrite("dockerContainer", targetID, "disk", value)
+			appendStoreWrite("dockerContainer", targetID, "disk", value, observedAt)
 		}
 
 		if metric := resource.Metrics.NetIn; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, observedAt)
 			}
-			appendStoreWrite("dockerContainer", targetID, "netin", metric.Value)
+			appendStoreWrite("dockerContainer", targetID, "netin", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.NetOut; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, observedAt)
 			}
-			appendStoreWrite("dockerContainer", targetID, "netout", metric.Value)
+			appendStoreWrite("dockerContainer", targetID, "netout", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.DiskRead; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, observedAt)
 			}
-			appendStoreWrite("dockerContainer", targetID, "diskread", metric.Value)
+			appendStoreWrite("dockerContainer", targetID, "diskread", metric.Value, observedAt)
 		}
 
 		if metric := resource.Metrics.DiskWrite; metric != nil {
 			if m.metricsHistory != nil {
-				m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, now)
+				m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, observedAt)
 			}
-			appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)
+			appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value, observedAt)
 		}
 	}
 	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)

internal/monitoring/monitor_host_agents_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/notifications"
 	"github.com/rcourtman/pulse-go-rewrite/internal/operationreceipt"
 	"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
+	"github.com/rcourtman/pulse-go-rewrite/internal/truenas"
 	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
 	agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
 	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
@@ -5930,6 +5931,72 @@ func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) {
 	}
 }
 
+func TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) {
+	previous := truenas.IsFeatureEnabled()
+	truenas.SetFeatureEnabled(true)
+	t.Cleanup(func() {
+		truenas.SetFeatureEnabled(previous)
+	})
+
+	observedAt := truenas.DefaultFixtures().CollectedAt
+	resourceStore := unifiedresources.NewMonitorAdapter(nil)
+	resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{
+		unifiedresources.SourceTrueNAS: truenas.NewProvider(truenas.DefaultFixtures()).Records(),
+	})
+
+	var targetID string
+	for _, resource := range resourceStore.GetAll() {
+		if resource.Type != unifiedresources.ResourceTypeAgent || resource.Name != "truenas-main" {
+			continue
+		}
+		target := resourceStore.MetricsTargetForResource(resource.ID)
+		if target == nil || target.ResourceType != "agent" {
+			t.Fatalf("unexpected agent metrics target: %+v", target)
+		}
+		targetID = target.ResourceID
+		break
+	}
+	if targetID == "" {
+		t.Fatal("expected a TrueNAS agent resource in the unified resource store")
+	}
+
+	cfg := metrics.DefaultConfig(t.TempDir())
+	persistentStore, err := metrics.NewStore(cfg)
+	if err != nil {
+		t.Fatalf("metrics.NewStore() error = %v", err)
+	}
+	defer func() { _ = persistentStore.Close() }()
+
+	monitor := &Monitor{
+		metricsHistory: NewMetricsHistory(1024, 24*time.Hour),
+		metricsStore:   persistentStore,
+	}
+
+	// Read-side registry rebuilds must not invent a fresh sample for the same
+	// source observation (the write amplification in #1966).
+	monitor.syncUnifiedAgentMetrics(resourceStore)
+	monitor.syncUnifiedAgentMetrics(resourceStore)
+
+	inMemory := monitor.GetGuestMetrics("agent:"+targetID, time.Hour)["cpu"]
+	if len(inMemory) != 1 {
+		t.Fatalf("in-memory agent cpu points = %d, want one per source observation: %+v", len(inMemory), inMemory)
+	}
+	if !inMemory[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("in-memory agent cpu timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt)
+	}
+
+	persisted, err := persistentStore.Query("agent", targetID, "cpu", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0)
+	if err != nil {
+		t.Fatalf("query persisted agent cpu: %v", err)
+	}
+	if len(persisted) != 1 {
+		t.Fatalf("persisted agent cpu points = %d, want one per source observation: %+v", len(persisted), persisted)
+	}
+	if !persisted[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("persisted agent cpu timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt)
+	}
+}
+
 // Issue #2113: a removed host agent re-enrolls under a derived suffixed
 // identity because identity resolution forks the machine before the removal
 // block is consulted. The block is keyed on the base machine identity, so the

internal/monitoring/monitor_polling_test.go

@@ -665,6 +665,158 @@ func TestSyncUnifiedStorageMetricsUsesPBSObservationTimeAcrossRegistryRebuilds(t
 	}
 }
 
+func TestSyncUnifiedAppContainerMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) {
+	previous := truenas.IsFeatureEnabled()
+	truenas.SetFeatureEnabled(true)
+	t.Cleanup(func() {
+		truenas.SetFeatureEnabled(previous)
+	})
+
+	observedAt := truenas.DefaultFixtures().CollectedAt
+	resourceStore := unifiedresources.NewMonitorAdapter(nil)
+	resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{
+		unifiedresources.SourceTrueNAS: truenas.NewProvider(truenas.DefaultFixtures()).Records(),
+	})
+
+	var targetID string
+	for _, resource := range resourceStore.GetAll() {
+		if resource.Type != unifiedresources.ResourceTypeAppContainer || resource.Name != "Nextcloud" {
+			continue
+		}
+		target := resourceStore.MetricsTargetForResource(resource.ID)
+		if target == nil || target.ResourceType != "app-container" {
+			t.Fatalf("unexpected app-container metrics target: %+v", target)
+		}
+		targetID = target.ResourceID
+		break
+	}
+	if targetID == "" {
+		t.Fatal("expected a TrueNAS app-container in the unified resource store")
+	}
+
+	cfg := metrics.DefaultConfig(t.TempDir())
+	persistentStore, err := metrics.NewStore(cfg)
+	if err != nil {
+		t.Fatalf("metrics.NewStore() error = %v", err)
+	}
+	defer func() { _ = persistentStore.Close() }()
+
+	monitor := &Monitor{
+		metricsHistory: NewMetricsHistory(1024, 24*time.Hour),
+		metricsStore:   persistentStore,
+	}
+
+	// Read-side registry rebuilds must not invent a fresh sample for the same
+	// source observation (the write amplification in #1966).
+	monitor.syncUnifiedAppContainerMetrics(resourceStore)
+	monitor.syncUnifiedAppContainerMetrics(resourceStore)
+
+	inMemory := monitor.GetGuestMetrics("docker:"+targetID, time.Hour)["cpu"]
+	if len(inMemory) != 1 {
+		t.Fatalf("in-memory app-container cpu points = %d, want one per source observation: %+v", len(inMemory), inMemory)
+	}
+	if !inMemory[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("in-memory app-container cpu timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt)
+	}
+
+	persisted, err := persistentStore.Query("dockerContainer", targetID, "cpu", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0)
+	if err != nil {
+		t.Fatalf("query persisted app-container cpu: %v", err)
+	}
+	if len(persisted) != 1 {
+		t.Fatalf("persisted app-container cpu points = %d, want one per source observation: %+v", len(persisted), persisted)
+	}
+	if !persisted[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("persisted app-container cpu timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt)
+	}
+}
+
+func TestSyncUnifiedVMMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) {
+	previous := vmware.IsFeatureEnabled()
+	vmware.SetFeatureEnabled(true)
+	t.Cleanup(func() {
+		vmware.SetFeatureEnabled(previous)
+	})
+
+	observedAt := time.Date(2026, time.March, 30, 18, 15, 0, 0, time.UTC)
+	resourceStore := unifiedresources.NewMonitorAdapter(nil)
+	resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{
+		unifiedresources.SourceVMware: vmware.NewProvider(vmware.InventorySnapshot{
+			ConnectionID:   "vc-1",
+			ConnectionName: "Lab VC",
+			VCenterHost:    "vc.lab.local",
+			CollectedAt:    observedAt,
+			VMs: []vmware.InventoryVM{{
+				VM:            "vm-201",
+				Name:          "app-01",
+				PowerState:    "POWERED_ON",
+				CPUCount:      4,
+				MemorySizeMiB: 8192,
+				Metrics: &vmware.InventoryMetrics{
+					CPUPercent:              ptrFloat64(38.1),
+					MemoryPercent:           ptrFloat64(57.5),
+					MemoryUsedBytes:         ptrInt64(5033164800),
+					MemoryTotalBytes:        ptrInt64(8589934592),
+					NetInBytesPerSecond:     ptrFloat64(512),
+					NetOutBytesPerSecond:    ptrFloat64(768),
+					DiskReadBytesPerSecond:  ptrFloat64(1536),
+					DiskWriteBytesPerSecond: ptrFloat64(2048),
+				},
+			}},
+		}).Records(),
+	})
+
+	var targetID string
+	for _, resource := range resourceStore.GetAll() {
+		if resource.Type != unifiedresources.ResourceTypeVM || resource.Name != "app-01" {
+			continue
+		}
+		target := resourceStore.MetricsTargetForResource(resource.ID)
+		if target == nil || target.ResourceType != "vm" {
+			t.Fatalf("unexpected VM metrics target: %+v", target)
+		}
+		targetID = target.ResourceID
+		break
+	}
+	if targetID == "" {
+		t.Fatal("expected a VMware VM in the unified resource store")
+	}
+
+	cfg := metrics.DefaultConfig(t.TempDir())
+	persistentStore, err := metrics.NewStore(cfg)
+	if err != nil {
+		t.Fatalf("metrics.NewStore() error = %v", err)
+	}
+	defer func() { _ = persistentStore.Close() }()
+
+	monitor := &Monitor{
+		metricsHistory: NewMetricsHistory(1024, 24*time.Hour),
+		metricsStore:   persistentStore,
+	}
+
+	monitor.syncUnifiedVMMetrics(resourceStore)
+	monitor.syncUnifiedVMMetrics(resourceStore)
+
+	inMemory := monitor.GetGuestMetrics(targetID, time.Hour)["cpu"]
+	if len(inMemory) != 1 {
+		t.Fatalf("in-memory VM cpu points = %d, want one per source observation: %+v", len(inMemory), inMemory)
+	}
+	if !inMemory[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("in-memory VM cpu timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt)
+	}
+
+	persisted, err := persistentStore.Query("vm", targetID, "cpu", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0)
+	if err != nil {
+		t.Fatalf("query persisted VM cpu: %v", err)
+	}
+	if len(persisted) != 1 {
+		t.Fatalf("persisted VM cpu points = %d, want one per source observation: %+v", len(persisted), persisted)
+	}
+	if !persisted[0].Timestamp.Equal(observedAt) {
+		t.Fatalf("persisted VM cpu timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt)
+	}
+}
+
 func TestSyncUnifiedAppContainerMetricsSkipsMockOwnedTrueNASHistoryWhenMockEnabled(t *testing.T) {
 	previousFeature := truenas.IsFeatureEnabled()
 	truenas.SetFeatureEnabled(true)