Stop presenting a silent agent's disk readings as collected

A host agent that stops reporting keeps its last SMART rows in state.
The Proxmox physical-disk merge copied that agent's temperature and I/O
counters into the disk together with its "available" collection state,
the registry merged the agent's retained rows into the canonical disk the
same way, and every full disk poll wrote the frozen temperature to SMART
history as a new sample. A dead agent's disk temperatures therefore stayed
"collected" in the API and kept extending the history chart indefinitely.

Expiring the agent's reporting lease is now the point where those
readings stop being current. ExpireHostTelemetry keeps the SMART
temperature and I/O values as last-known evidence but marks them
unavailable ("host agent stopped reporting"), as the physical-disk
contract asks for retained evidence. Because MergeStatus lets any
available state win, a Proxmox disk row still holding the earlier copy
would mask that, both on skipped disk polls and when the PVE host is down
and never disk-polled again. diskinventory.MergeReportedStatus lets a
source's later word on its own field supersede the availability it
supplied earlier; the monitoring merge and both registry ingest orders
use it. Every SMART temperature history writer now records a
temperature only when its current state is collected, so a retained
reading (an agent past its lease, or a disk in standby) is not written as
a new sample.

Agents before 6.2 send no collection provenance, so their merged
temperature is recorded as collected from "host_agent" instead of
inheriting Proxmox's "unsupported", keeping their history intact.
Expiry is now compare-and-set on the report time the evaluation judged
stale, so a report accepted in between is not expired.

SMART attributes, wearout and health stay last-known on purpose: they
are durable health evidence, and dropping them would resolve disk-health
and low-life alerts when an agent goes quiet. Disk temperature alerts
are evaluated from the agent's own report, so they never read this
merged value. The node-temperature path (agent sensors feeding
node.Temperature.SMART, which the disk merge labels proxmox_node_smart)
is gated by the agent's sighting in #2582; a carried node reading can
still stamp a disk as collected and belongs to that path.

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

@@ -728,6 +728,12 @@ into monitoring's models. Absent evidence has to carry its declared sentinel
 rather than a zero value that reads as a real measurement: an absent
 physical-disk view projects `Wearout` as `unifiedresources.WearoutUnreported`,
 never `0`, which would announce a spent disk the agent never reported.
+When the host heartbeat lease expires, `State.ExpireHostTelemetry` keeps that
+agent's SMART temperature and I/O counters as last-known values but marks them
+`unavailable` ("host agent stopped reporting"); the agent's next accepted
+report replaces them. Expiry is compare-and-set on the report time the offline
+sweep judged stale, so a report admitted between that judgement and the expiry
+is never expired.
 
 An enabled availability target assigned to a host agent creates an
 agent-lifecycle lease for that exact target/agent pairing. First assignment

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

@@ -3684,7 +3684,21 @@ transiently `unavailable`, provider/controller `unsupported`, or unexpectedly
 `missing`. Normalization may retain the last known value when the current
 observation is not available, but it must preserve the current state and
 reason so API and UI consumers do not present retained evidence as freshly
-collected. Unified-resource physical-disk round trips must retain named
+collected. A host agent's SMART temperature and I/O counters follow its
+reporting lease: once `State.ExpireHostTelemetry` expires the agent, they stay
+as last-known values marked `unavailable` ("host agent stopped reporting"),
+and the linked Proxmox disk merge and the registry let the agent's own later
+state supersede the availability it supplied earlier
+(`diskinventory.MergeReportedStatus`), so neither a skipped disk poll nor a
+host that is never disk-polled again can carry it forward. Expiry is
+compare-and-set on the report time the evaluation judged stale, so a report
+accepted in between is never expired. Every SMART temperature history writer
+records a temperature only when its current collection state is available (or
+predates collection state), so a retained reading has to keep its non-available
+state to stay out of history; a path that relabels a carried reading as
+available, such as a node-temperature carry stamped `proxmox_node_smart`, is
+not covered by this rule.
+Unified-resource physical-disk round trips must retain named
 `StorageGroup` membership rather than degrading it to the generic `Used`
 filesystem label.
 That same host-agent temperature boundary must prefer a recent linked host-agent

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

@@ -3516,6 +3516,13 @@ the five field statuses for serial, temperature, I/O, controller, and pool.
 `unavailable`, `unsupported`, or `missing` state may retain a prior value for
 continuity, but the state and reason must survive so the consumer cannot claim
 fresh evidence or synthesize controller-level activity for one member.
+The one exception is a source withdrawing its own evidence: when the agent
+row reports a field as no longer available, that supersedes an `available`
+state the Proxmox row still carries from the same source
+(`diskinventory.MergeReportedStatus`), in either ingest order. A host agent
+past its reporting lease must not stay "collected" through the Proxmox row's
+copy of the agent's own state, including when its host is down and no disk
+poll refreshes that copy.
 
 Cross-source correlation compares normalized serial and WWN values across
 fields without truncation, allowing a PVE bare-hex array-volume serial to join

internal/models/host_transient_expiry_test.go

@@ -1,6 +1,11 @@
 package models
 
-import "testing"
+import (
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory"
+)
 
 func TestExpireHostTelemetryClearsTransientStorageOperationsOnly(t *testing.T) {
 	state := NewState()
@@ -24,7 +29,7 @@ func TestExpireHostTelemetryClearsTransientStorageOperationsOnly(t *testing.T) {
 		},
 	})
 
