fix(monitoring): drop replayed unified metric samples

Read-side registry rebuilds re-run the unified metric sync as often as every two seconds, so one poll observation was re-written many times between polls. The metrics store upserts on (resource, metric, tier, timestamp), so each replay UPDATEs the same row and commits a fresh transaction for no new data, churning the SQLite WAL and driving the disk-write amplification reported in #1966.

Track the last sample handed to the store per series and skip exact timestamp+value repeats before the batch is enqueued. A corrected value at the same observation time is still written, so the guard cannot mask a real change. Record the behaviour in the monitoring and agent-lifecycle contracts and move the focused regression into the accepted monitor.go proof file.

Change-source: pulse-maintainer

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

@@ -8071,3 +8071,12 @@ This does not change token scopes, agent admission, or existing agent cleanup.
 Regression coverage: `TestQuickSecuritySetupForcePreservesSystemSettings` and
 `TestInitializeSystemSettingsPreservesExistingBytes`, `TestInitializeSystemSettingsMissing`,
 `TestInitializeSystemSettingsReadError`.
+
+### 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`).

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

@@ -4065,3 +4065,18 @@ provider keys, canonical IDs, JSON projection, temperature/size/cadence, separat
 nodes/controller members, and confirmed inventory removal.
 `TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata.
 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
+`internal/monitoring/monitor_host_agents_test.go`
+(`TestDedupeUnifiedMetricWritesDropsExactReplays`).

internal/monitoring/monitor.go

@@ -1247,6 +1247,14 @@ type Monitor struct {
 	// that already hold m.mu can consult it without lock re-entrancy.
 	licenseCheckerMu sync.RWMutex
 	licenseChecker   func(feature string) bool
+
+	// Unified metric sync replay guard. The sync runs on both ingest
+	// boundaries and read-side registry rebuilds, so one observation is often
+	// replayed many times between polls. Without this every replay re-writes
+	// the same metrics row (an UPDATE in SQLite) and churns the WAL for no new
+	// data, which is the dominant write amplification in #1966.
+	unifiedMetricSyncMu   sync.Mutex
+	unifiedMetricSyncLast map[unifiedMetricSampleKey]unifiedMetricSample
 }
 
 func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) {
@@ -5171,6 +5179,63 @@ func recordSupplementalResourceChanges(store ResourceStoreInterface, changes []u
 	}
 }
 
+// unifiedMetricSampleKey identifies one persisted unified-metric series. The
+// metrics store keys rows by (resource type, resource id, metric type, tier,
+// timestamp), so the replay guard must match at least those coordinates.
+type unifiedMetricSampleKey struct {
+	resourceType string
+	resourceID   string
+	metricType   string
+}
+
+// unifiedMetricSample is the last sample handed to the metrics store for a
+// series. Timestamps are compared at the store's second resolution so a
+// sub-second replay of the same observation is still recognised as redundant.
+type unifiedMetricSample struct {
+	timestampUnix int64
+	value         float64
+}
+
+// unifiedMetricSyncMaxSeries bounds the replay guard so long-lived monitors
+// with churning resource IDs cannot pin memory. The map is reset wholesale
+// when the cap is crossed; the worst case is one extra write per series.
+const unifiedMetricSyncMaxSeries = 100000
+
+// dedupeUnifiedMetricWrites drops exact repeats of a sample already handed to
+// the metrics store. The unified sync runs on ingest boundaries and on
+// read-side registry rebuilds (which fire as often as every two seconds), so
+// the same observation is replayed many times between polls. Each replay would
+// otherwise UPDATE the same row and commit a fresh transaction, churning the
+// SQLite WAL with no new data (#1966). A changed value at the same timestamp is
+// still written so corrections are not lost.
+func (m *Monitor) dedupeUnifiedMetricWrites(writes []metrics.WriteMetric) []metrics.WriteMetric {
+	if len(writes) == 0 {
+		return writes
+	}
+
+	m.unifiedMetricSyncMu.Lock()
+	defer m.unifiedMetricSyncMu.Unlock()
+	if m.unifiedMetricSyncLast == nil || len(m.unifiedMetricSyncLast) > unifiedMetricSyncMaxSeries {
+		m.unifiedMetricSyncLast = make(map[unifiedMetricSampleKey]unifiedMetricSample, len(writes))
+	}
+
+	out := writes[:0]
+	for _, write := range writes {
+		key := unifiedMetricSampleKey{
+			resourceType: write.ResourceType,
+			resourceID:   write.ResourceID,
+			metricType:   write.MetricType,
+		}
+		sample := unifiedMetricSample{timestampUnix: write.Timestamp.Unix(), value: write.Value}
+		if previous, ok := m.unifiedMetricSyncLast[key]; ok && previous == sample {
+			continue
+		}
+		m.unifiedMetricSyncLast[key] = sample
+		out = append(out, write)
+	}
+	return out
+}
+
 func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
 	if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
 		return
@@ -5276,6 +5341,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
 			appendStoreWrite("agent", targetID, "diskwrite", metric.Value)
 		}
 	}
+	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
 	if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}
@@ -5391,6 +5457,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
 			appendStoreWrite("vm", targetID, "diskwrite", metric.Value)
 		}
 	}
+	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
 	if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}
@@ -5490,6 +5557,7 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
 			appendStoreWrite("storage", targetID, "avail", float64(free), observedAt)
 		}
 	}
+	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
 	if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}
@@ -5697,6 +5765,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
 			appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)
 		}
 	}
+	storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
 	if len(storeWrites) > 0 {
 		m.metricsStore.WriteBatchBounded(storeWrites)
 	}

internal/monitoring/monitor_host_agents_test.go

@@ -5884,3 +5884,48 @@ func TestAgentReportsPreserveExplicitDiskIncludes(t *testing.T) {
 		check(t, host.Disks)
 	})
 }
+
+// Issue #1966: read-side registry rebuilds re-run the unified metric sync as
+// often as every two seconds, so one poll observation was re-written many
+// times between polls. The store upserts on (resource, metric, tier,
+// timestamp), so each replay committed a fresh transaction for no new data and
+// churned the SQLite WAL. The replay guard drops exact timestamp+value repeats
+// before the batch is enqueued while still writing a corrected value.
+func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) {
+	monitor := &Monitor{}
+	observedAt := time.Date(2026, time.September, 19, 12, 0, 0, 0, time.UTC)
+	write := metrics.WriteMetric{
+		ResourceType: "storage",
+		ResourceID:   "pool:tank",
+		MetricType:   "usage",
+		Value:        62,
+		Timestamp:    observedAt,
+		Tier:         metrics.TierRaw,
+	}
+
+	first := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write})
+	if len(first) != 1 {
+		t.Fatalf("first write count = %d, want 1", len(first))
+	}
+	// The same observation replayed by a read-side registry rebuild must not
+	// reach the store again: that UPDATE is the write amplification in #1966.
+	replay := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write})
+	if len(replay) != 0 {
+		t.Fatalf("replayed write count = %d, want 0", len(replay))
+	}
+
+	// A corrected value at the same observation time is still written.
+	corrected := write
+	corrected.Value = 63
+	if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{corrected}); len(got) != 1 {
+		t.Fatalf("corrected write count = %d, want 1", len(got))
+	}
+
+	// A newer observation for the same series is written.
+	advanced := write
+	advanced.Value = 63
+	advanced.Timestamp = observedAt.Add(time.Minute)
+	if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{advanced}); len(got) != 1 {
+		t.Fatalf("advanced write count = %d, want 1", len(got))
+	}
+}