-	host, changed := state.ExpireHostTelemetry("storage-host")
+	host, changed := state.ExpireHostTelemetry("storage-host", time.Time{})
 	if !changed {
 		t.Fatal("expected host telemetry to change")
 	}
@@ -47,3 +52,92 @@ func TestExpireHostTelemetryClearsTransientStorageOperationsOnly(t *testing.T) {
 		t.Fatalf("RAID last-known health was lost: %+v", host.RAID[0])
 	}
 }
+
+func TestExpireHostTelemetryMarksSMARTReadingsNoLongerCollected(t *testing.T) {
+	state := NewState()
+	judged := time.Now().Add(-time.Hour)
+	state.UpsertHost(Host{
+		ID:       "agent",
+		Status:   "online",
+		LastSeen: judged,
+		Sensors: HostSensorSummary{SMART: []HostDiskSMART{
+			{
+				Device: "sda", Serial: "CURRENT", Temperature: 41, Health: "PASSED",
+				IO: &DiskIO{Device: "sda", ReadBytes: 1000},
+				Collection: &diskinventory.CollectionStatus{
+					Serial:      diskinventory.Available("smartctl"),
+					Temperature: diskinventory.Available("smartctl"),
+					IO:          diskinventory.Available("kernel_diskstats"),
+				},
+			},
+			// Agents before collection provenance report only what they collected.
+			{Device: "sdb", Serial: "LEGACY", Temperature: 38, IO: &DiskIO{Device: "sdb"}},
+			{
+				Device: "sdc", Serial: "ASLEEP", Standby: true,
+				Collection: &diskinventory.CollectionStatus{
+					Temperature: diskinventory.Unavailable("smartctl", "disk is in standby"),
+					IO:          diskinventory.Unsupported("controller", "per-member counters unavailable"),
+				},
+			},
+		}},
+	})
+
+	host, changed := state.ExpireHostTelemetry("agent", judged)
+	if !changed {
+		t.Fatal("expected host telemetry to change")
+	}
+	stopped := diskinventory.Unavailable("smartctl", "host agent stopped reporting")
+	current := host.Sensors.SMART[0]
+	if current.Temperature != 41 || current.IO == nil || current.IO.ReadBytes != 1000 {
+		t.Fatalf("last-known SMART readings were dropped: %+v", current)
+	}
+	if current.Collection.Temperature != stopped ||
+		current.Collection.IO != diskinventory.Unavailable("kernel_diskstats", "host agent stopped reporting") {
+		t.Fatalf("expired readings still claim to be collected: %+v", current.Collection)
+	}
+	if current.Collection.Serial != diskinventory.Available("smartctl") {
+		t.Fatalf("disk identity evidence must stay last-known: %+v", current.Collection.Serial)
+	}
+	legacy := host.Sensors.SMART[1]
+	if legacy.Temperature != 38 || legacy.Collection == nil ||
+		legacy.Collection.Temperature.State != diskinventory.FieldUnavailable ||
+		legacy.Collection.IO.State != diskinventory.FieldUnavailable {
+		t.Fatalf("provenance-less readings not expired: %+v", legacy)
+	}
+	asleep := host.Sensors.SMART[2].Collection
+	if asleep.Temperature != diskinventory.Unavailable("smartctl", "disk is in standby") ||
+		asleep.IO.State != diskinventory.FieldUnsupported {
+		t.Fatalf("readings that were already not collected must keep their state: %+v", asleep)
+	}
+
+	if _, changed := state.ExpireHostTelemetry("agent", judged); changed {
+		t.Fatal("re-evaluating an expired host must not report a change")
+	}
+	stored, _ := state.GetHost("agent")
+	if stored.Sensors.SMART[0].Collection.Temperature != stopped {
+		t.Fatalf("expiry was not kept in state: %+v", stored.Sensors.SMART[0].Collection)
+	}
+}
+
+// Expiry is judged on a host snapshot. A report accepted before the expiry
+// takes the state lock must not be expired on the strength of that judgement.
+func TestExpireHostTelemetryLeavesHostThatReportedSinceJudgement(t *testing.T) {
+	state := NewState()
+	judged := time.Now().Add(-time.Hour)
+	fresh := Host{
+		ID: "agent", Status: "online", LastSeen: time.Now(),
+		Sensors: HostSensorSummary{SMART: []HostDiskSMART{{
+			Device: "sda", Temperature: 41,
+			Collection: &diskinventory.CollectionStatus{Temperature: diskinventory.Available("smartctl")},
+		}}},
+	}
+	state.UpsertHost(fresh)
+
+	if host, changed := state.ExpireHostTelemetry("agent", judged); changed || host.ID != "" {
+		t.Fatalf("a host that reported after the stale judgement was expired: %+v changed=%v", host, changed)
+	}
+	stored, _ := state.GetHost("agent")
+	if stored.Status != "online" || stored.Sensors.SMART[0].Collection.Temperature.State != diskinventory.FieldAvailable {
+		t.Fatalf("fresh report was altered: %+v", stored)
+	}
+}

internal/models/models.go

@@ -5634,7 +5634,13 @@ func (s *State) SetHostStatus(hostID, status string) bool {
 // ExpireHostTelemetry marks a host offline and clears transient operation
 // claims that cannot remain authoritative after the reporting lease expires.
 // Static topology and health counters remain available as last-known context.
-func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) {
+// SMART temperature and I/O readings stay as last-known values, but their
+// collection state says they are no longer collected.
+//
+// lastSeen is the report time the caller judged stale. A host that has
+// reported since then is left untouched and nothing is returned, so a report
+// accepted between that judgement and this call is never expired.
+func (s *State) ExpireHostTelemetry(hostID string, lastSeen time.Time) (Host, bool) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -5642,6 +5648,9 @@ func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) {
 		if host.ID != hostID {
 			continue
 		}
+		if !host.LastSeen.Equal(lastSeen) {
+			return Host{}, false
+		}
 		changed := host.Status != "offline"
 		host.Status = "offline"
 		for idx := range host.RAID {
@@ -5658,6 +5667,11 @@ func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) {
 			host.Unraid.SyncProgress = 0
 			changed = true
 		}
+		for idx := range host.Sensors.SMART {
+			if expireHostSMARTReadings(&host.Sensors.SMART[idx]) {
+				changed = true
+			}
+		}
 		s.Hosts[i] = host
 		if changed {
 			s.LastUpdate = time.Now()
@@ -5667,6 +5681,40 @@ func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) {
 	return Host{}, false
 }
 
+const hostAgentStoppedReportingReason = "host agent stopped reporting"
+
+// expireHostSMARTReadings marks the temperature and I/O counters of one SMART
+// row as no longer collected, keeping their values as last-known evidence.
+// Reports from agents that predate collection provenance carry no state, so a
+// present reading is treated as collected. It reports whether anything changed.
+func expireHostSMARTReadings(disk *HostDiskSMART) bool {
+	collection := diskinventory.CloneStatus(disk.Collection)
+	if collection == nil {
+		collection = &diskinventory.CollectionStatus{}
+	}
+	temperature := expireCollectedReading(&collection.Temperature, disk.Temperature > 0)
+	io := expireCollectedReading(&collection.IO, disk.IO != nil)
+	if !temperature && !io {
+		return false
+	}
+	disk.Collection = collection
+	return true
+}
+
+func expireCollectedReading(status *diskinventory.FieldStatus, hasValue bool) bool {
+	switch status.State {
+	case diskinventory.FieldAvailable:
+	case "":
+		if !hasValue {
+			return false
+		}
+	default:
+		return false
+	}
+	*status = diskinventory.Unavailable(status.Source, hostAgentStoppedReportingReason)
+	return true
+}
+
 // TouchHost updates the last seen timestamp for a host.
 func (s *State) TouchHost(hostID string, ts time.Time) bool {
 	s.mu.Lock()

internal/monitoring/issue1595_collection_trust_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/models"
 	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
 	"github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/metrics"
 )
 
 type issue1595TopologyFixture struct {
@@ -316,3 +317,46 @@ func hasIssue1595Source(sources []unifiedresources.DataSource, want unifiedresou
 	}
 	return false
 }
+
+// An agent report's SMART row may carry a temperature its collector did not
+// collect this time (retained across a standby or failed probe). Only a
+// collected temperature becomes a history sample.
+func TestWriteHostSMARTMetricsRecordsOnlyCollectedTemperatures(t *testing.T) {
+	storeConfig := metrics.DefaultConfig(t.TempDir())
+	storeConfig.FlushInterval = time.Hour
+	store, err := metrics.NewStore(storeConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = store.Close() })
+	m := &Monitor{metricsStore: store}
+
+	host := models.Host{ID: "agent-1", Hostname: "node1", Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{
+		{Device: "sda", Serial: "COLLECTED1", Temperature: 41,
+			Collection: &diskinventory.CollectionStatus{Temperature: diskinventory.Available("smartctl")}},
+		{Device: "sdb", Serial: "LEGACY1", Temperature: 38},
+		{Device: "sdc", Serial: "RETAINED1", Temperature: 36,
+			Collection: &diskinventory.CollectionStatus{Temperature: diskinventory.Unavailable("smartctl", "disk is in standby")}},
+	}}}
+	now := time.Now()
+	m.writeHostSMARTMetrics(host, now)
+	store.Flush()
+
+	for _, tc := range []struct {
+		disk models.HostDiskSMART
+		want int
+	}{
+		{host.Sensors.SMART[0], 1},
+		{host.Sensors.SMART[1], 1},
+		{host.Sensors.SMART[2], 0},
+	} {
+		id := unifiedresources.HostSMARTDiskSourceID(host, tc.disk)
+		points, err := store.Query("disk", id, "smart_temp", now.Add(-time.Minute), now.Add(time.Minute), 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(points) != tc.want {
+			t.Fatalf("%s: temperature samples = %d, want %d", tc.disk.Serial, len(points), tc.want)
+		}
+	}
+}

internal/monitoring/monitor.go

@@ -648,7 +648,10 @@ func mergeHostAgentSMARTIntoDisks(disks []models.PhysicalDisk, nodes []models.No
 			ioCopy := *matched.IO
 			updated[i].IO = &ioCopy
 		}
-		updated[i].Collection = diskinventory.MergeStatus(updated[i].Collection, matched.Collection)
+		// The agent's later word on a field it supplied (its lease expired, or
+		// the disk went into standby) must win over the copy a skipped disk
+		// poll carries forward from the previous merge.
+		updated[i].Collection = diskinventory.MergeReportedStatus(updated[i].Collection, hostAgentSMARTCollection(*matched))
 
 		if shouldUseHostAgentPhysicalDiskHealth(updated[i].Health, matched.Health) {
 			updated[i].Health = matched.Health
@@ -658,6 +661,28 @@ func mergeHostAgentSMARTIntoDisks(disks []models.PhysicalDisk, nodes []models.No
 	return updated
 }
 
+// hostAgentLegacySource is the provenance recorded for a temperature from an
+// agent that predates collection provenance (before 6.2). Those agents only
+// send readings they collected.
+const hostAgentLegacySource = "host_agent"
+
+// hostAgentSMARTCollection returns the agent's collection state for one SMART
+// row, giving a provenance-less temperature the legacy agent source so that
+// history and later lease expiry treat it like any other agent reading.
+func hostAgentSMARTCollection(smart models.HostDiskSMART) *diskinventory.CollectionStatus {
+	collection := diskinventory.CloneStatus(smart.Collection)
+	if collection == nil {
+		collection = &diskinventory.CollectionStatus{}
+	}
+	switch {
+	case collection.Temperature.State == "" && smart.Temperature > 0 && !smart.Standby:
+		collection.Temperature = diskinventory.Available(hostAgentLegacySource)
+	case collection.Temperature.State != "" && strings.TrimSpace(collection.Temperature.Source) == "":
+		collection.Temperature.Source = hostAgentLegacySource
+	}
+	return collection
+}
+
 func shouldUseHostAgentPhysicalDiskHealth(existing, incoming string) bool {
 	incoming = strings.ToUpper(strings.TrimSpace(incoming))
 	if incoming == "" || incoming == "UNKNOWN" {
@@ -914,8 +939,7 @@ func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) {
 		return
 	}
 
-	// Temperature (always write if > 0)
-	if disk.Temperature > 0 && m.metricsHistory != nil {
+	if diskTemperatureCollected(disk.Temperature, disk.Collection) && m.metricsHistory != nil {
 		m.metricsHistory.AddDiskMetric(resourceID, "smart_temp", float64(disk.Temperature), now)
 	}
 
@@ -929,6 +953,22 @@ func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) {
 	}
 }
 
+// diskTemperatureCollected reports whether a disk temperature was collected by
+// its current observation. Normalization may carry a last-known temperature
+// when it was not (standby, an agent past its reporting lease); that value must
+// not be recorded as a new history sample. A temperature without collection
+// state predates the contract and keeps the old behavior.
+func diskTemperatureCollected(temperature int, collection *diskinventory.CollectionStatus) bool {
+	if temperature <= 0 {
+		return false
+	}
+	if collection == nil {
+		return true
+	}
+	state := collection.Temperature.State
+	return state == "" || state == diskinventory.FieldAvailable
+}
+
 // smartMetricStoreWrites builds the persisted SMART writes for one physical
 // disk observation. The caller owns batching so the unified sync can anchor the
 // observation time and drop exact replays (#1966).
@@ -949,8 +989,7 @@ func (m *Monitor) smartMetricStoreWrites(disk models.PhysicalDisk, resourceID st
 		})
 	}
 
-	// Temperature (always write if > 0)
-	if disk.Temperature > 0 {
+	if diskTemperatureCollected(disk.Temperature, disk.Collection) {
 		appendWrite("smart_temp", float64(disk.Temperature))
 	}
 
@@ -5984,6 +6023,7 @@ func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) {
 			RPM:             resource.PhysicalDisk.RPM,
 			Used:            resource.PhysicalDisk.Used,
 			SmartAttributes: smartAttributesFromUnifiedMeta(resource.PhysicalDisk.SMART),
+			Collection:      diskinventory.CloneStatus(resource.PhysicalDisk.Collection),
 			LastChecked:     resource.LastSeen,
 		}
 		if disk.Serial == "" {

internal/monitoring/monitor_agents.go

@@ -4086,7 +4086,7 @@ func (m *Monitor) writeHostSMARTMetrics(host models.Host, now time.Time) {
 			continue
 		}
 
-		if disk.Temperature > 0 {
+		if diskTemperatureCollected(disk.Temperature, disk.Collection) {
 			m.metricsStore.Write("disk", resourceID, "smart_temp", float64(disk.Temperature), now)
 		}
 
@@ -5090,7 +5090,7 @@ func (m *Monitor) evaluateHostAgents(now time.Time) {
 					Bool("lastSeenZero", host.LastSeen.IsZero()).
 					Msg("Host agent appears offline")
 			}
-			if expiredHost, changed := m.state.ExpireHostTelemetry(host.ID); expiredHost.ID != "" {
+			if expiredHost, changed := m.state.ExpireHostTelemetry(host.ID, host.LastSeen); expiredHost.ID != "" {
 				hostCopy = expiredHost
 				resourceRefreshNeeded = resourceRefreshNeeded || changed
 			} else {

internal/monitoring/physical_disk_roundtrip_test.go

@@ -11,6 +11,9 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/models"
 	"github.com/rcourtman/pulse-go-rewrite/internal/notifications"
 	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/metrics"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
 )
 
 // Monitor construction restores notification destinations before admitting
@@ -326,3 +329,228 @@ func TestMergeHostAgentSMARTIntoDisks_AgentWearoutFillsUnreportedPVELife(t *test
 		t.Fatalf("agent endurance did not fill unreported PVE life: wearout=%d", result[0].Wearout)
 	}
 }
+
+type silentAgentDiskPVEClient struct {
+	fakeStorageClient
+}
+
+// The Proxmox inventory carries no temperature or I/O counters; for this disk
+// both come only from the node's linked host agent.
+func (*silentAgentDiskPVEClient) GetDisks(ctx context.Context, node string) ([]proxmox.Disk, error) {
+	return []proxmox.Disk{{
+		DevPath: "/dev/sdb",
+		Model:   "WDC WD40EFRX",
+		Serial:  "WD-SILENT1",
+		Type:    "hdd",
+		Health:  "PASSED",
+		Wearout: 100,
+		Size:    4000787030016,
+	}}, nil
+}
+
+// A host agent that stops reporting keeps its last SMART rows in state. Once
+// its reporting lease expires, the linked Proxmox disk may retain the agent's
+// last temperature and I/O counters, but the disk state, the canonical
+// resource and temperature history must stop treating them as collected.
+func TestSilentLinkedAgentSMARTIsRetainedButNotPresentedAsCollected(t *testing.T) {
+	t.Setenv("PULSE_DATA_DIR", t.TempDir())
+	now := time.Now()
+	state := models.NewState()
+	state.UpdateNodesForInstance("pve1", []models.Node{{
+		ID: "pve1-node1", Name: "node1", Instance: "pve1", Status: "online",
+		LastSeen: now, LinkedAgentID: "agent-1",
+	}})
+	report := func(seen time.Time, temperature int, readBytes uint64) {
+		state.UpsertHost(models.Host{
+			ID: "agent-1", Hostname: "node1", LinkedNodeID: "pve1-node1",
+			Status: "online", IntervalSeconds: 30, LastSeen: seen,
+			Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{
+				Device: "/dev/sdb", Serial: "WD-SILENT1", Type: "sata", Health: "PASSED",
+				Temperature: temperature,
+				IO:          &models.DiskIO{Device: "sdb", ReadBytes: readBytes},
+				Collection: &diskinventory.CollectionStatus{
+					Serial:      diskinventory.Available("smartctl"),
+					Temperature: diskinventory.Available("smartctl"),
+					IO:          diskinventory.Available("kernel_diskstats"),
+				},
+			}}},
+		})
+	}
+
+	alertManager := alerts.NewManager()
+	t.Cleanup(alertManager.Stop)
+	storeConfig := metrics.DefaultConfig(t.TempDir())
+	storeConfig.FlushInterval = time.Hour
+	store, err := metrics.NewStore(storeConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = store.Close() })
+	adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil))
+	history := NewMetricsHistory(100, time.Hour)
+	m := &Monitor{
+		state: state, resourceStore: adapter, metricsHistory: history, metricsStore: store,
+		alertManager: alertManager, startTime: now.Add(-time.Hour),
+		lastPhysicalDiskPoll: make(map[string]time.Time),
+	}
+	pveNodes := []proxmox.Node{{Node: "node1", Status: "online"}}
+	nodeStatus := map[string]string{"node1": "online"}
+
+	disk := func() models.PhysicalDisk {
+		t.Helper()
+		disks := state.GetSnapshot().PhysicalDisks
+		if len(disks) != 1 {
+			t.Fatalf("physical disks = %+v, want one", disks)
+		}
+		return disks[0]
+	}
+	fullPoll := func() models.PhysicalDisk {
+		t.Helper()
+		adapter.PopulateFromSnapshot(state.GetSnapshot())
+		started := time.Now()
+		delete(m.lastPhysicalDiskPoll, "pve1")
+		m.maybePollPhysicalDisksAsync(context.Background(), "pve1", &config.PVEInstance{}, &silentAgentDiskPVEClient{}, pveNodes, nodeStatus, nil)
+		deadline := time.Now().Add(3 * time.Second)
+		for {
+			if disks := state.GetSnapshot().PhysicalDisks; len(disks) == 1 && !disks[0].LastChecked.Before(started) {
+				return disks[0]
+			}
+			if time.Now().After(deadline) {
+				t.Fatal("full physical disk poll did not land in state")
+			}
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+	skippedPoll := func() models.PhysicalDisk {
+		t.Helper()
+		adapter.PopulateFromSnapshot(state.GetSnapshot())
+		m.lastPhysicalDiskPoll["pve1"] = time.Now()
+		m.maybePollPhysicalDisksAsync(context.Background(), "pve1", &config.PVEInstance{}, nil, nil, nil, nil)
+		return disk()
+	}
+	// Chart samples, and the persisted values. The store keys samples by
+	// second, so polls run back to back in this test can share one row.
+	temperatureSamples := func(d models.PhysicalDisk) (int, []float64) {
+		t.Helper()
+		id := unifiedresources.PhysicalDiskMetricID(d)
+		chart := len(history.GetDiskMetrics(id, "smart_temp", time.Hour))
+		store.Flush()
+		points, err := store.Query("disk", id, "smart_temp", now.Add(-time.Hour), time.Now().Add(time.Hour), 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		stored := make([]float64, 0, len(points))
+		for _, point := range points {
+			stored = append(stored, point.Value)
+		}
+		return chart, stored
+	}
+	canonicalCollection := func() *diskinventory.CollectionStatus {
+		t.Helper()
+		adapter.PopulateFromSnapshot(state.GetSnapshot())
+		views := adapter.PhysicalDisks()
+		if len(views) != 1 {
+			t.Fatalf("canonical physical disks = %d, want one", len(views))
+		}
+		if views[0].Temperature() != 41 {
+			t.Fatalf("canonical disk temperature = %d, want the retained 41", views[0].Temperature())
+		}
+		return views[0].Collection()
+	}
+	assertRetainedNotCollected := func(step string, got models.PhysicalDisk) {
+		t.Helper()
+		if got.Temperature != 41 || got.IO == nil || got.IO.ReadBytes != 1000 {
+			t.Fatalf("%s: last-known temperature and I/O were not retained: temp=%d io=%+v", step, got.Temperature, got.IO)
+		}
+		if got.Collection == nil ||
+			got.Collection.Temperature.State != diskinventory.FieldUnavailable ||
+			got.Collection.IO.State != diskinventory.FieldUnavailable {
+			t.Fatalf("%s: silent agent's readings presented as collected: %+v", step, got.Collection)
+		}
+	}
+
+	report(now, 41, 1000)
+	got := fullPoll()
+	if got.Temperature != 41 || got.Collection == nil || got.Collection.Temperature.State != diskinventory.FieldAvailable {
+		t.Fatalf("reporting agent temperature not merged: temp=%d collection=%+v", got.Temperature, got.Collection)
+	}
+	if chart, stored := temperatureSamples(got); chart != 1 || len(stored) != 1 || stored[0] != 41 {
+		t.Fatalf("reporting agent temperature history: chart=%d stored=%v, want one 41 sample", chart, stored)
+	}
+
+	// The agent stops reporting; its lease expires on the next evaluation.
+	state.TouchHost("agent-1", now.Add(-hostAgentHealthWindow(30)-time.Minute))
+	m.evaluateHostAgents(now)
+
+	// No disk poll runs when the whole host is down (the PVE poll fails before
+	// reaching it), so the canonical disk must reflect the expiry on its own
+	// even though the Proxmox disk row still carries the earlier merge.
+	if collection := canonicalCollection(); collection == nil ||
+		collection.Temperature.State != diskinventory.FieldUnavailable ||
+		collection.IO.State != diskinventory.FieldUnavailable {
+		t.Fatalf("canonical disk presents the silent agent's readings as collected before any disk poll: %+v", collection)
+	}
+	assertRetainedNotCollected("skipped poll", skippedPoll())
+	got = fullPoll()
+	assertRetainedNotCollected("full poll", got)
+	if chart, stored := temperatureSamples(got); chart != 1 || len(stored) != 1 {
+		t.Fatalf("silent agent's retained temperature was recorded as new history: chart=%d stored=%v", chart, stored)
+	}
+	if collection := canonicalCollection(); collection == nil ||
+		collection.Temperature.State != diskinventory.FieldUnavailable ||
+		collection.IO.State != diskinventory.FieldUnavailable {
+		t.Fatalf("canonical disk presents the silent agent's readings as collected: %+v", collection)
+	}
+
+	// A resumed agent is collected again.
+	report(time.Now(), 39, 2000)
+	m.evaluateHostAgents(time.Now())
+	got = fullPoll()
+	if got.Temperature != 39 || got.Collection.Temperature.State != diskinventory.FieldAvailable ||
+		got.IO == nil || got.IO.ReadBytes != 2000 || got.Collection.IO.State != diskinventory.FieldAvailable {
+		t.Fatalf("resumed agent readings not collected: temp=%d io=%+v collection=%+v", got.Temperature, got.IO, got.Collection)
+	}
+	if chart, stored := temperatureSamples(got); chart != 2 || len(stored) == 0 || stored[len(stored)-1] != 39 {
+		t.Fatalf("resumed agent temperature history: chart=%d stored=%v, want a new 39 sample", chart, stored)
+	}
+}
+
+// Agents before 6.2 send SMART rows without collection provenance. Their
+// temperature is still a collected reading (history records it), and once the
+// lease expires the disk must stop presenting it as collected even when a
+// skipped disk poll carries the previous merge forward.
+func TestMergeHostAgentSMARTIntoDisks_LegacyAgentTemperatureFollowsLease(t *testing.T) {
+	proxmoxDisk := models.PhysicalDisk{
+		ID: "pve1-node1-sdb", Node: "node1", Instance: "pve1", DevPath: "/dev/sdb", Serial: "LEGACY1",
+		Collection: &diskinventory.CollectionStatus{
+			Temperature: diskinventory.Unsupported("proxmox_disks", "Proxmox disk inventory does not expose temperature"),
+		},
+	}
+	nodes := []models.Node{{Name: "node1", LinkedAgentID: "agent-1"}}
+	state := models.NewState()
+	state.UpsertHost(models.Host{ID: "agent-1", Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{
+		Device: "/dev/sdb", Serial: "LEGACY1", Temperature: 38,
+	}}}})
+
+	reporting := mergeHostAgentSMARTIntoDisks([]models.PhysicalDisk{proxmoxDisk}, nodes, state.GetHosts())[0]
+	if reporting.Temperature != 38 || reporting.Collection.Temperature != diskinventory.Available(hostAgentLegacySource) {
+		t.Fatalf("legacy agent temperature not recorded as collected: temp=%d collection=%+v", reporting.Temperature, reporting.Collection)
+	}
+	if !diskTemperatureCollected(reporting.Temperature, reporting.Collection) {
+		t.Fatal("a reporting legacy agent's temperature must reach history")
+	}
+
+	state.ExpireHostTelemetry("agent-1", time.Time{})
+	for name, disk := range map[string]models.PhysicalDisk{
+		"full poll":    proxmoxDisk,
+		"skipped poll": reporting,
+	} {
+		got := mergeHostAgentSMARTIntoDisks([]models.PhysicalDisk{disk}, nodes, state.GetHosts())[0]
+		if got.Temperature != 38 || got.Collection.Temperature.State != diskinventory.FieldUnavailable {
+			t.Fatalf("%s: silent legacy agent temperature presented as collected: temp=%d collection=%+v", name, got.Temperature, got.Collection)
+		}
+		if diskTemperatureCollected(got.Temperature, got.Collection) {
+			t.Fatalf("%s: a silent legacy agent's retained temperature must not reach history", name)
+		}
+	}
+}

internal/unifiedresources/registry.go

@@ -3719,11 +3719,19 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou
 			if previous.IO != nil {
 				existing.PhysicalDisk.IO = clonePhysicalDiskIOMeta(previous.IO)
 			}
-			existing.PhysicalDisk.Collection = diskinventory.MergeStatus(
+			// The agent's own report is the later word on fields it supplied;
+			// the Proxmox row may still carry an earlier copy of them.
+			existing.PhysicalDisk.Collection = diskinventory.MergeReportedStatus(
 				incoming.PhysicalDisk.Collection,
 				previous.Collection,
 			)
 		}
+		if source == SourceAgent && previous != nil {
+			existing.PhysicalDisk.Collection = diskinventory.MergeReportedStatus(
+				previous.Collection,
+				incoming.PhysicalDisk.Collection,
+			)
+		}
 	}
 	if existing.PhysicalDisk != nil && (mergedPhysicalDisk || len(incoming.Incidents) > 0) {
 		existing.PhysicalDisk.Risk = physicalDiskRiskFromMeta(existing.PhysicalDisk, existing.Incidents)

internal/unifiedresources/registry_merge_policy_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory"
 )
 
 func TestLinkedMergeAllowsOneSidedNodeHostLinkWhenHostnameCorroborates(t *testing.T) {
@@ -557,3 +558,79 @@ func TestProxmoxOneWayLinkRejectsHostLocalEndpointCorroboration(t *testing.T) {
 		t.Fatal("explicit private report-IP corroboration lost")
 	}
 }
+
+// The Proxmox disk row carries a copy of the agent's collection state from the
+// last disk poll. When the agent's own row later withdraws a reading (its
+// reporting lease expired), the canonical disk must follow the agent in either
+// ingest order, even if no disk poll has refreshed the Proxmox copy (a host that
+// is down entirely is never disk-polled).
+func TestPhysicalDiskMergeFollowsAgentWithdrawalOfItsOwnReadings(t *testing.T) {
+	proxmoxDisk := func(agent diskinventory.FieldStatus) Resource {
+		return Resource{
+			Type: ResourceTypePhysicalDisk, Name: "WDC", Status: StatusOnline,
+			PhysicalDisk: &PhysicalDiskMeta{
+				DevPath: "/dev/sdb", Serial: "WD-SILENT1", Temperature: 41,
+				Collection: &diskinventory.CollectionStatus{
+					Serial:      diskinventory.Available("proxmox_disks"),
+					Temperature: agent,
+					Pool:        diskinventory.Available("proxmox_zfs"),
+				},
+			},
+		}
+	}
+	agentDisk := func(temperature diskinventory.FieldStatus) Resource {
+		return Resource{
+			Type: ResourceTypePhysicalDisk, Name: "WDC", Status: StatusOnline,
+			PhysicalDisk: &PhysicalDiskMeta{
+				DevPath: "/dev/sdb", Serial: "WD-SILENT1", Temperature: 41,
+				Collection: &diskinventory.CollectionStatus{
+					Serial:      diskinventory.Available("smartctl"),
+					Temperature: temperature,
+				},
+			},
+		}
+	}
+	identity := ResourceIdentity{MachineID: "WD-SILENT1", Hostnames: []string{"node1"}}
+	collected := diskinventory.Available("smartctl")
+	expired := diskinventory.Unavailable("smartctl", "host agent stopped reporting")
+
+	for _, tc := range []struct {
+		name         string
+		proxmoxFirst bool
+		agent        diskinventory.FieldStatus
+		want         diskinventory.FieldStatus
+	}{
+		{"proxmox then expired agent", true, expired, expired},
+		{"expired agent then proxmox", false, expired, expired},
+		{"proxmox then reporting agent", true, collected, collected},
+		{"reporting agent then proxmox", false, collected, collected},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			registry := NewRegistry(nil)
+			// The Proxmox row still holds the earlier "collected" copy.
+			ingestProxmox := func() { registry.ingest(SourceProxmox, "pve1-node1-sdb", proxmoxDisk(collected), identity) }
+			ingestAgent := func() { registry.ingest(SourceAgent, "agent-1-sdb", agentDisk(tc.agent), identity) }
+			if tc.proxmoxFirst {
+				ingestProxmox()
+				ingestAgent()
+			} else {
+				ingestAgent()
+				ingestProxmox()
+			}
+			disks := registry.ListByType(ResourceTypePhysicalDisk)
+			if len(disks) != 1 || disks[0].PhysicalDisk == nil || disks[0].PhysicalDisk.Collection == nil {
+				t.Fatalf("expected one merged disk, got %+v", disks)
+			}
+			collection := disks[0].PhysicalDisk.Collection
+			if collection.Temperature != tc.want {
+				t.Fatalf("temperature collection = %+v, want %+v", collection.Temperature, tc.want)
+			}
+			if disks[0].PhysicalDisk.Temperature != 41 {
+				t.Fatalf("last-known temperature was not retained: %d", disks[0].PhysicalDisk.Temperature)
+			}
+			if collection.Pool != diskinventory.Available("proxmox_zfs") {
+				t.Fatalf("Proxmox-owned evidence changed: %+v", collection.Pool)
+			}
+		})
+	}
+}

pkg/diskinventory/status.go

@@ -89,6 +89,35 @@ func MergeStatus(existing, incoming *CollectionStatus) *CollectionStatus {
 	return &merged
 }
 
+// MergeReportedStatus merges a source's own report into existing state that
+// may still hold an earlier copy of that report. It behaves like MergeStatus,
+// except that a field the report no longer marks available supersedes an
+// available state that existing carries from the same source: the source's
+// later word on its own field wins, so a stale copy cannot keep claiming the
+// field was collected.
+func MergeReportedStatus(existing, reported *CollectionStatus) *CollectionStatus {
+	merged := MergeStatus(existing, reported)
+	if existing == nil || reported == nil {
+		return merged
+	}
+	merged.Serial = supersedeFieldStatus(merged.Serial, existing.Serial, reported.Serial)
+	merged.Temperature = supersedeFieldStatus(merged.Temperature, existing.Temperature, reported.Temperature)
+	merged.IO = supersedeFieldStatus(merged.IO, existing.IO, reported.IO)
+	merged.Controller = supersedeFieldStatus(merged.Controller, existing.Controller, reported.Controller)
+	merged.Pool = supersedeFieldStatus(merged.Pool, existing.Pool, reported.Pool)
+	return merged
+}
+
+func supersedeFieldStatus(merged, existing, reported FieldStatus) FieldStatus {
+	source := strings.TrimSpace(existing.Source)
+	if existing.State == FieldAvailable &&
+		reported.State != "" && reported.State != FieldAvailable &&
+		source != "" && strings.EqualFold(source, strings.TrimSpace(reported.Source)) {
+		return reported
+	}
+	return merged
+}
+
 func mergeFieldStatus(existing, incoming FieldStatus) FieldStatus {
 	if incoming.State == "" {
 		return existing

pkg/diskinventory/status_test.go

@@ -24,3 +24,44 @@ func TestMergeStatusPrefersAvailableEvidencePerField(t *testing.T) {
 		t.Fatalf("missing I/O state was not retained: %+v", got.IO)
 	}
 }
+
+func TestMergeReportedStatusLetsTheSourceWithdrawItsOwnAvailability(t *testing.T) {
+	existing := &CollectionStatus{
+		Serial:      Available("smartctl"),
+		Temperature: Available("smartctl"),
+		IO:          Available("kernel_diskstats"),
+		Pool:        Available("proxmox_zfs"),
+	}
+	reported := &CollectionStatus{
+		Serial:      Available("smartctl"),
+		Temperature: Unavailable("smartctl", "host agent stopped reporting"),
+		IO:          Unavailable("kernel_diskstats", "host agent stopped reporting"),
+		Pool:        Unavailable("zpool", "command failed"),
+	}
+
+	got := MergeReportedStatus(existing, reported)
+	if got.Temperature != reported.Temperature || got.IO != reported.IO {
+		t.Fatalf("the source's later word on its own fields was ignored: %+v", got)
+	}
+	if got.Serial != Available("smartctl") {
+		t.Fatalf("still-available field changed: %+v", got.Serial)
+	}
+	if got.Pool != Available("proxmox_zfs") {
+		t.Fatalf("another source's available evidence was withdrawn: %+v", got.Pool)
+	}
+
+	// A copy that is already weaker never overrides a current report, and the
+	// rule is plain MergeStatus when either side is absent.
+	if got := MergeReportedStatus(reported, existing); got.Temperature != Available("smartctl") {
+		t.Fatalf("current report lost to a weaker copy: %+v", got.Temperature)
+	}
+	if got := MergeReportedStatus(nil, reported); got.Temperature != reported.Temperature {
+		t.Fatalf("nil existing: %+v", got)
+	}
+	if got := MergeReportedStatus(existing, nil); got.Temperature != existing.Temperature {
+		t.Fatalf("nil report: %+v", got)
+	}
+	if existing.Temperature != Available("smartctl") {
+		t.Fatal("MergeReportedStatus mutated its input")
+	}
+}