fix: correct Proxmox guest rate sampling

2026-09-10 02:25:56 +00:00 · 2026-07-24 10:17:21 +01:00
parent ac0fb263c2
commit 49217d284d
42 changed files with 1717 additions and 719 deletions
@@ -36,6 +36,11 @@ one compressed unified-agent report without a collector-side suffix cap.
 websocket payloads. Carrying plural availability facets through that serializer
 is an adjacent monitoring/API projection and does not change agent enrollment,
 report admission, removal, update, profile, or command authority.
+The JSON-excluded Proxmox VM/LXC I/O-rate validity fields carried by
+`internal/models/models.go` are likewise monitoring-owned sidecar evidence.
+They distinguish a valid idle interval from an unknown rate for history,
+alerts, and unified-resource projection; they do not describe agent report
+admission, heartbeat, enrollment, command reachability, or fleet liveness.
 Docker / Podman report sizing is a two-dimensional transport contract:
 `pkg/agents/docker/report_limits.go` owns the inclusive encoded HTTP-body and
 decoded JSON byte ceilings, derives the 80% early-warning boundaries, and
@@ -43,6 +43,13 @@ used/free/total evidence must not open or clear a memory alert. If such a
 sample follows an active alert, the alert remains active with its last trusted
 value until a subsequent trusted sample proves recovery; missing evidence is
 not evidence that pressure disappeared.
+Proxmox guest disk-read, disk-write, network-in, and network-out thresholds
+consume only monitoring-owned valid rate observations. A valid idle interval
+is explicit zero and may prove recovery; a first sample, missing/null counter,
+partial response, or rejected out-of-order sample is unknown and must not
+start, clear, or match a custom metric filter. Alert units remain MiB/s at the
+threshold boundary (`bytes/s / 1024 / 1024`); that display/threshold conversion
+must not be applied to the upstream cumulative-counter divisor.

 ## Canonical Files

@@ -144,6 +144,31 @@ no core-count division and no in-guest host-agent substitution. A linked host
 agent may supplement metrics the platform does not provide, but guest CPU and
 its `vm` / `system-container` history target remain platform-owned so dashboard,
 details, API state, alerts, and history cannot select different authorities.
+Proxmox guest disk and network throughput has one cumulative-counter sampling
+contract. `diskread`, `diskwrite`, `netin`, and `netout` are cumulative bytes;
+the canonical rate is `(current counter - previous counter) / elapsed
+observation seconds`, in bytes per second, with no 1024 divisor. Elapsed time
+comes from the receipt time stamped immediately after the relevant Proxmox API
+response is decoded, not from later guest-agent, filesystem, or metadata
+enrichment. Each counter keeps an independent adjacent-sample baseline:
+explicitly unchanged counters produce a valid zero, missing/null fields produce
+unknown, out-of-order samples produce unknown without moving the baseline, and
+a counter decrease caused by restart, reconnect, migration epoch change, or
+wrap rebases that counter and produces a valid zero for the reset interval. A
+source-uptime rollback rebases the complete counter epoch and leaves the first
+post-restart rate unknown, including when a busy guest already surpassed its
+pre-restart counter value before the next poll.
+First-sample and missing-field unknowns remain internal validity state; the
+legacy API/websocket guest number fields stay numeric, while history, unified
+metrics, and alerts omit the unknown observation instead of manufacturing
+zero. The rate-tracker identity is `(configured PVE instance, guest kind,
+VMID)`: it survives node migration, separates QEMU from LXC, and prevents
+duplicate configured cluster identities from sharing a concurrent baseline.
+Idle and partial samples still refresh tracker liveness.
+Proxmox row liveness uses the same cadence-derived threshold as source
+freshness (`max(2 * configured poll interval, 60s)`). Node offline grace and
+guest preservation must not expire between healthy 60- or 90-second polls, and
+must not use a separate fixed 60-second timer.
 Tenant monitor enumeration is monitoring-owned runtime topology, not a
 reporting source of truth. `MultiTenantMonitor.ListOrganizationIDs` may expose
 persisted organization IDs to API-owned background workers, but it must not
@@ -202,6 +227,8 @@ node-local Agent evidence.
 20. `internal/dockeragent/collect.go`
 21. `pkg/proxmox/ceph.go`
 21a. `pkg/proxmox/cluster_client.go`
+21b. `pkg/proxmox/client.go`
+21c. `pkg/proxmox/io_counters.go`
 22. `pkg/proxmox/zfs.go`
 23. `internal/monitoring/guest_memory_sources.go`
 24. `internal/monitoring/guest_memory_stability.go`
@@ -222,6 +249,7 @@ node-local Agent evidence.
 37. `pkg/agents/docker/report.go`
 38. `internal/models/models.go`
 38a. `internal/models/proxmox_guest_state.go`
+38b. `internal/models/metrics_types.go`
 39. `internal/models/models_frontend.go`
 40. `internal/models/converters.go`
 41. `internal/models/deepcopy.go`
@@ -1076,14 +1104,17 @@ legacy PBS and PMG summaries on current `LastSeen` and health state each tick,
 so long-lived infrastructure, workloads, storage, and recovery demos do not
 decay into synthetic stale-state warnings while mock mode remains enabled.
 That same Proxmox container monitoring boundary now also owns runtime counter
-recovery when the lower-fidelity container list or cluster-resources payload
-reports stale or zero I/O totals. `internal/monitoring/monitor_pve.go`,
+authority when the lower-fidelity container list or cluster-resources payload
+and the current-status payload differ. `internal/monitoring/monitor_pve.go`,
 `internal/monitoring/monitor_pve_guest_lxc.go`, and
 `internal/monitoring/monitor_polling_containers.go` must merge the current
 `GetContainerStatus` counters through one canonical `mergeContainerRuntimeCounters`
-path before LXC rate calculation and must reuse the same prefetched status
-snapshot for metadata enrichment instead of paying disconnected metric and
-metadata status reads that can diverge.
+path before LXC rate calculation. A present status field is newer authority
+even when it is zero or lower after a restart; an absent/null status field
+retains the listing field and its presence state. The merge must retain the
+status response receipt time and reuse the same prefetched status snapshot for
+metadata enrichment instead of paying disconnected metric and metadata status
+reads that can diverge.
 That same Proxmox backup/snapshot boundary owns bounded concurrent guest
 snapshot enumeration. `internal/monitoring/monitor_backups.go` must query VM
 and LXC snapshot endpoints through one capped worker pool and preserve
@@ -1349,6 +1349,7 @@
            "exact_files": [
              "internal/api/host_agent_removal_lifecycle_integration_test.go",
              "internal/config/host_continuity_test.go",
+              "internal/models/metrics_types_test.go",
              "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
              "internal/monitoring/monitor_host_agents_test.go",
              "scripts/installtests/agent_state_dir_lifecycle_test.go",
@@ -5184,6 +5185,7 @@
        "internal/mock/platform_fixtures.go",
        "internal/models/converters.go",
        "internal/models/deepcopy.go",
+        "internal/models/metrics_types.go",
        "internal/models/models.go",
        "internal/models/models_frontend.go",
        "internal/models/proxmox_guest_state.go",
@@ -5192,7 +5194,9 @@
        "pkg/agents/host/report.go",
        "pkg/agents/kubernetes/report.go",
        "pkg/proxmox/ceph.go",
+        "pkg/proxmox/client.go",
        "pkg/proxmox/cluster_client.go",
+        "pkg/proxmox/io_counters.go",
        "pkg/proxmox/zfs.go"
      ],
      "verification": {
@@ -5397,6 +5401,7 @@
            "match_files": [
              "internal/models/converters.go",
              "internal/models/deepcopy.go",
+              "internal/models/metrics_types.go",
              "internal/models/models.go",
              "internal/models/models_frontend.go",
              "internal/models/proxmox_guest_state.go",
@@ -5408,6 +5413,7 @@
              "internal/dockeragent/swarm_coverage_test.go",
              "internal/models/deepcopy_test.go",
              "internal/models/host_transient_expiry_test.go",
+              "internal/models/metrics_types_test.go",
              "internal/models/state_host_test.go",
              "internal/monitoring/issue1595_collection_trust_test.go",
              "internal/monitoring/monitor_full_coverage_test.go",
@@ -5418,6 +5424,24 @@
              "internal/unifiedresources/registry_test.go"
            ]
          },
+          {
+            "id": "proxmox-guest-counter-runtime",
+            "label": "Proxmox guest cumulative-counter presence and receipt-time proof",
+            "match_prefixes": [],
+            "match_files": [
+              "pkg/proxmox/client.go",
+              "pkg/proxmox/io_counters.go"
+            ],
+            "allow_same_subsystem_tests": false,
+            "test_prefixes": [],
+            "exact_files": [
+              "internal/api/router_state_test.go",
+              "internal/monitoring/issue1613_contract_test.go",
+              "internal/monitoring/monitor_pve_guest_lxc_test.go",
+              "internal/monitoring/ratetracker_test.go",
+              "pkg/proxmox/io_counters_test.go"
+            ]
+          },
          {
            "id": "proxmox-zfs-runtime",
            "label": "proxmox zfs compatibility proof",
@@ -5587,6 +5611,7 @@
              "internal/monitoring/canonical_guardrails_test.go",
              "internal/monitoring/issue1485_unraid_lifecycle_test.go",
              "internal/monitoring/issue1595_collection_trust_test.go",
+              "internal/monitoring/issue1613_contract_test.go",
              "internal/monitoring/monitor_additional_test.go",
              "internal/monitoring/monitor_alert_intent_test.go",
              "internal/monitoring/monitor_alert_override_migration_test.go",
@@ -5595,6 +5620,8 @@
              "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
              "internal/monitoring/monitor_host_agents_test.go",
              "internal/monitoring/monitor_pve_cluster_refresh_test.go",
+              "internal/monitoring/monitor_pve_guest_lxc_test.go",
+              "internal/monitoring/ratetracker_test.go",
              "internal/unifiedresources/code_standards_test.go"
            ]
          }
@@ -1568,6 +1568,11 @@ recovery scope, or a storage/recovery-owned secret source.
    remain absent optional timestamps on that shared facet; storage and recovery
    must not reinterpret absence or a year-one zero-time serialization as
    recovery age, missed backup cadence, or restore freshness.
+    The same shared type file keeps `MetricValue.value` numeric whenever a
+    metric object is present. Storage/recovery consumers must preserve explicit
+    zero as an observed idle value and use absence of the metric object for
+    unknown Proxmox guest I/O; they must not reinterpret zero as missing or
+    accept a JSON `null` value as recovery, backup, or storage-health evidence.

 ## Forbidden Paths

@@ -55,6 +55,14 @@ at the adapter boundary; a mutation that arrives during a rebuild must apply to
 the newly published registry rather than a superseded pointer. Authoritative
 snapshot omissions still remove resources and emit the normal canonical
 history change.
+Proxmox guest I/O projection preserves the monitoring-owned rate validity
+contract. A valid idle interval publishes an explicit zero-valued `bytes/s`
+metric; an unknown first, missing, partial, or out-of-order observation leaves
+that metric absent. `MetricValue.value` is always a numeric JSON field when a
+metric object exists, so `/api/resources`, `/api/state`, websocket resources,
+history, alerts, and UI adapters never need to interpret `null` as either idle
+or unknown. Unified resources must not infer validity from the numeric guest
+compatibility field alone.
 Physical-disk resources own cross-source disk identity. When Proxmox inventory
 and host-agent SMART telemetry describe the same device, the merged resource
 must retain Proxmox node/instance source payloads while carrying SMART
@@ -1199,6 +1199,23 @@ describe('getOutlierEmphasis (via I/O column rendering)', () => {
    expect(screen.getAllByText('—').length).toBeGreaterThan(0);
  });

+  it('renders valid zero I/O rates for a running guest on an online parent', () => {
+    renderGuestRow({
+      guest: makeGuest({
+        status: 'running',
+        networkIn: 0,
+        networkOut: 0,
+        diskRead: 0,
+        diskWrite: 0,
+      }),
+      parentNodeOnline: true,
+      visibleColumnIds: ['name', 'netIo', 'diskIo'],
+    });
+
+    expect(screen.getAllByText('0 B/s')).toHaveLength(4);
+    expect(screen.queryByText('—')).toBeNull();
+  });
+
  it('shows dash for disk I/O when guest is stopped', () => {
    renderGuestRow({
      guest: makeGuest({ status: 'stopped', diskRead: 0, diskWrite: 0 }),
@@ -20,17 +20,18 @@ func (m *Manager) evaluateFilterCondition(guest any, condition FilterCondition)

 // guestMetrics holds common metrics for filter evaluation
 type guestMetrics struct {
-	CPU        float64 // CPU usage as percentage (0-100)
-	MemUsage   float64 // Memory usage percentage
-	DiskUsage  float64 // Disk usage percentage
-	DiskRead   int64   // Bytes/s
-	DiskWrite  int64   // Bytes/s
-	NetworkIn  int64   // Bytes/s
-	NetworkOut int64   // Bytes/s
-	Name       string
-	Node       string
-	ID         string
-	Status     string
+	CPU            float64 // CPU usage as percentage (0-100)
+	MemUsage       float64 // Memory usage percentage
+	DiskUsage      float64 // Disk usage percentage
+	DiskRead       int64   // Bytes/s
+	DiskWrite      int64   // Bytes/s
+	NetworkIn      int64   // Bytes/s
+	NetworkOut     int64   // Bytes/s
+	IORateValidity models.IORateValidity
+	Name           string
+	Node           string
+	ID             string
+	Status         string
 }

 // extractGuestMetrics extracts common metrics from a VM or Container
@@ -47,6 +48,12 @@ func evaluateGuestCondition(metrics guestMetrics, condition FilterCondition) boo
 	switch condition.Type {
 	case "metric":
 		value := 0.0
+		validity := metrics.IORateValidity.EffectiveForRates(
+			metrics.DiskRead,
+			metrics.DiskWrite,
+			metrics.NetworkIn,
+			metrics.NetworkOut,
+		)
 		switch strings.ToLower(condition.Field) {
 		case "cpu":
 			value = metrics.CPU
@@ -55,12 +62,24 @@ func evaluateGuestCondition(metrics guestMetrics, condition FilterCondition) boo
 		case "disk":
 			value = metrics.DiskUsage
 		case "diskread":
+			if !validity.DiskRead {
+				return false
+			}
 			value = float64(metrics.DiskRead) / 1024 / 1024 // Convert to MB/s
 		case "diskwrite":
+			if !validity.DiskWrite {
+				return false
+			}
 			value = float64(metrics.DiskWrite) / 1024 / 1024
 		case "networkin":
+			if !validity.NetworkIn {
+				return false
+			}
 			value = float64(metrics.NetworkIn) / 1024 / 1024
 		case "networkout":
+			if !validity.NetworkOut {
+				return false
+			}
 			value = float64(metrics.NetworkOut) / 1024 / 1024
 		default:
 			return false
@@ -374,6 +374,28 @@ func TestEvaluateVMCondition(t *testing.T) {
 	}
 }

+func TestEvaluateVMConditionTreatsUnknownIORateAsNonMatch(t *testing.T) {
+	manager := NewManager()
+	vm := models.VM{
+		DiskRead:       0,
+		IORateValidity: models.IORateValidity{Explicit: true},
+	}
+	condition := FilterCondition{
+		Type:     "metric",
+		Field:    "diskread",
+		Operator: "<=",
+		Value:    0,
+	}
+	if manager.evaluateVMCondition(vm, condition) {
+		t.Fatal("unknown disk read rate matched a zero-valued filter")
+	}
+
+	vm.IORateValidity.DiskRead = true
+	if !manager.evaluateVMCondition(vm, condition) {
+		t.Fatal("valid zero disk read rate did not match a zero-valued filter")
+	}
+}
+
 func TestEvaluateContainerCondition(t *testing.T) {
 	// t.Parallel()
 	m := NewManager()
@@ -124,10 +124,6 @@ func (m *Manager) CheckGuest(guest any, instanceName string) {
 	cpu := snapshot.CPUPercent
 	memUsage := snapshot.MemUsage
 	diskUsage := snapshot.DiskUsage
-	diskRead := snapshot.DiskRead
-	diskWrite := snapshot.DiskWrite
-	netIn := snapshot.NetworkIn
-	netOut := snapshot.NetworkOut
 	disks := snapshot.Disks

 	// Debug logging for high memory VMs
@@ -265,6 +261,7 @@ func (m *Manager) CheckGuest(guest any, instanceName string) {
 	if !snapshot.MemoryUnavailable {
 		memoryMetric = &UnifiedResourceMetric{Percent: memUsage}
 	}
+	diskReadMetric, diskWriteMetric, networkInMetric, networkOutMetric := guestIORateMetrics(snapshot)
 	m.evaluateUnifiedMetrics(&UnifiedResourceInput{
 		ID:         guestID,
 		Type:       snapshot.resourceType(),
@@ -274,10 +271,10 @@ func (m *Manager) CheckGuest(guest any, instanceName string) {
 		CPU:        &UnifiedResourceMetric{Percent: cpu},
 		Memory:     memoryMetric,
 		Disk:       &UnifiedResourceMetric{Percent: diskUsage},
-		DiskRead:   &UnifiedResourceMetric{Value: float64(diskRead) / 1024 / 1024},
-		DiskWrite:  &UnifiedResourceMetric{Value: float64(diskWrite) / 1024 / 1024},
-		NetworkIn:  &UnifiedResourceMetric{Value: float64(netIn) / 1024 / 1024},
-		NetworkOut: &UnifiedResourceMetric{Value: float64(netOut) / 1024 / 1024},
+		DiskRead:   diskReadMetric,
+		DiskWrite:  diskWriteMetric,
+		NetworkIn:  networkInMetric,
+		NetworkOut: networkOutMetric,
 	}, thresholds, evalOpts)

 	if thresholds.Disk != nil && thresholds.Disk.Trigger > 0 && len(disks) > 0 {
@@ -369,6 +366,28 @@ func (m *Manager) CheckGuest(guest any, instanceName string) {
 	}
 }

+func guestIORateMetrics(snapshot guestSnapshot) (diskRead, diskWrite, networkIn, networkOut *UnifiedResourceMetric) {
+	validity := snapshot.IORateValidity.EffectiveForRates(
+		snapshot.DiskRead,
+		snapshot.DiskWrite,
+		snapshot.NetworkIn,
+		snapshot.NetworkOut,
+	)
+	if validity.DiskRead {
+		diskRead = &UnifiedResourceMetric{Value: float64(snapshot.DiskRead) / 1024 / 1024}
+	}
+	if validity.DiskWrite {
+		diskWrite = &UnifiedResourceMetric{Value: float64(snapshot.DiskWrite) / 1024 / 1024}
+	}
+	if validity.NetworkIn {
+		networkIn = &UnifiedResourceMetric{Value: float64(snapshot.NetworkIn) / 1024 / 1024}
+	}
+	if validity.NetworkOut {
+		networkOut = &UnifiedResourceMetric{Value: float64(snapshot.NetworkOut) / 1024 / 1024}
+	}
+	return
+}
+
 // checkGuestPoweredOff creates an alert for powered-off guests.
 func (m *Manager) checkGuestPoweredOff(guestID, name, node, instanceName, guestType string, monitorOnly bool) {
 	m.mu.RLock()
@@ -36,6 +36,7 @@ type guestSnapshot struct {
 	DiskWrite         int64
 	NetworkIn         int64
 	NetworkOut        int64
+	IORateValidity    models.IORateValidity

 	Disks  []models.Disk
 	Tags   []string
@@ -80,17 +81,18 @@ func (g guestSnapshot) resourceType() string {

 func (g guestSnapshot) metrics() guestMetrics {
 	return guestMetrics{
-		CPU:        g.CPUPercent,
-		MemUsage:   g.MemUsage,
-		DiskUsage:  g.DiskUsage,
-		DiskRead:   g.DiskRead,
-		DiskWrite:  g.DiskWrite,
-		NetworkIn:  g.NetworkIn,
-		NetworkOut: g.NetworkOut,
-		Name:       g.Name,
-		Node:       g.Node,
-		ID:         g.ID,
-		Status:     g.Status,
+		CPU:            g.CPUPercent,
+		MemUsage:       g.MemUsage,
+		DiskUsage:      g.DiskUsage,
+		DiskRead:       g.DiskRead,
+		DiskWrite:      g.DiskWrite,
+		NetworkIn:      g.NetworkIn,
+		NetworkOut:     g.NetworkOut,
+		IORateValidity: g.IORateValidity,
+		Name:           g.Name,
+		Node:           g.Node,
+		ID:             g.ID,
+		Status:         g.Status,
 	}
 }

@@ -112,6 +114,7 @@ func guestSnapshotFromVM(vm models.VM) guestSnapshot {
 		DiskWrite:         vm.DiskWrite,
 		NetworkIn:         vm.NetworkIn,
 		NetworkOut:        vm.NetworkOut,
+		IORateValidity:    vm.IORateValidity,
 		Disks:             append([]models.Disk(nil), vm.Disks...),
 		Tags:              append([]string(nil), vm.Tags...),
 		OnBoot:            vm.OnBoot,
@@ -136,6 +139,7 @@ func guestSnapshotFromContainer(container models.Container) guestSnapshot {
 		DiskWrite:         container.DiskWrite,
 		NetworkIn:         container.NetworkIn,
 		NetworkOut:        container.NetworkOut,
+		IORateValidity:    container.IORateValidity,
 		Disks:             append([]models.Disk(nil), container.Disks...),
 		Tags:              append([]string(nil), container.Tags...),
 		OnBoot:            container.OnBoot,
@@ -94,3 +94,27 @@ func TestExtractGuestSnapshot_UnknownTypeReturnsCanonicalEmptySnapshot(t *testin
 		t.Fatalf("expected canonical empty guest snapshot on unknown type, got %#v", snapshot)
 	}
 }
+
+func TestGuestIORateMetricsDistinguishValidZeroFromUnknown(t *testing.T) {
+	validZero := guestSnapshotFromVM(models.VM{
+		IORateValidity: models.IORateValidity{
+			Explicit:   true,
+			DiskRead:   true,
+			DiskWrite:  true,
+			NetworkIn:  true,
+			NetworkOut: true,
+		},
+	})
+	diskRead, diskWrite, networkIn, networkOut := guestIORateMetrics(validZero)
+	if diskRead == nil || diskRead.Value != 0 || diskWrite == nil || networkIn == nil || networkOut == nil {
+		t.Fatalf("valid zero rates were not alert candidates: %+v %+v %+v %+v", diskRead, diskWrite, networkIn, networkOut)
+	}
+
+	unknown := guestSnapshotFromVM(models.VM{
+		IORateValidity: models.IORateValidity{Explicit: true},
+	})
+	diskRead, diskWrite, networkIn, networkOut = guestIORateMetrics(unknown)
+	if diskRead != nil || diskWrite != nil || networkIn != nil || networkOut != nil {
+		t.Fatalf("unknown rates became alert candidates: %+v %+v %+v %+v", diskRead, diskWrite, networkIn, networkOut)
+	}
+}
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"

@@ -107,6 +108,83 @@ func TestRouter_HandleState_MockIsolation(t *testing.T) {
 	})
 }

+func TestRouterHandleStatePreservesNumericIdleRatesAndOmitsUnknownRates(t *testing.T) {
+	dataPath := t.TempDir()
+	hp, err := auth.HashPassword("password")
+	if err != nil {
+		t.Fatalf("hash password: %v", err)
+	}
+	cfg := &config.Config{
+		DataPath: dataPath,
+		AuthUser: "admin",
+		AuthPass: hp,
+	}
+	InitSessionStore(dataPath)
+	InitCSRFStore(dataPath)
+
+	monitor, state, _ := newTestMonitor(t)
+	state.VMs = []models.VM{
+		{
+			ID:       "site-a:pve-a:100",
+			VMID:     100,
+			Name:     "idle-vm",
+			Node:     "pve-a",
+			Instance: "site-a",
+			Status:   "running",
+			IORateValidity: models.IORateValidity{
+				Explicit:   true,
+				DiskRead:   true,
+				DiskWrite:  true,
+				NetworkIn:  true,
+				NetworkOut: true,
+			},
+		},
+		{
+			ID:       "site-a:pve-a:101",
+			VMID:     101,
+			Name:     "unknown-vm",
+			Node:     "pve-a",
+			Instance: "site-a",
+			Status:   "running",
+			IORateValidity: models.IORateValidity{
+				Explicit: true,
+			},
+		},
+	}
+	syncTestResourceStore(t, monitor, state)
+
+	router := &Router{config: cfg, monitor: monitor}
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/api/state", nil)
+	req.SetBasicAuth("admin", "password")
+	router.handleState(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("/api/state status = %d, body=%s", rec.Code, rec.Body.String())
+	}
+
+	var payload struct {
+		Resources []models.ResourceFrontend `json:"resources"`
+	}
+	if err := json.NewDecoder(rec.Body).Decode(&payload); err != nil {
+		t.Fatalf("decode /api/state: %v", err)
+	}
+
+	byName := make(map[string]models.ResourceFrontend, len(payload.Resources))
+	for _, resource := range payload.Resources {
+		byName[resource.Name] = resource
+	}
+	idle := byName["idle-vm"]
+	if idle.DiskIO == nil || idle.DiskIO.ReadRate != 0 || idle.DiskIO.WriteRate != 0 {
+		t.Fatalf("valid idle rates were not emitted as numeric zero: %+v", idle.DiskIO)
+	}
+	if unknown := byName["unknown-vm"]; unknown.DiskIO != nil {
+		t.Fatalf("unknown rates projected a disk I/O object: %+v", unknown.DiskIO)
+	}
+	if strings.Contains(rec.Body.String(), `"readRate":null`) || strings.Contains(rec.Body.String(), `"writeRate":null`) {
+		t.Fatalf("/api/state emitted unstable null rates: %s", rec.Body.String())
+	}
+}
+
 func TestRouter_HandleStateSummary(t *testing.T) {
 	dataPath := t.TempDir()
 	hp, _ := auth.HashPassword("password")
@@ -10,10 +10,83 @@ type MetricPoint struct {

 // IOMetrics represents I/O metrics at a point in time
 type IOMetrics struct {
-	DiskRead   int64     `json:"diskRead"`
-	DiskWrite  int64     `json:"diskWrite"`
-	DiskBusy   int64     `json:"diskBusy,omitempty"`
-	NetworkIn  int64     `json:"networkIn"`
-	NetworkOut int64     `json:"networkOut"`
-	Timestamp  time.Time `json:"timestamp"`
+	DiskRead   int64                     `json:"diskRead"`
+	DiskWrite  int64                     `json:"diskWrite"`
+	DiskBusy   int64                     `json:"diskBusy,omitempty"`
+	NetworkIn  int64                     `json:"networkIn"`
+	NetworkOut int64                     `json:"networkOut"`
+	Timestamp  time.Time                 `json:"timestamp"`
+	Presence   IOCounterPresence         `json:"-"`
+	ObservedAt IOCounterObservationTimes `json:"-"`
+	// SourceUptime is an optional counter-epoch hint. A decrease proves the
+	// source restarted even when a busy guest has already surpassed its old
+	// counter value before the next poll.
+	SourceUptime uint64 `json:"-"`
+}
+
+// IOCounterObservationTimes keeps the receipt time of each independently
+// sampled counter. A zero field falls back to IOMetrics.Timestamp for legacy
+// producers that obtain every counter in one response.
+type IOCounterObservationTimes struct {
+	DiskRead   time.Time
+	DiskWrite  time.Time
+	DiskBusy   time.Time
+	NetworkIn  time.Time
+	NetworkOut time.Time
+}
+
+// IOCounterPresence distinguishes a counter that was explicitly observed at
+// zero from one that was absent in the upstream sample. Explicit=false keeps
+// legacy producers compatible by treating every counter as present.
+type IOCounterPresence struct {
+	Explicit   bool
+	DiskRead   bool
+	DiskWrite  bool
+	DiskBusy   bool
+	NetworkIn  bool
+	NetworkOut bool
+}
+
+// Effective returns the presence contract used by rate calculation.
+func (p IOCounterPresence) Effective() IOCounterPresence {
+	if p.Explicit {
+		return p
+	}
+	return IOCounterPresence{
+		Explicit:   true,
+		DiskRead:   true,
+		DiskWrite:  true,
+		DiskBusy:   true,
+		NetworkIn:  true,
+		NetworkOut: true,
+	}
+}
+
+// IORateValidity records which numeric guest rate fields represent an
+// observed rate. It is intentionally excluded from JSON: legacy API and
+// websocket contracts continue to expose stable numbers, while internal
+// history, alerting, and unified-resource consumers can preserve unknown.
+type IORateValidity struct {
+	Explicit   bool
+	DiskRead   bool
+	DiskWrite  bool
+	NetworkIn  bool
+	NetworkOut bool
+}
+
+// EffectiveForRates preserves compatibility with legacy producers that did
+// not carry explicit validity. Their non-zero values are usable evidence, but
+// a legacy zero remains ambiguous and must not override a supplemental source
+// or prove alert recovery.
+func (v IORateValidity) EffectiveForRates(diskRead, diskWrite, networkIn, networkOut int64) IORateValidity {
+	if v.Explicit {
+		return v
+	}
+	return IORateValidity{
+		Explicit:   true,
+		DiskRead:   diskRead != 0,
+		DiskWrite:  diskWrite != 0,
+		NetworkIn:  networkIn != 0,
+		NetworkOut: networkOut != 0,
+	}
 }
@@ -2,6 +2,7 @@ package models

 import (
 	"encoding/json"
+	"strings"
 	"testing"
 	"time"
 )
@@ -26,6 +27,7 @@ func TestIOMetrics_Fields(t *testing.T) {
 		NetworkIn:  3000,
 		NetworkOut: 4000,
 		Timestamp:  now,
+		ObservedAt: IOCounterObservationTimes{DiskRead: now},
 	}

 	if metrics.DiskRead != 1000 {
@@ -116,4 +118,60 @@ func TestIOMetrics_JSONSerializationUsesCamelCaseFields(t *testing.T) {
 	if _, ok := decoded["Timestamp"]; ok {
 		t.Error("expected Timestamp key to be absent")
 	}
+	if _, ok := decoded["ObservedAt"]; ok {
+		t.Error("expected per-counter observation times to be absent")
+	}
+}
+
+func TestIORateValidityLegacyFallbackTreatsOnlyNonZeroRatesAsKnown(t *testing.T) {
+	validity := (IORateValidity{}).EffectiveForRates(1024, 0, 2048, 0)
+	if !validity.DiskRead || validity.DiskWrite || !validity.NetworkIn || validity.NetworkOut {
+		t.Fatalf("legacy inferred validity = %+v", validity)
+	}
+
+	explicitIdle := IORateValidity{
+		Explicit:   true,
+		DiskRead:   true,
+		DiskWrite:  true,
+		NetworkIn:  true,
+		NetworkOut: true,
+	}.EffectiveForRates(0, 0, 0, 0)
+	if !explicitIdle.DiskRead || !explicitIdle.DiskWrite || !explicitIdle.NetworkIn || !explicitIdle.NetworkOut {
+		t.Fatalf("explicit idle validity was not preserved: %+v", explicitIdle)
+	}
+}
+
+func TestUnknownGuestRatesRemainNumericOnAPIAndWebsocketShapes(t *testing.T) {
+	vm := VM{
+		ID:        "site:node:100",
+		DiskRead:  0,
+		DiskWrite: 0,
+		IORateValidity: IORateValidity{
+			Explicit: true,
+		},
+	}
+	modelPayload, err := json.Marshal(vm)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(modelPayload), "IORateValidity") || strings.Contains(string(modelPayload), "ioRateValidity") {
+		t.Fatalf("internal validity leaked into guest model JSON: %s", modelPayload)
+	}
+
+	payload, err := json.Marshal(vm.ToFrontend())
+	if err != nil {
+		t.Fatal(err)
+	}
+	wire := string(payload)
+	for _, field := range []string{`"diskRead":0`, `"diskWrite":0`, `"networkIn":0`, `"networkOut":0`} {
+		if !strings.Contains(wire, field) {
+			t.Fatalf("wire payload %s does not contain numeric field %s", wire, field)
+		}
+	}
+	if strings.Contains(wire, "null") {
+		t.Fatalf("guest wire payload contains unstable null: %s", wire)
+	}
+	if strings.Contains(wire, "IORateValidity") {
+		t.Fatalf("internal validity leaked into wire payload: %s", wire)
+	}
 }
@@ -162,6 +162,7 @@ type VM struct {
 	NetworkOut         int64                   `json:"networkOut"`
 	DiskRead           int64                   `json:"diskRead"`
 	DiskWrite          int64                   `json:"diskWrite"`
+	IORateValidity     IORateValidity          `json:"-"`
 	Uptime             int64                   `json:"uptime"`
 	Template           bool                    `json:"template"`
 	OnBoot             *bool                   `json:"onBoot,omitempty"`
@@ -209,6 +210,7 @@ type Container struct {
 	NetworkOut        int64                   `json:"networkOut"`
 	DiskRead          int64                   `json:"diskRead"`
 	DiskWrite         int64                   `json:"diskWrite"`
+	IORateValidity    IORateValidity          `json:"-"`
 	Uptime            int64                   `json:"uptime"`
 	Template          bool                    `json:"template"`
 	OnBoot            *bool                   `json:"onBoot,omitempty"`
@@ -1819,7 +1819,7 @@ func TestProxmoxGuestPollersCarryPoolIntoCanonicalModels(t *testing.T) {
 	requiredSnippets := map[string][]string{
 		"monitor_pve_guest_builders.go":  {"Pool:     strings.TrimSpace(res.Pool)"},
 		"monitor_pve_guest_lxc.go":       {"Pool:     strings.TrimSpace(res.Pool)"},
-		"monitor_pve_node_vm_builder.go": {"Pool:      vm.Pool"},
+		"monitor_pve_node_vm_builder.go": {"Pool:       vm.Pool"},
 		"monitor_polling_containers.go":  {"Pool:     strings.TrimSpace(container.Pool)"},
 	}

@@ -2021,14 +2021,16 @@ func TestMockNativePollersDeferToCanonicalMockSampler(t *testing.T) {
 			file: "monitor_polling_vm.go",
 			snippets: []string{
 				"if !shouldSkipNativeMockStateMetricWrites() {",
-				`m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, -1, -1, -1, -1, now)`,
+				"guestHistoryRates(",
+				"m.recordGuestMetric(",
 			},
 		},
 		{
 			file: "monitor_polling_containers.go",
 			snippets: []string{
 				"if !shouldSkipNativeMockStateMetricWrites() {",
-				`m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), historyMemoryUsage(ct.Memory), ct.Disk.Usage, -1, -1, -1, -1, now)`,
+				"guestHistoryRates(",
+				"m.recordGuestMetric(",
 			},
 		},
 		{
@@ -0,0 +1,180 @@
+package monitoring
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
+)
+
+func TestIssue1613GuestRateIdentitySurvivesMigrationAndIsolatesDuplicates(t *testing.T) {
+	beforeMigration := makeGuestRateKey("site-a", "qemu", 100)
+	afterMigration := makeGuestRateKey("site-a", "qemu", 100)
+	if beforeMigration != afterMigration {
+		t.Fatalf("migration changed rate key: %q != %q", beforeMigration, afterMigration)
+	}
+	if beforeMigration == makeGuestRateKey("site-b", "qemu", 100) {
+		t.Fatal("duplicate cluster registrations shared a rate key")
+	}
+	if beforeMigration == makeGuestRateKey("site-a", "lxc", 100) {
+		t.Fatal("QEMU and LXC with the same VMID shared a rate key")
+	}
+}
+
+func TestIssue1613QEMUStatusUsesPresentLowerCounterAndKeepsMissingListingCounter(t *testing.T) {
+	listingAt := time.Unix(1_700_000_000, 0)
+	statusAt := listingAt.Add(3 * time.Second)
+	state := vmBuildState{
+		diskReadBytes:     9_000,
+		diskWriteBytes:    8_000,
+		networkInBytes:    7_000,
+		networkOutBytes:   6_000,
+		counterPresence:   models.IOCounterPresence{Explicit: true, DiskRead: true, DiskWrite: true, NetworkIn: true, NetworkOut: true},
+		counterObservedAt: listingAt,
+		counterObservationTimes: models.IOCounterObservationTimes{
+			DiskRead:   listingAt,
+			DiskWrite:  listingAt,
+			NetworkIn:  listingAt,
+			NetworkOut: listingAt,
+		},
+		counterUptime: 10_000,
+	}
+	mergeVMRuntimeCounters(&state, &proxmox.VMStatus{
+		DiskRead: 0,
+		NetOut:   500,
+		Uptime:   20,
+		IOCounters: proxmox.IOCounterPresence{
+			Explicit:   true,
+			DiskRead:   true,
+			NetworkOut: true,
+		},
+		ObservedAt: statusAt,
+	})
+
+	if state.diskReadBytes != 0 || state.networkOutBytes != 500 {
+		t.Fatalf("present status counters did not replace listing values: %+v", state)
+	}
+	if state.diskWriteBytes != 8_000 || state.networkInBytes != 7_000 {
+		t.Fatalf("missing status counters replaced listing values: %+v", state)
+	}
+	if !state.counterObservationTimes.DiskRead.Equal(statusAt) ||
+		!state.counterObservationTimes.NetworkOut.Equal(statusAt) ||
+		!state.counterObservationTimes.DiskWrite.Equal(listingAt) ||
+		!state.counterObservationTimes.NetworkIn.Equal(listingAt) {
+		t.Fatalf("per-counter receipt authority was not preserved: %+v", state.counterObservationTimes)
+	}
+	if state.counterUptime != 20 {
+		t.Fatalf("status uptime = %d, want 20", state.counterUptime)
+	}
+}
+
+func TestIssue1613PVEGraceTracksConfiguredPollingCadence(t *testing.T) {
+	for _, test := range []struct {
+		interval time.Duration
+		want     time.Duration
+	}{
+		{interval: 30 * time.Second, want: 60 * time.Second},
+		{interval: 60 * time.Second, want: 120 * time.Second},
+		{interval: 90 * time.Second, want: 180 * time.Second},
+	} {
+		monitor := &Monitor{config: &config.Config{PVEPollingInterval: test.interval}}
+		if got := monitor.pveNodeOfflineGracePeriod(); got != test.want {
+			t.Fatalf("interval %s grace = %s, want %s", test.interval, got, test.want)
+		}
+	}
+}
+
+func TestIssue1613HistoryWritesValidZeroAndSkipsUnknownRates(t *testing.T) {
+	monitor := &Monitor{metricsHistory: NewMetricsHistory(32, time.Hour)}
+	now := time.Now()
+
+	diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(0, 0, 0, 0, models.IORateValidity{Explicit: true})
+	monitor.recordGuestMetric("vm", "unknown", 0, 0, 0, diskRead, diskWrite, networkIn, networkOut, now)
+	if points := monitor.metricsHistory.GetGuestMetrics("unknown", "diskread", time.Hour); len(points) != 0 {
+		t.Fatalf("unknown disk read rate was written to history: %+v", points)
+	}
+
+	diskRead, diskWrite, networkIn, networkOut = guestHistoryRates(0, 0, 0, 0, models.IORateValidity{
+		Explicit:   true,
+		DiskRead:   true,
+		DiskWrite:  true,
+		NetworkIn:  true,
+		NetworkOut: true,
+	})
+	monitor.recordGuestMetric("vm", "idle", 0, 0, 0, diskRead, diskWrite, networkIn, networkOut, now)
+	points := monitor.metricsHistory.GetGuestMetrics("idle", "diskread", time.Hour)
+	if len(points) != 1 || points[0].Value != 0 {
+		t.Fatalf("valid idle disk read rate was not written as zero: %+v", points)
+	}
+}
+
+func TestIssue1613NodeDoesNotGreyBetweenNinetySecondPolls(t *testing.T) {
+	instance := &config.PVEInstance{Name: "site-a"}
+	monitor := &Monitor{
+		config:         &config.Config{PVEPollingInterval: 90 * time.Second},
+		nodeLastOnline: map[string]time.Time{"site-a-pve-a": time.Now().Add(-100 * time.Second)},
+	}
+
+	_, status := monitor.determineNodeIDAndStatus("site-a", instance, proxmox.Node{
+		Node:   "pve-a",
+		Status: "offline",
+	})
+	if status != "online" {
+		t.Fatalf("node status = %q inside cadence grace, want online", status)
+	}
+
+	monitor.nodeLastOnline["site-a-pve-a"] = time.Now().Add(-181 * time.Second)
+	_, status = monitor.determineNodeIDAndStatus("site-a", instance, proxmox.Node{
+		Node:   "pve-a",
+		Status: "offline",
+	})
+	if status != "offline" {
+		t.Fatalf("node status = %q after cadence grace, want offline", status)
+	}
+}
+
+func TestIssue1613WebsocketStateKeepsUnknownRatesNumeric(t *testing.T) {
+	monitor := &Monitor{
+		state: models.NewState(),
+		resourceStore: &resourceOnlyStore{resources: []unifiedresources.Resource{
+			{
+				ID:      "site-a:pve-a:100",
+				Type:    unifiedresources.ResourceTypeVM,
+				Name:    "idle-vm",
+				Status:  unifiedresources.StatusOnline,
+				Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox},
+				Metrics: &unifiedresources.ResourceMetrics{
+					DiskRead: &unifiedresources.MetricValue{Value: 0, Unit: "bytes/s"},
+				},
+			},
+			{
+				ID:      "site-a:pve-a:101",
+				Type:    unifiedresources.ResourceTypeVM,
+				Name:    "unknown-vm",
+				Status:  unifiedresources.StatusOnline,
+				Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox},
+				Metrics: &unifiedresources.ResourceMetrics{},
+			},
+		}},
+	}
+	frontend := monitor.BuildBroadcastFrontendState()
+	payload, err := json.Marshal(frontend)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wire := string(payload)
+	if !strings.Contains(wire, `"diskIO":{"readRate":0,"writeRate":0}`) {
+		t.Fatalf("websocket payload does not contain numeric valid zero disk rate: %s", wire)
+	}
+	if strings.Count(wire, `"diskIO"`) != 1 {
+		t.Fatalf("unknown rate projected a disk I/O object: %s", wire)
+	}
+	if strings.Contains(wire, `"diskWrite":null`) || strings.Contains(wire, `"netIn":null`) {
+		t.Fatalf("websocket payload contains unstable null I/O values: %s", wire)
+	}
+}
@@ -146,7 +146,8 @@ func TestMockVMPollingDefersMemoryHistoryToCanonicalSampler(t *testing.T) {

 	requiredSnippets := []string{
 		"if !shouldSkipNativeMockStateMetricWrites() {",
-		`m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, -1, -1, -1, -1, now)`,
+		"guestHistoryRates(",
+		"m.recordGuestMetric(",
 	}
 	for _, snippet := range requiredSnippets {
 		if !strings.Contains(source, snippet) {
@@ -12,6 +12,7 @@ import (
 	agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
 	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
 	"github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory"
+	"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
 	"github.com/rs/zerolog/log"
 )

@@ -62,6 +63,62 @@ func makeGuestID(instanceName string, node string, vmid int) string {
 	return fmt.Sprintf("%s:%s:%d", instanceName, node, vmid)
 }

+// makeGuestRateKey identifies one counter stream independently of placement.
+// A live migration changes the canonical row ID's node coordinate but not the
+// cumulative counter epoch. The configured instance remains in the key so
+// duplicate cluster registrations never race on one baseline.
+func makeGuestRateKey(instanceName, guestType string, vmid int) string {
+	return fmt.Sprintf("pve:%s:%s:%d", instanceName, strings.ToLower(strings.TrimSpace(guestType)), vmid)
+}
+
+func pveCounterPresence(p proxmox.IOCounterPresence) models.IOCounterPresence {
+	effective := p.Effective()
+	return models.IOCounterPresence{
+		Explicit:   true,
+		DiskRead:   effective.DiskRead,
+		DiskWrite:  effective.DiskWrite,
+		NetworkIn:  effective.NetworkIn,
+		NetworkOut: effective.NetworkOut,
+	}
+}
+
+func observedAtOr(value, fallback time.Time) time.Time {
+	if !value.IsZero() {
+		return value
+	}
+	return fallback
+}
+
+func counterObservationTimes(observedAt time.Time) models.IOCounterObservationTimes {
+	return models.IOCounterObservationTimes{
+		DiskRead:   observedAt,
+		DiskWrite:  observedAt,
+		NetworkIn:  observedAt,
+		NetworkOut: observedAt,
+	}
+}
+
+func numericGuestRate(rate float64) (int64, bool) {
+	if rate < 0 || math.IsNaN(rate) || math.IsInf(rate, 0) {
+		return 0, false
+	}
+	return max(0, int64(rate)), true
+}
+
+func guestRateValues(diskRead, diskWrite, networkIn, networkOut float64) (int64, int64, int64, int64, models.IORateValidity) {
+	diskReadValue, diskReadKnown := numericGuestRate(diskRead)
+	diskWriteValue, diskWriteKnown := numericGuestRate(diskWrite)
+	networkInValue, networkInKnown := numericGuestRate(networkIn)
+	networkOutValue, networkOutKnown := numericGuestRate(networkOut)
+	return diskReadValue, diskWriteValue, networkInValue, networkOutValue, models.IORateValidity{
+		Explicit:   true,
+		DiskRead:   diskReadKnown,
+		DiskWrite:  diskWriteKnown,
+		NetworkIn:  networkInKnown,
+		NetworkOut: networkOutKnown,
+	}
+}
+
 // parseBoolEnv parses a boolean from an environment variable, returning defaultVal if not set or invalid
 func parseBoolEnv(key string, defaultVal bool) bool {
 	val := strings.TrimSpace(os.Getenv(key))
@@ -65,6 +65,7 @@ func TestMonitor_PollPBSInstance_Fallback_Extra(t *testing.T) {
 		},
 		state:            models.NewState(),
 		stalenessTracker: NewStalenessTracker(nil),
+		pollStatusMap:    make(map[string]*pollStatus),
 	}

 	ctx := context.Background()
@@ -105,12 +105,16 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 				guestID := makeGuestID(instanceName, n.Node, int(container.VMID))

 				sampleTime := time.Now()
+				counterObservedAt := observedAtOr(container.ObservedAt, sampleTime)
 				currentMetrics := IOMetrics{
-					DiskRead:   int64(container.DiskRead),
-					DiskWrite:  int64(container.DiskWrite),
-					NetworkIn:  int64(container.NetIn),
-					NetworkOut: int64(container.NetOut),
-					Timestamp:  sampleTime,
+					DiskRead:     int64(container.DiskRead),
+					DiskWrite:    int64(container.DiskWrite),
+					NetworkIn:    int64(container.NetIn),
+					NetworkOut:   int64(container.NetOut),
+					Timestamp:    counterObservedAt,
+					Presence:     pveCounterPresence(container.IOCounters),
+					ObservedAt:   counterObservationTimes(counterObservedAt),
+					SourceUptime: uint64(container.Uptime),
 				}
 				statusSnapshot := (*proxmox.Container)(nil)
 				if container.Status == "running" {
@@ -124,7 +128,16 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 					)
 					currentMetrics = mergeContainerRuntimeCounters(currentMetrics, statusSnapshot)
 				}
-				diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics)
+				diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(
+					makeGuestRateKey(instanceName, "lxc", int(container.VMID)),
+					currentMetrics,
+				)
+				diskReadValue, diskWriteValue, networkInValue, networkOutValue, rateValidity := guestRateValues(
+					diskReadRate,
+					diskWriteRate,
+					netInRate,
+					netOutRate,
+				)

 				// Set CPU to 0 for non-running containers
 				cpuUsage := safeFloat(container.CPU)
@@ -201,14 +214,15 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 						Free:  diskFreeBytes,
 						Usage: diskUsagePercent,
 					},
-					NetworkIn:  max(0, int64(netInRate)),
-					NetworkOut: max(0, int64(netOutRate)),
-					DiskRead:   max(0, int64(diskReadRate)),
-					DiskWrite:  max(0, int64(diskWriteRate)),
-					Uptime:     int64(container.Uptime),
-					Template:   container.Template == 1,
-					LastSeen:   sampleTime,
-					Tags:       tags,
+					NetworkIn:      networkInValue,
+					NetworkOut:     networkOutValue,
+					DiskRead:       diskReadValue,
+					DiskWrite:      diskWriteValue,
+					IORateValidity: rateValidity,
+					Uptime:         int64(container.Uptime),
+					Template:       container.Template == 1,
+					LastSeen:       sampleTime,
+					Tags:           tags,
 				}

 				if prevContainerIsOCI[modelContainer.VMID] {
@@ -258,6 +272,13 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 					modelContainer.NetworkOut = 0
 					modelContainer.DiskRead = 0
 					modelContainer.DiskWrite = 0
+					modelContainer.IORateValidity = models.IORateValidity{
+						Explicit:   true,
+						DiskRead:   true,
+						DiskWrite:  true,
+						NetworkIn:  true,
+						NetworkOut: true,
+					}
 				}

 				// Trigger guest metadata migration if old format exists
@@ -349,9 +370,25 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s
 			if ct.Status != "running" {
 				continue
 			}
-			// IO/network series are not recorded on the traditional polling
-			// path (parity with the historical inline writes).
-			m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), historyMemoryUsage(ct.Memory), ct.Disk.Usage, -1, -1, -1, -1, now)
+			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),
+				ct.Disk.Usage,
+				diskRead,
+				diskWrite,
+				networkIn,
+				networkOut,
+				now,
+			)
 		}
 	}

@@ -47,6 +47,7 @@ func (m *Monitor) determineNodeIDAndStatus(instanceName string, instanceCfg *con
 	}
 	effectiveStatus := node.Status
 	now := time.Now()
+	gracePeriod := m.pveNodeOfflineGracePeriod()

 	m.mu.Lock()
 	if strings.ToLower(node.Status) == "online" {
@@ -55,14 +56,14 @@ func (m *Monitor) determineNodeIDAndStatus(instanceName string, instanceCfg *con
 	} else {
 		// Node is reported as offline - check grace period
 		lastOnline, exists := m.nodeLastOnline[nodeID]
-		if exists && now.Sub(lastOnline) < nodeOfflineGracePeriod {
+		if exists && now.Sub(lastOnline) < gracePeriod {
 			// Still within grace period - preserve online status
 			effectiveStatus = "online"
 			log.Debug().
 				Str("instance", instanceName).
 				Str("node", node.Node).
 				Dur("timeSinceOnline", now.Sub(lastOnline)).
-				Dur("gracePeriod", nodeOfflineGracePeriod).
+				Dur("gracePeriod", gracePeriod).
 				Msg("Node offline but within grace period - preserving online status")
 		} else {
 			// Grace period expired or never seen online - mark as offline
@@ -126,9 +126,25 @@ func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string,
 			if vm.Status != "running" {
 				continue
 			}
-			// IO/network series are not recorded on the traditional polling
-			// path (parity with the historical inline writes).
-			m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, -1, -1, -1, -1, now)
+			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),
+				vm.Disk.Usage,
+				diskRead,
+				diskWrite,
+				networkIn,
+				networkOut,
+				now,
+			)
 		}
 	}

@@ -63,10 +63,35 @@ func mergeContainerRuntimeCounters(current IOMetrics, status *proxmox.Container)
 		return current
 	}

-	current.DiskRead = max(current.DiskRead, int64(status.DiskRead))
-	current.DiskWrite = max(current.DiskWrite, int64(status.DiskWrite))
-	current.NetworkIn = max(current.NetworkIn, int64(status.NetIn))
-	current.NetworkOut = max(current.NetworkOut, int64(status.NetOut))
+	currentPresence := current.Presence.Effective()
+	statusPresence := status.IOCounters.Effective()
+	if statusPresence.DiskRead {
+		current.DiskRead = int64(status.DiskRead)
+		currentPresence.DiskRead = true
+		current.ObservedAt.DiskRead = status.ObservedAt
+	}
+	if statusPresence.DiskWrite {
+		current.DiskWrite = int64(status.DiskWrite)
+		currentPresence.DiskWrite = true
+		current.ObservedAt.DiskWrite = status.ObservedAt
+	}
+	if statusPresence.NetworkIn {
+		current.NetworkIn = int64(status.NetIn)
+		currentPresence.NetworkIn = true
+		current.ObservedAt.NetworkIn = status.ObservedAt
+	}
+	if statusPresence.NetworkOut {
+		current.NetworkOut = int64(status.NetOut)
+		currentPresence.NetworkOut = true
+		current.ObservedAt.NetworkOut = status.ObservedAt
+	}
+	current.Presence = currentPresence
+	if !status.ObservedAt.IsZero() {
+		current.Timestamp = status.ObservedAt
+	}
+	if status.Uptime > 0 {
+		current.SourceUptime = status.Uptime
+	}
 	return current
 }

@@ -712,6 +737,7 @@ func (m *Monitor) placeholderNodesForInstance(instanceName string) []models.Node
 func (m *Monitor) preserveOrExpireNodes(prevInstanceNodes []models.Node) []models.Node {
 	preserved := make([]models.Node, 0, len(prevInstanceNodes))
 	now := time.Now()
+	gracePeriod := m.pveNodeOfflineGracePeriod()
 	for _, prevNode := range prevInstanceNodes {
 		nodeCopy := prevNode

@@ -729,9 +755,9 @@ func (m *Monitor) preserveOrExpireNodes(prevInstanceNodes []models.Node) []model
 		lastOnline, sawOnline := m.nodeLastOnline[prevNode.ID]
 		m.mu.Unlock()

-		withinGrace := sawOnline && now.Sub(lastOnline) < nodeOfflineGracePeriod
+		withinGrace := sawOnline && now.Sub(lastOnline) < gracePeriod
 		if !withinGrace && strings.EqualFold(strings.TrimSpace(prevNode.Status), "online") {
-			withinGrace = !prevNode.LastSeen.IsZero() && now.Sub(prevNode.LastSeen) < nodeOfflineGracePeriod
+			withinGrace = !prevNode.LastSeen.IsZero() && now.Sub(prevNode.LastSeen) < gracePeriod
 		}
 		if withinGrace {
 			if strings.TrimSpace(nodeCopy.Status) == "" || strings.EqualFold(nodeCopy.Status, "offline") {
@@ -13,30 +13,34 @@ import (
 )

 type vmBuildState struct {
-	memTotal           uint64
-	memUsed            uint64
-	memorySource       string
-	guestRaw           VMMemoryRaw
-	diskReadBytes      int64
-	diskWriteBytes     int64
-	networkInBytes     int64
-	networkOutBytes    int64
-	diskTotal          uint64
-	diskUsed           uint64
-	diskFree           uint64
-	diskUsage          float64
-	diskFromAgent      bool
-	diskStatusReason   string
-	guestAgentStatus   string
-	guestAgentExpected bool
-	individualDisks    []models.Disk
-	ipAddresses        []string
-	networkInterfaces  []models.GuestNetworkInterface
-	osName             string
-	osVersion          string
-	agentVersion       string
-	detailedStatus     *proxmox.VMStatus
-	onBoot             *bool
+	memTotal                uint64
+	memUsed                 uint64
+	memorySource            string
+	guestRaw                VMMemoryRaw
+	diskReadBytes           int64
+	diskWriteBytes          int64
+	networkInBytes          int64
+	networkOutBytes         int64
+	counterPresence         models.IOCounterPresence
+	counterObservedAt       time.Time
+	counterObservationTimes models.IOCounterObservationTimes
+	counterUptime           uint64
+	diskTotal               uint64
+	diskUsed                uint64
+	diskFree                uint64
+	diskUsage               float64
+	diskFromAgent           bool
+	diskStatusReason        string
+	guestAgentStatus        string
+	guestAgentExpected      bool
+	individualDisks         []models.Disk
+	ipAddresses             []string
+	networkInterfaces       []models.GuestNetworkInterface
+	osName                  string
+	osVersion               string
+	agentVersion            string
+	detailedStatus          *proxmox.VMStatus
+	onBoot                  *bool
 }

 func (m *Monitor) applyVMStatusDetails(
@@ -71,11 +75,7 @@ func (m *Monitor) applyVMStatusDetails(
 		&state.guestRaw,
 	)

-	// Use actual disk I/O values from detailed status
-	state.diskReadBytes = int64(status.DiskRead)
-	state.diskWriteBytes = int64(status.DiskWrite)
-	state.networkInBytes = int64(status.NetIn)
-	state.networkOutBytes = int64(status.NetOut)
+	mergeVMRuntimeCounters(state, status)

 	// Gather guest metadata from the agent when available
 	guestIPs, guestIfaces, guestOSName, guestOSVersion, guestAgentVersion := m.fetchGuestAgentMetadata(ctx, client, instanceName, res.Node, res.Name, res.VMID, status, false)
@@ -132,6 +132,42 @@ func (m *Monitor) applyVMStatusDetails(

 }

+func mergeVMRuntimeCounters(state *vmBuildState, status *proxmox.VMStatus) {
+	if state == nil || status == nil {
+		return
+	}
+
+	// A detailed status response is newer and more authoritative than the
+	// cluster listing, but only for fields it actually contains.
+	statusPresence := status.IOCounters.Effective()
+	if statusPresence.DiskRead {
+		state.diskReadBytes = int64(status.DiskRead)
+		state.counterPresence.DiskRead = true
+		state.counterObservationTimes.DiskRead = status.ObservedAt
+	}
+	if statusPresence.DiskWrite {
+		state.diskWriteBytes = int64(status.DiskWrite)
+		state.counterPresence.DiskWrite = true
+		state.counterObservationTimes.DiskWrite = status.ObservedAt
+	}
+	if statusPresence.NetworkIn {
+		state.networkInBytes = int64(status.NetIn)
+		state.counterPresence.NetworkIn = true
+		state.counterObservationTimes.NetworkIn = status.ObservedAt
+	}
+	if statusPresence.NetworkOut {
+		state.networkOutBytes = int64(status.NetOut)
+		state.counterPresence.NetworkOut = true
+		state.counterObservationTimes.NetworkOut = status.ObservedAt
+	}
+	if !status.ObservedAt.IsZero() {
+		state.counterObservedAt = status.ObservedAt
+	}
+	if status.Uptime > 0 {
+		state.counterUptime = status.Uptime
+	}
+}
+
 func vmGuestAgentRuntimeState(status *proxmox.VMStatus, recentGuestAgentEvidence bool) (string, bool) {
 	if status == nil {
 		return "", false
@@ -171,16 +207,20 @@ func (m *Monitor) buildVMFromClusterResource(
 		m.hasRecentGuestMetadataEvidence(instanceName, res.Node, res.VMID, prePollTime)

 	state := vmBuildState{
-		memTotal:        res.MaxMem,
-		memUsed:         res.Mem,
-		memorySource:    "cluster-resources",
-		guestRaw:        VMMemoryRaw{ListingMem: res.Mem, ListingMaxMem: res.MaxMem},
-		diskReadBytes:   int64(res.DiskRead),
-		diskWriteBytes:  int64(res.DiskWrite),
-		networkInBytes:  int64(res.NetIn),
-		networkOutBytes: int64(res.NetOut),
-		diskTotal:       res.MaxDisk,
-		diskUsed:        res.Disk,
+		memTotal:                res.MaxMem,
+		memUsed:                 res.Mem,
+		memorySource:            "cluster-resources",
+		guestRaw:                VMMemoryRaw{ListingMem: res.Mem, ListingMaxMem: res.MaxMem},
+		diskReadBytes:           int64(res.DiskRead),
+		diskWriteBytes:          int64(res.DiskWrite),
+		networkInBytes:          int64(res.NetIn),
+		networkOutBytes:         int64(res.NetOut),
+		counterPresence:         pveCounterPresence(res.IOCounters),
+		counterObservedAt:       observedAtOr(res.ObservedAt, prePollTime),
+		counterObservationTimes: counterObservationTimes(observedAtOr(res.ObservedAt, prePollTime)),
+		counterUptime:           res.Uptime,
+		diskTotal:               res.MaxDisk,
+		diskUsed:                res.Disk,
 	}
 	state.diskFree = state.diskTotal - state.diskUsed
 	state.diskUsage = safePercentage(float64(state.diskUsed), float64(state.diskTotal))
@@ -339,13 +379,25 @@ func (m *Monitor) buildVMFromClusterResource(
 		memFree = state.memTotal - state.memUsed
 	}
 	currentMetrics := IOMetrics{
-		DiskRead:   state.diskReadBytes,
-		DiskWrite:  state.diskWriteBytes,
-		NetworkIn:  state.networkInBytes,
-		NetworkOut: state.networkOutBytes,
-		Timestamp:  sampleTime,
+		DiskRead:     state.diskReadBytes,
+		DiskWrite:    state.diskWriteBytes,
+		NetworkIn:    state.networkInBytes,
+		NetworkOut:   state.networkOutBytes,
+		Timestamp:    state.counterObservedAt,
+		Presence:     state.counterPresence,
+		ObservedAt:   state.counterObservationTimes,
+		SourceUptime: state.counterUptime,
 	}
-	diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics)
+	diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(
+		makeGuestRateKey(instanceName, "qemu", res.VMID),
+		currentMetrics,
+	)
+	diskReadValue, diskWriteValue, networkInValue, networkOutValue, rateValidity := guestRateValues(
+		diskReadRate,
+		diskWriteRate,
+		netInRate,
+		netOutRate,
+	)

 	memory := models.UnavailableMemory(clampToInt64(state.memTotal))
 	if CanonicalMemorySource(state.memorySource) != "unavailable" {
@@ -403,10 +455,11 @@ func (m *Monitor) buildVMFromClusterResource(
 		OSVersion:          state.osVersion,
 		AgentVersion:       state.agentVersion,
 		NetworkInterfaces:  state.networkInterfaces,
-		NetworkIn:          max(0, int64(netInRate)),
-		NetworkOut:         max(0, int64(netOutRate)),
-		DiskRead:           max(0, int64(diskReadRate)),
-		DiskWrite:          max(0, int64(diskWriteRate)),
+		NetworkIn:          networkInValue,
+		NetworkOut:         networkOutValue,
+		DiskRead:           diskReadValue,
+		DiskWrite:          diskWriteValue,
+		IORateValidity:     rateValidity,
 		Uptime:             int64(res.Uptime),
 		Template:           res.Template == 1,
 		OnBoot:             state.onBoot,
@@ -429,6 +482,19 @@ func (m *Monitor) buildVMFromClusterResource(
 			}
 		}
 	}
+	if res.Status != "running" {
+		vm.NetworkIn = 0
+		vm.NetworkOut = 0
+		vm.DiskRead = 0
+		vm.DiskWrite = 0
+		vm.IORateValidity = models.IORateValidity{
+			Explicit:   true,
+			DiskRead:   true,
+			DiskWrite:  true,
+			NetworkIn:  true,
+			NetworkOut: true,
+		}
+	}

 	return vm, state.guestRaw, state.memorySource, snapshotNotes, sampleTime, true
 }
@@ -110,16 +110,35 @@ func (m *Monitor) recordGuestMetrics(allVMs []models.VM, allContainers []models.
 	now := time.Now()
 	for _, vm := range allVMs {
 		if vm.Status == "running" {
-			m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, vm.DiskRead, vm.DiskWrite, vm.NetworkIn, vm.NetworkOut, now)
+			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), vm.Disk.Usage, diskRead, diskWrite, networkIn, networkOut, now)
 		}
 	}
 	for _, ct := range allContainers {
 		if ct.Status == "running" {
-			m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), historyMemoryUsage(ct.Memory), ct.Disk.Usage, ct.DiskRead, ct.DiskWrite, ct.NetworkIn, ct.NetworkOut, now)
+			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), ct.Disk.Usage, diskRead, diskWrite, networkIn, networkOut, now)
 		}
 	}
 }

+func guestHistoryRates(diskRead, diskWrite, networkIn, networkOut int64, validity models.IORateValidity) (int64, int64, int64, int64) {
+	effective := validity.EffectiveForRates(diskRead, diskWrite, networkIn, networkOut)
+	if !effective.DiskRead {
+		diskRead = -1
+	}
+	if !effective.DiskWrite {
+		diskWrite = -1
+	}
+	if !effective.NetworkIn {
+		networkIn = -1
+	}
+	if !effective.NetworkOut {
+		networkOut = -1
+	}
+	return diskRead, diskWrite, networkIn, networkOut
+}
+
 func historyMemoryUsage(memory models.Memory) float64 {
 	if !memory.HasKnownUsage() {
 		return -1
@@ -101,12 +101,16 @@ func (m *Monitor) buildContainerFromClusterResource(
 	guestID := makeGuestID(instanceName, res.Node, res.VMID)

 	sampleTime := time.Now()
+	counterObservedAt := observedAtOr(res.ObservedAt, sampleTime)
 	currentMetrics := IOMetrics{
-		DiskRead:   int64(res.DiskRead),
-		DiskWrite:  int64(res.DiskWrite),
-		NetworkIn:  int64(res.NetIn),
-		NetworkOut: int64(res.NetOut),
-		Timestamp:  sampleTime,
+		DiskRead:     int64(res.DiskRead),
+		DiskWrite:    int64(res.DiskWrite),
+		NetworkIn:    int64(res.NetIn),
+		NetworkOut:   int64(res.NetOut),
+		Timestamp:    counterObservedAt,
+		Presence:     pveCounterPresence(res.IOCounters),
+		ObservedAt:   counterObservationTimes(counterObservedAt),
+		SourceUptime: res.Uptime,
 	}
 	statusSnapshot := (*proxmox.Container)(nil)
 	if res.Status == "running" {
@@ -120,7 +124,16 @@ func (m *Monitor) buildContainerFromClusterResource(
 		)
 		currentMetrics = mergeContainerRuntimeCounters(currentMetrics, statusSnapshot)
 	}
-	diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics)
+	diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(
+		makeGuestRateKey(instanceName, "lxc", res.VMID),
+		currentMetrics,
+	)
+	diskReadValue, diskWriteValue, networkInValue, networkOutValue, rateValidity := guestRateValues(
+		diskReadRate,
+		diskWriteRate,
+		netInRate,
+		netOutRate,
+	)

 	memTotal, memUsed, memorySource, guestRaw := m.calculateLXCMemory(ctx, instanceName, res, client)
 	memUsed, memorySource, _ = stabilizeGuestLowTrustMemory(
@@ -181,13 +194,14 @@ func (m *Monitor) buildContainerFromClusterResource(
 			Free:  diskFree,
 			Usage: safePercentage(float64(diskUsed), float64(res.MaxDisk)),
 		},
-		NetworkIn:  max(0, int64(netInRate)),
-		NetworkOut: max(0, int64(netOutRate)),
-		DiskRead:   max(0, int64(diskReadRate)),
-		DiskWrite:  max(0, int64(diskWriteRate)),
-		Uptime:     int64(res.Uptime),
-		Template:   res.Template == 1,
-		LastSeen:   lastSeen,
+		NetworkIn:      networkInValue,
+		NetworkOut:     networkOutValue,
+		DiskRead:       diskReadValue,
+		DiskWrite:      diskWriteValue,
+		IORateValidity: rateValidity,
+		Uptime:         int64(res.Uptime),
+		Template:       res.Template == 1,
+		LastSeen:       lastSeen,
 	}

 	if prevContainerIsOCI[container.VMID] {
@@ -231,6 +245,13 @@ func (m *Monitor) buildContainerFromClusterResource(
 		container.NetworkOut = 0
 		container.DiskRead = 0
 		container.DiskWrite = 0
+		container.IORateValidity = models.IORateValidity{
+			Explicit:   true,
+			DiskRead:   true,
+			DiskWrite:  true,
+			NetworkIn:  true,
+			NetworkOut: true,
+		}
 	}

 	return container, guestRaw, memorySource, sampleTime, true
@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"

+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
 	"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
 )

@@ -20,7 +21,7 @@ func (s *stubPVEClientLXCStatus) GetContainerStatus(ctx context.Context, node st
 	return s.containerStatus, nil
 }

-func TestMergeContainerRuntimeCounters_PrefersHigherStatusCounters(t *testing.T) {
+func TestMergeContainerRuntimeCounters_PrefersNewerStatusCounters(t *testing.T) {
 	t.Parallel()

 	current := IOMetrics{
@@ -41,17 +42,66 @@ func TestMergeContainerRuntimeCounters_PrefersHigherStatusCounters(t *testing.T)
 	if merged.DiskRead != 128 {
 		t.Fatalf("expected DiskRead to upgrade from status snapshot, got %d", merged.DiskRead)
 	}
-	if merged.DiskWrite != 8 {
-		t.Fatalf("expected DiskWrite to preserve the higher baseline counter, got %d", merged.DiskWrite)
+	if merged.DiskWrite != 4 {
+		t.Fatalf("expected DiskWrite to follow the newer reset counter, got %d", merged.DiskWrite)
 	}
-	if merged.NetworkIn != 12 {
-		t.Fatalf("expected NetworkIn to preserve the higher baseline counter, got %d", merged.NetworkIn)
+	if merged.NetworkIn != 10 {
+		t.Fatalf("expected NetworkIn to follow the newer reset counter, got %d", merged.NetworkIn)
 	}
 	if merged.NetworkOut != 256 {
 		t.Fatalf("expected NetworkOut to upgrade from status snapshot, got %d", merged.NetworkOut)
 	}
 }

+func TestMergeContainerRuntimeCounters_OverridesOnlyPresentStatusFields(t *testing.T) {
+	t.Parallel()
+
+	listingObservedAt := time.Unix(10, 0)
+	current := IOMetrics{
+		DiskRead:   8,
+		DiskWrite:  16,
+		NetworkIn:  32,
+		NetworkOut: 64,
+		Timestamp:  listingObservedAt,
+		ObservedAt: counterObservationTimes(listingObservedAt),
+		Presence: models.IOCounterPresence{
+			Explicit:   true,
+			DiskRead:   true,
+			DiskWrite:  true,
+			NetworkIn:  true,
+			NetworkOut: true,
+		},
+	}
+	status := &proxmox.Container{
+		DiskRead:  0,
+		DiskWrite: 999,
+		IOCounters: proxmox.IOCounterPresence{
+			Explicit: true,
+			DiskRead: true,
+		},
+		ObservedAt: time.Unix(20, 0),
+	}
+
+	merged := mergeContainerRuntimeCounters(current, status)
+	if merged.DiskRead != 0 {
+		t.Fatalf("explicit status zero was not authoritative: %d", merged.DiskRead)
+	}
+	if merged.DiskWrite != 16 || merged.NetworkIn != 32 || merged.NetworkOut != 64 {
+		t.Fatalf("missing status fields overwrote listing counters: %+v", merged)
+	}
+	if !merged.Timestamp.Equal(status.ObservedAt) {
+		t.Fatalf("timestamp = %v, want status receipt time %v", merged.Timestamp, status.ObservedAt)
+	}
+	if !merged.ObservedAt.DiskRead.Equal(status.ObservedAt) {
+		t.Fatalf("disk-read receipt = %v, want status receipt %v", merged.ObservedAt.DiskRead, status.ObservedAt)
+	}
+	if !merged.ObservedAt.DiskWrite.Equal(listingObservedAt) ||
+		!merged.ObservedAt.NetworkIn.Equal(listingObservedAt) ||
+		!merged.ObservedAt.NetworkOut.Equal(listingObservedAt) {
+		t.Fatalf("missing status fields lost listing receipt times: %+v", merged.ObservedAt)
+	}
+}
+
 func TestBuildContainerFromClusterResource_UsesContainerStatusCountersForRates(t *testing.T) {
 	t.Parallel()

@@ -31,26 +31,28 @@ func (m *Monitor) pollNodeVMsWithClusterResourceBuilder(
 		resources = append(resources, indexedClusterResource{
 			order: len(resources),
 			resource: proxmox.ClusterResource{
-				ID:        guestID,
-				Type:      "qemu",
-				Node:      node,
-				Pool:      vm.Pool,
-				Status:    vm.Status,
-				Name:      vm.Name,
-				VMID:      vm.VMID,
-				CPU:       vm.CPU,
-				MaxCPU:    vm.CPUs,
-				Mem:       vm.Mem,
-				MaxMem:    vm.MaxMem,
-				Disk:      vm.Disk,
-				MaxDisk:   vm.MaxDisk,
-				NetIn:     vm.NetIn,
-				NetOut:    vm.NetOut,
-				DiskRead:  vm.DiskRead,
-				DiskWrite: vm.DiskWrite,
-				Uptime:    vm.Uptime,
-				Template:  vm.Template,
-				Tags:      vm.Tags,
+				ID:         guestID,
+				Type:       "qemu",
+				Node:       node,
+				Pool:       vm.Pool,
+				Status:     vm.Status,
+				Name:       vm.Name,
+				VMID:       vm.VMID,
+				CPU:        vm.CPU,
+				MaxCPU:     vm.CPUs,
+				Mem:        vm.Mem,
+				MaxMem:     vm.MaxMem,
+				Disk:       vm.Disk,
+				MaxDisk:    vm.MaxDisk,
+				NetIn:      vm.NetIn,
+				NetOut:     vm.NetOut,
+				DiskRead:   vm.DiskRead,
+				DiskWrite:  vm.DiskWrite,
+				Uptime:     vm.Uptime,
+				Template:   vm.Template,
+				Tags:       vm.Tags,
+				IOCounters: vm.IOCounters,
+				ObservedAt: vm.ObservedAt,
 			},
 			guestID: guestID,
 		})
@@ -7,81 +7,49 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/models"
 )

-// IOMetrics is an alias for models.IOMetrics
+// IOMetrics is an alias for models.IOMetrics.
 type IOMetrics = models.IOMetrics

-// rateWindowSize is the number of counter samples retained per guest.
-// Rate is computed from the oldest to the newest sample, giving an average
-// over (rateWindowSize-1) polling intervals. With a 10s poll interval and
-// window size 4, this produces a 30-second sliding window — the same approach
-// Prometheus rate() uses to smooth out per-interval counter jitter.
-const rateWindowSize = 4
-
-// counterRing is a fixed-size ring buffer of IOMetrics samples.
-type counterRing struct {
-	entries [rateWindowSize]IOMetrics
-	count   int // number of entries stored (up to rateWindowSize)
-	head    int // next write position
+type counterBaseline struct {
+	value       int64
+	observedAt  time.Time
+	initialized bool
 }

-func (r *counterRing) add(m IOMetrics) {
-	r.entries[r.head] = m
-	r.head = (r.head + 1) % rateWindowSize
-	if r.count < rateWindowSize {
-		r.count++
-	}
+type counterHistory struct {
+	diskRead       counterBaseline
+	diskWrite      counterBaseline
+	diskBusy       counterBaseline
+	networkIn      counterBaseline
+	networkOut     counterBaseline
+	lastObservedAt time.Time
+	sourceUptime   uint64
 }

-func (r *counterRing) oldest() IOMetrics {
-	if r.count < rateWindowSize {
-		return r.entries[0]
-	}
-	return r.entries[r.head] // head points to the oldest when full
-}
-
-func (r *counterRing) newest() IOMetrics {
-	return r.entries[(r.head-1+rateWindowSize)%rateWindowSize]
-}
-
-// newestTimestamp returns the timestamp of the most recent entry.
-func (r *counterRing) newestTimestamp() time.Time {
-	return r.newest().Timestamp
-}
-
-// RateTracker tracks I/O metrics to calculate rates
+// RateTracker converts cumulative byte counters into adjacent-sample rates.
+// Each counter has an independent baseline because Proxmox may omit only part
+// of an otherwise valid status payload.
 type RateTracker struct {
-	mu        sync.RWMutex
-	history   map[string]*counterRing
-	lastRates map[string]RateCache
+	mu      sync.RWMutex
+	history map[string]*counterHistory
 }

-// RateCache stores the last calculated rates for a guest
-type RateCache struct {
-	DiskReadRate  float64
-	DiskWriteRate float64
-	DiskBusyPct   float64
-	NetInRate     float64
-	NetOutRate    float64
-}
-
-// NewRateTracker creates a new rate tracker
+// NewRateTracker creates a new rate tracker.
 func NewRateTracker() *RateTracker {
-	return &RateTracker{
-		history:   make(map[string]*counterRing),
-		lastRates: make(map[string]RateCache),
-	}
+	return &RateTracker{history: make(map[string]*counterHistory)}
 }

-// CalculateRates calculates I/O rates for a guest
-// Returns -1 for rates that don't have enough data yet (will be converted to null in JSON)
+// CalculateRates calculates disk and network rates in bytes per second.
+// A negative result means the upstream counter was absent, the sample was
+// out-of-order, or no earlier observation exists. A returned zero is a valid
+// observed idle/reset interval.
 func (rt *RateTracker) CalculateRates(guestID string, current IOMetrics) (diskReadRate, diskWriteRate, netInRate, netOutRate float64) {
 	diskReadRate, diskWriteRate, _, netInRate, netOutRate = rt.calculateRates(guestID, current)
 	return
 }

-// CalculateRatesWithBusy calculates disk/network rates plus disk busy percent
-// from cumulative counters. Returns -1 for metrics that do not yet have enough
-// data to produce a rate.
+// CalculateRatesWithBusy also calculates disk busy percent from a cumulative
+// millisecond counter.
 func (rt *RateTracker) CalculateRatesWithBusy(guestID string, current IOMetrics) (diskReadRate, diskWriteRate, diskBusyPct, netInRate, netOutRate float64) {
 	return rt.calculateRates(guestID, current)
 }
@@ -90,102 +58,104 @@ func (rt *RateTracker) calculateRates(guestID string, current IOMetrics) (diskRe
 	rt.mu.Lock()
 	defer rt.mu.Unlock()

-	ring, exists := rt.history[guestID]
-
-	if !exists {
-		// No previous data, store it and return -1 to indicate no data available
-		ring = &counterRing{}
-		ring.add(current)
-		rt.history[guestID] = ring
+	if current.Timestamp.IsZero() {
 		return -1, -1, -1, -1, -1
 	}

-	prev := ring.newest()
-
-	// Check if the values have actually changed (detect stale data)
-	// If all cumulative values are the same, we're getting cached data from Proxmox
-	if current.DiskRead == prev.DiskRead &&
-		current.DiskWrite == prev.DiskWrite &&
-		current.DiskBusy == prev.DiskBusy &&
-		current.NetworkIn == prev.NetworkIn &&
-		current.NetworkOut == prev.NetworkOut {
-		// Data hasn't changed - return last known good rates
-		if lastRate, hasRate := rt.lastRates[guestID]; hasRate {
-			return lastRate.DiskReadRate, lastRate.DiskWriteRate, lastRate.DiskBusyPct, lastRate.NetInRate, lastRate.NetOutRate
+	history := rt.history[guestID]
+	if history == nil {
+		history = &counterHistory{}
+		rt.history[guestID] = history
+	}
+	if current.SourceUptime > 0 {
+		if history.sourceUptime > 0 && current.SourceUptime < history.sourceUptime {
+			history.resetCounterEpoch()
 		}
-		// No last rates available, return 0
-		return 0, 0, 0, 0, 0
+		history.sourceUptime = current.SourceUptime
 	}

-	// Data has changed, add to ring buffer
-	ring.add(current)
+	presence := current.Presence.Effective()
+	diskReadRate = calculateCounterRate(&history.diskRead, current.DiskRead, observationTime(current.ObservedAt.DiskRead, current.Timestamp), presence.DiskRead)
+	diskWriteRate = calculateCounterRate(&history.diskWrite, current.DiskWrite, observationTime(current.ObservedAt.DiskWrite, current.Timestamp), presence.DiskWrite)
+	diskBusyRate := calculateCounterRate(&history.diskBusy, current.DiskBusy, observationTime(current.ObservedAt.DiskBusy, current.Timestamp), presence.DiskBusy)
+	netInRate = calculateCounterRate(&history.networkIn, current.NetworkIn, observationTime(current.ObservedAt.NetworkIn, current.Timestamp), presence.NetworkIn)
+	netOutRate = calculateCounterRate(&history.networkOut, current.NetworkOut, observationTime(current.ObservedAt.NetworkOut, current.Timestamp), presence.NetworkOut)

-	// Calculate rate over the full window (oldest to current), like Prometheus rate().
-	// This naturally smooths out per-interval jitter from Proxmox's lumpy counter
-	// reporting by averaging over a wider time span.
-	oldest := ring.oldest()
-	timeDiff := current.Timestamp.Sub(oldest.Timestamp).Seconds()
-	if timeDiff <= 0 {
-		// Return last known rates if time hasn't advanced
-		if lastRate, hasRate := rt.lastRates[guestID]; hasRate {
-			return lastRate.DiskReadRate, lastRate.DiskWriteRate, lastRate.DiskBusyPct, lastRate.NetInRate, lastRate.NetOutRate
-		}
-		return 0, 0, 0, 0, 0
-	}
-
-	// Calculate rates (bytes per second) over the window
-	if current.DiskRead >= oldest.DiskRead {
-		diskReadRate = float64(current.DiskRead-oldest.DiskRead) / timeDiff
-	}
-	if current.DiskWrite >= oldest.DiskWrite {
-		diskWriteRate = float64(current.DiskWrite-oldest.DiskWrite) / timeDiff
-	}
-	if current.DiskBusy >= oldest.DiskBusy {
-		diskBusyPct = (float64(current.DiskBusy-oldest.DiskBusy) / (timeDiff * 1000)) * 100
-		if diskBusyPct < 0 {
-			diskBusyPct = 0
-		}
+	if diskBusyRate < 0 {
+		diskBusyPct = -1
+	} else {
+		// DiskBusy is cumulative busy milliseconds, so ms/s divided by ten is
+		// the percentage of wall time spent busy.
+		diskBusyPct = diskBusyRate / 10
 		if diskBusyPct > 100 {
 			diskBusyPct = 100
 		}
 	}
-	if current.NetworkIn >= oldest.NetworkIn {
-		netInRate = float64(current.NetworkIn-oldest.NetworkIn) / timeDiff
-	}
-	if current.NetworkOut >= oldest.NetworkOut {
-		netOutRate = float64(current.NetworkOut-oldest.NetworkOut) / timeDiff
-	}

-	// Cache the calculated rates
-	rt.lastRates[guestID] = RateCache{
-		DiskReadRate:  diskReadRate,
-		DiskWriteRate: diskWriteRate,
-		DiskBusyPct:   diskBusyPct,
-		NetInRate:     netInRate,
-		NetOutRate:    netOutRate,
+	if current.Timestamp.After(history.lastObservedAt) {
+		history.lastObservedAt = current.Timestamp
 	}
-
 	return
 }

-// Clear removes all tracked data
+func (history *counterHistory) resetCounterEpoch() {
+	history.diskRead = counterBaseline{}
+	history.diskWrite = counterBaseline{}
+	history.diskBusy = counterBaseline{}
+	history.networkIn = counterBaseline{}
+	history.networkOut = counterBaseline{}
+}
+
+func observationTime(counterTime, fallback time.Time) time.Time {
+	if counterTime.IsZero() {
+		return fallback
+	}
+	return counterTime
+}
+
+func calculateCounterRate(baseline *counterBaseline, value int64, observedAt time.Time, present bool) float64 {
+	if !present {
+		return -1
+	}
+	if !baseline.initialized {
+		baseline.value = value
+		baseline.observedAt = observedAt
+		baseline.initialized = true
+		return -1
+	}
+	if !observedAt.After(baseline.observedAt) {
+		return -1
+	}
+
+	elapsed := observedAt.Sub(baseline.observedAt).Seconds()
+	previous := baseline.value
+	baseline.value = value
+	baseline.observedAt = observedAt
+
+	if value < previous {
+		// A guest restart, migration reconnect, or counter wrap starts a new
+		// cumulative epoch. Rebase without inventing a negative or huge rate.
+		return 0
+	}
+	return float64(value-previous) / elapsed
+}
+
+// Clear removes all tracked data.
 func (rt *RateTracker) Clear() {
 	rt.mu.Lock()
 	defer rt.mu.Unlock()
-	rt.history = make(map[string]*counterRing)
-	rt.lastRates = make(map[string]RateCache)
+	rt.history = make(map[string]*counterHistory)
 }

-// Cleanup removes entries for resources that haven't reported data since the cutoff time.
-// This prevents unbounded memory growth when containers/VMs are deleted.
+// Cleanup removes resources that have not supplied any newer sample since the
+// cutoff. Idle and partial samples still refresh resource observation time.
 func (rt *RateTracker) Cleanup(cutoff time.Time) (removed int) {
 	rt.mu.Lock()
 	defer rt.mu.Unlock()

-	for guestID, ring := range rt.history {
-		if ring.newestTimestamp().Before(cutoff) {
+	for guestID, history := range rt.history {
+		if history.lastObservedAt.Before(cutoff) {
 			delete(rt.history, guestID)
-			delete(rt.lastRates, guestID)
 			removed++
 		}
 	}
@@ -1,397 +1,236 @@
 package monitoring

 import (
+	"math"
 	"testing"
 	"time"

 	"github.com/rcourtman/pulse-go-rewrite/internal/models"
 )

-func TestCalculateRates_FirstCallReturnsNegativeOnes(t *testing.T) {
-	rt := NewRateTracker()
-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: time.Now(),
-	})
-	if d != -1 || w != -1 || ni != -1 || no != -1 {
-		t.Errorf("first call: got (%v, %v, %v, %v), want (-1, -1, -1, -1)", d, w, ni, no)
+func fullCounterSample(at time.Time, diskRead, diskWrite, networkIn, networkOut int64) models.IOMetrics {
+	return models.IOMetrics{
+		DiskRead:   diskRead,
+		DiskWrite:  diskWrite,
+		NetworkIn:  networkIn,
+		NetworkOut: networkOut,
+		Timestamp:  at,
 	}
 }

-func TestCalculateRates_StaleDataReturnsCachedRates(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
+func TestRateTrackerConstantRatesUseActualElapsedTime(t *testing.T) {
+	for _, interval := range []time.Duration{30 * time.Second, 60 * time.Second, 90 * time.Second} {
+		t.Run(interval.String(), func(t *testing.T) {
+			tracker := NewRateTracker()
+			start := time.Unix(1_700_000_000, 0)
+			tracker.CalculateRates("guest", fullCounterSample(start, 1_000, 2_000, 3_000, 4_000))

-	// Seed
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: base,
-	})
-	// Establish rates
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000,
-		Timestamp: base.Add(10 * time.Second),
-	})
-
-	// Send stale data (same counter values, different timestamp)
-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000,
-		Timestamp: base.Add(20 * time.Second),
-	})
-	if d != 500 || w != 1000 || ni != 1500 || no != 2000 {
-		t.Errorf("stale data: got (%v, %v, %v, %v), want (500, 1000, 1500, 2000)", d, w, ni, no)
+			seconds := int64(interval / time.Second)
+			read, write, in, out := tracker.CalculateRates(
+				"guest",
+				fullCounterSample(
+					start.Add(interval),
+					1_000+seconds*1_024,
+					2_000+seconds*2_048,
+					3_000+seconds*4_096,
+					4_000+seconds*8_192,
+				),
+			)
+			if read != 1_024 || write != 2_048 || in != 4_096 || out != 8_192 {
+				t.Fatalf("rates = (%v, %v, %v, %v), want (1024, 2048, 4096, 8192)", read, write, in, out)
+			}
+		})
 	}
 }

-func TestCalculateRates_StaleDataWithoutCachedRatesReturnsZeros(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
+func TestRateTrackerBurstUsesAdjacentObservationInterval(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0))

-	// Seed with values
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: base,
-	})
+	read, _, _, _ := tracker.CalculateRates("guest", fullCounterSample(start.Add(90*time.Second), 90_000, 0, 0, 0))
+	if read != 1_000 {
+		t.Fatalf("burst rate = %v, want 1000 B/s", read)
+	}

-	// Send identical values (stale) — no cached rates yet
-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: base.Add(10 * time.Second),
-	})
-	if d != 0 || w != 0 || ni != 0 || no != 0 {
-		t.Errorf("stale without cache: got (%v, %v, %v, %v), want (0, 0, 0, 0)", d, w, ni, no)
+	read, _, _, _ = tracker.CalculateRates("guest", fullCounterSample(start.Add(180*time.Second), 90_000, 0, 0, 0))
+	if read != 0 {
+		t.Fatalf("idle interval rate = %v, want valid zero", read)
 	}
 }

-func TestCalculateRates_NormalRateCalculation(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
-
-	// Seed
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: base,
+func TestRateTrackerMissingAndExplicitZeroAreDistinct(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	diskOnly := models.IOCounterPresence{Explicit: true, DiskRead: true}
+	tracker.CalculateRates("guest", models.IOMetrics{
+		DiskRead:  0,
+		Timestamp: start,
+		Presence:  diskOnly,
 	})

-	// Second call — rate over 1 interval (ring only has 2 entries)
-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000,
-		Timestamp: base.Add(10 * time.Second),
+	read, write, in, out := tracker.CalculateRates("guest", models.IOMetrics{
+		DiskRead:  0,
+		Timestamp: start.Add(60 * time.Second),
+		Presence:  diskOnly,
 	})
-	if d != 500 || w != 1000 || ni != 1500 || no != 2000 {
-		t.Errorf("normal rate: got (%v, %v, %v, %v), want (500, 1000, 1500, 2000)", d, w, ni, no)
+	if read != 0 {
+		t.Fatalf("explicit zero rate = %v, want 0", read)
+	}
+	if write != -1 || in != -1 || out != -1 {
+		t.Fatalf("missing rates = (%v, %v, %v), want unknown", write, in, out)
 	}
 }

-func TestCalculateRates_CounterRolloverReturnsZero(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
+func TestRateTrackerPartialSampleKeepsIndependentBaseline(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0))

-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 5000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: base,
+	read, write, in, out := tracker.CalculateRates("guest", models.IOMetrics{
+		DiskRead:  60_000,
+		Timestamp: start.Add(60 * time.Second),
+		Presence: models.IOCounterPresence{
+			Explicit: true,
+			DiskRead: true,
+		},
 	})
-
-	// DiskRead decreased (counter rollover)
-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000,
-		Timestamp: base.Add(10 * time.Second),
-	})
-	if d != 0 {
-		t.Errorf("DiskRead rollover: got %v, want 0", d)
+	if read != 1_000 || write != -1 || in != -1 || out != -1 {
+		t.Fatalf("partial rates = (%v, %v, %v, %v)", read, write, in, out)
 	}
-	if w != 1000 || ni != 1500 || no != 2000 {
-		t.Errorf("other rates: got (%v, %v, %v), want (1000, 1500, 2000)", w, ni, no)
+
+	_, write, _, _ = tracker.CalculateRates("guest", models.IOMetrics{
+		DiskWrite: 180_000,
+		Timestamp: start.Add(90 * time.Second),
+		Presence: models.IOCounterPresence{
+			Explicit:  true,
+			DiskWrite: true,
+		},
+	})
+	if write != 2_000 {
+		t.Fatalf("disk write rate = %v, want 2000 over its 90s observation gap", write)
 	}
 }

-func TestCalculateRates_AllCountersRollover(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
+func TestRateTrackerUsesEachCounterReceiptTimeForPartialSources(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0))

-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000,
-		Timestamp: base,
+	read, write, _, _ := tracker.CalculateRates("guest", models.IOMetrics{
+		DiskRead:  30_000,
+		DiskWrite: 180_000,
+		Timestamp: start.Add(90 * time.Second),
+		Presence: models.IOCounterPresence{
+			Explicit:  true,
+			DiskRead:  true,
+			DiskWrite: true,
+		},
+		ObservedAt: models.IOCounterObservationTimes{
+			DiskRead:  start.Add(30 * time.Second),
+			DiskWrite: start.Add(90 * time.Second),
+		},
 	})
-
-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: base.Add(10 * time.Second),
-	})
-	if d != 0 || w != 0 || ni != 0 || no != 0 {
-		t.Errorf("all rollover: got (%v, %v, %v, %v), want (0, 0, 0, 0)", d, w, ni, no)
+	if read != 1_000 || write != 2_000 {
+		t.Fatalf("rates = (%v, %v), want per-counter receipt rates (1000, 2000)", read, write)
 	}
 }

-func TestCalculateRates_FractionalTimeDifference(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
+func TestRateTrackerResetOrWrapRebasesCounterEpoch(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	tracker.CalculateRates("guest", fullCounterSample(start, math.MaxInt64-1_000, 50_000, 10_000, 20_000))

-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: base,
-	})
+	read, write, in, out := tracker.CalculateRates(
+		"guest",
+		fullCounterSample(start.Add(30*time.Second), 500, 1_000, 100, 200),
+	)
+	if read != 0 || write != 0 || in != 0 || out != 0 {
+		t.Fatalf("reset rates = (%v, %v, %v, %v), want zeros", read, write, in, out)
+	}

-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1500, DiskWrite: 2500, NetworkIn: 3500, NetworkOut: 4500,
-		Timestamp: base.Add(500 * time.Millisecond),
-	})
-	// 500 / 0.5 = 1000
-	if d != 1000 || w != 1000 || ni != 1000 || no != 1000 {
-		t.Errorf("fractional time: got (%v, %v, %v, %v), want (1000, 1000, 1000, 1000)", d, w, ni, no)
+	read, _, _, _ = tracker.CalculateRates("guest", fullCounterSample(start.Add(60*time.Second), 30_500, 1_000, 100, 200))
+	if read != 1_000 {
+		t.Fatalf("post-reset rate = %v, want 1000", read)
 	}
 }

-func TestCalculateRates_LargeValues(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
+func TestRateTrackerUptimeRollbackRebasesEvenWhenCounterSurpassesOldValue(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	beforeRestart := fullCounterSample(start, 1_000, 2_000, 3_000, 4_000)
+	beforeRestart.SourceUptime = 10_000
+	tracker.CalculateRates("guest", beforeRestart)

-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000000000, DiskWrite: 2000000000, NetworkIn: 3000000000, NetworkOut: 4000000000,
-		Timestamp: base,
-	})
+	afterRestart := fullCounterSample(start.Add(90*time.Second), 91_000, 182_000, 273_000, 364_000)
+	afterRestart.SourceUptime = 30
+	read, write, in, out := tracker.CalculateRates("guest", afterRestart)
+	if read != -1 || write != -1 || in != -1 || out != -1 {
+		t.Fatalf("first sample in restarted epoch = (%v, %v, %v, %v), want unknown", read, write, in, out)
+	}

-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1100000000, DiskWrite: 2200000000, NetworkIn: 3300000000, NetworkOut: 4400000000,
-		Timestamp: base.Add(100 * time.Second),
-	})
-	if d != 1000000 || w != 2000000 || ni != 3000000 || no != 4000000 {
-		t.Errorf("large values: got (%v, %v, %v, %v), want (1000000, 2000000, 3000000, 4000000)", d, w, ni, no)
+	next := fullCounterSample(start.Add(120*time.Second), 121_000, 242_000, 363_000, 484_000)
+	next.SourceUptime = 60
+	read, write, in, out = tracker.CalculateRates("guest", next)
+	if read != 1_000 || write != 2_000 || in != 3_000 || out != 4_000 {
+		t.Fatalf("post-restart rates = (%v, %v, %v, %v), want (1000, 2000, 3000, 4000)", read, write, in, out)
 	}
 }

-func TestCalculateRates_WindowSmooths(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
+func TestRateTrackerRejectsOutOfOrderSamplesWithoutChangingBaseline(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0))
+	tracker.CalculateRates("guest", fullCounterSample(start.Add(60*time.Second), 60_000, 0, 0, 0))

-	// Simulate a steady 1000 bytes/sec download with Proxmox's lumpy counter updates.
-	// Over 4 intervals (40 seconds), 40000 bytes should arrive.
-	// But Proxmox distributes them unevenly across intervals.
-
-	// T=0: seed
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		NetworkIn: 0, Timestamp: base,
-	})
-
-	// T=10: normal interval (10000 bytes in 10s = 1000 B/s)
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		NetworkIn: 10000, Timestamp: base.Add(10 * time.Second),
-	})
-
-	// T=20: short-changed interval (only 5000 bytes reported)
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		NetworkIn: 15000, Timestamp: base.Add(20 * time.Second),
-	})
-
-	// T=30: lumpy interval (15000 bytes — makes up for the deficit + normal)
-	// Without windowing, raw rate would be 15000/10 = 1500 B/s (50% spike).
-	// With windowing (oldest=T=0, current=T=30), rate = 30000/30 = 1000 B/s.
-	_, _, ni, _ := rt.CalculateRates("vm-100", models.IOMetrics{
-		NetworkIn: 30000, Timestamp: base.Add(30 * time.Second),
-	})
-
-	if ni != 1000 {
-		t.Errorf("windowed rate during lumpy interval: got %v, want 1000", ni)
+	read, _, _, _ := tracker.CalculateRates("guest", fullCounterSample(start.Add(30*time.Second), 90_000, 0, 0, 0))
+	if read != -1 {
+		t.Fatalf("out-of-order rate = %v, want unknown", read)
 	}
-
-	// T=40: ring is now full (4 entries), oldest is T=10.
-	// Rate = (40000-10000)/(40-10) = 30000/30 = 1000 B/s
-	_, _, ni, _ = rt.CalculateRates("vm-100", models.IOMetrics{
-		NetworkIn: 40000, Timestamp: base.Add(40 * time.Second),
-	})
-
-	if ni != 1000 {
-		t.Errorf("windowed rate after ring full: got %v, want 1000", ni)
+	read, _, _, _ = tracker.CalculateRates("guest", fullCounterSample(start.Add(90*time.Second), 90_000, 0, 0, 0))
+	if read != 1_000 {
+		t.Fatalf("rate after rejected sample = %v, want 1000", read)
 	}
 }

-func TestCalculateRates_MultipleGuestsTrackedIndependently(t *testing.T) {
-	rt := NewRateTracker()
-	baseTime := time.Unix(1000, 0)
-
-	// First call for guest A - should return -1 for all
-	diskReadA1, diskWriteA1, netInA1, netOutA1 := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: baseTime,
-	})
-	if diskReadA1 != -1 || diskWriteA1 != -1 || netInA1 != -1 || netOutA1 != -1 {
-		t.Errorf("first call for vm-100: got (%v, %v, %v, %v), want (-1, -1, -1, -1)",
-			diskReadA1, diskWriteA1, netInA1, netOutA1)
-	}
-
-	// First call for guest B - should also return -1 for all
-	diskReadB1, diskWriteB1, netInB1, netOutB1 := rt.CalculateRates("vm-200", models.IOMetrics{
-		DiskRead: 5000, DiskWrite: 6000, NetworkIn: 7000, NetworkOut: 8000,
-		Timestamp: baseTime,
-	})
-	if diskReadB1 != -1 || diskWriteB1 != -1 || netInB1 != -1 || netOutB1 != -1 {
-		t.Errorf("first call for vm-200: got (%v, %v, %v, %v), want (-1, -1, -1, -1)",
-			diskReadB1, diskWriteB1, netInB1, netOutB1)
-	}
-
-	// Second call for guest A
-	diskReadA2, diskWriteA2, netInA2, netOutA2 := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 11000, DiskWrite: 22000, NetworkIn: 33000, NetworkOut: 44000,
-		Timestamp: baseTime.Add(10 * time.Second),
-	})
-	if diskReadA2 != 1000 || diskWriteA2 != 2000 || netInA2 != 3000 || netOutA2 != 4000 {
-		t.Errorf("second call for vm-100: got (%v, %v, %v, %v), want (1000, 2000, 3000, 4000)",
-			diskReadA2, diskWriteA2, netInA2, netOutA2)
-	}
-
-	// Second call for guest B - different rates
-	diskReadB2, diskWriteB2, netInB2, netOutB2 := rt.CalculateRates("vm-200", models.IOMetrics{
-		DiskRead: 10000, DiskWrite: 16000, NetworkIn: 22000, NetworkOut: 28000,
-		Timestamp: baseTime.Add(5 * time.Second),
-	})
-	if diskReadB2 != 1000 || diskWriteB2 != 2000 || netInB2 != 3000 || netOutB2 != 4000 {
-		t.Errorf("second call for vm-200: got (%v, %v, %v, %v), want (1000, 2000, 3000, 4000)",
-			diskReadB2, diskWriteB2, netInB2, netOutB2)
-	}
-}
-
-func TestCalculateRates_CachesRates(t *testing.T) {
-	rt := NewRateTracker()
-	baseTime := time.Unix(1000, 0)
-
-	// Seed
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000,
-		Timestamp: baseTime,
-	})
-
-	// Calculate rates
-	diskRead2, diskWrite2, netIn2, netOut2 := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 11000, DiskWrite: 22000, NetworkIn: 33000, NetworkOut: 44000,
-		Timestamp: baseTime.Add(10 * time.Second),
-	})
-
-	// Verify rates are cached
-	cachedRates, exists := rt.lastRates["vm-100"]
-	if !exists {
-		t.Fatal("expected rates to be cached for vm-100")
-	}
-	if cachedRates.DiskReadRate != diskRead2 {
-		t.Errorf("cached DiskReadRate = %v, want %v", cachedRates.DiskReadRate, diskRead2)
-	}
-	if cachedRates.DiskWriteRate != diskWrite2 {
-		t.Errorf("cached DiskWriteRate = %v, want %v", cachedRates.DiskWriteRate, diskWrite2)
-	}
-	if cachedRates.NetInRate != netIn2 {
-		t.Errorf("cached NetInRate = %v, want %v", cachedRates.NetInRate, netIn2)
-	}
-	if cachedRates.NetOutRate != netOut2 {
-		t.Errorf("cached NetOutRate = %v, want %v", cachedRates.NetOutRate, netOut2)
-	}
-
-	// Stale data returns cached rates
-	diskRead3, diskWrite3, netIn3, netOut3 := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 11000, DiskWrite: 22000, NetworkIn: 33000, NetworkOut: 44000,
-		Timestamp: baseTime.Add(15 * time.Second),
-	})
-	if diskRead3 != diskRead2 || diskWrite3 != diskWrite2 || netIn3 != netIn2 || netOut3 != netOut2 {
-		t.Errorf("stale data call: got (%v, %v, %v, %v), want cached (%v, %v, %v, %v)",
-			diskRead3, diskWrite3, netIn3, netOut3, diskRead2, diskWrite2, netIn2, netOut2)
-	}
-}
-
-func TestCalculateRates_DoesNotAddStaleDataToRing(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
-
-	// Seed
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, NetworkIn: 3000, Timestamp: base,
-	})
-
-	// Real data
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 6000, NetworkIn: 18000, Timestamp: base.Add(10 * time.Second),
-	})
-
-	// Stale data — should not be added to ring
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 6000, NetworkIn: 18000, Timestamp: base.Add(20 * time.Second),
-	})
-
-	// Ring should still have 2 entries (seed + one real update)
-	ring := rt.history["vm-100"]
-	if ring.count != 2 {
-		t.Errorf("ring count after stale data: got %d, want 2", ring.count)
-	}
-}
-
-func TestClear(t *testing.T) {
-	rt := NewRateTracker()
-	base := time.Unix(1000, 0)
-
-	rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, Timestamp: base,
-	})
-	rt.CalculateRates("vm-200", models.IOMetrics{
-		DiskRead: 1000, Timestamp: base,
-	})
-
-	if len(rt.history) != 2 {
-		t.Fatalf("expected 2 entries in history, got %d", len(rt.history))
-	}
-
-	rt.Clear()
-
-	if len(rt.history) != 0 {
-		t.Errorf("expected history to be empty after Clear, got %d entries", len(rt.history))
-	}
-	if len(rt.lastRates) != 0 {
-		t.Errorf("expected lastRates to be empty after Clear, got %d entries", len(rt.lastRates))
-	}
-
-	// After clear, first call returns -1
-	d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{
-		DiskRead: 1000, Timestamp: base,
-	})
-	if d != -1 || w != -1 || ni != -1 || no != -1 {
-		t.Errorf("after Clear: got (%v, %v, %v, %v), want (-1, -1, -1, -1)", d, w, ni, no)
-	}
-}
-
-func TestRateTrackerCleanup(t *testing.T) {
-	rt := NewRateTracker()
+func TestRateTrackerCleanupUsesLatestSampleEvenWhenIdleOrPartial(t *testing.T) {
+	tracker := NewRateTracker()
 	now := time.Now()
-
-	rt.CalculateRates("active-guest", models.IOMetrics{
-		DiskRead: 1000, Timestamp: now.Add(-1 * time.Hour),
-	})
-	rt.CalculateRates("stale-guest", models.IOMetrics{
-		DiskRead: 1000, Timestamp: now.Add(-48 * time.Hour),
+	tracker.CalculateRates("idle", fullCounterSample(now.Add(-2*time.Hour), 100, 100, 100, 100))
+	tracker.CalculateRates("idle", models.IOMetrics{
+		Timestamp: now,
+		Presence:  models.IOCounterPresence{Explicit: true},
 	})
+	tracker.CalculateRates("stale", fullCounterSample(now.Add(-2*time.Hour), 100, 100, 100, 100))

-	if len(rt.history) != 2 {
-		t.Fatalf("expected 2 entries, got %d", len(rt.history))
+	if removed := tracker.Cleanup(now.Add(-time.Hour)); removed != 1 {
+		t.Fatalf("removed = %d, want 1", removed)
 	}
-
-	cutoff := now.Add(-24 * time.Hour)
-	removed := rt.Cleanup(cutoff)
-
-	if removed != 1 {
-		t.Errorf("expected 1 entry removed, got %d", removed)
-	}
-	if len(rt.history) != 1 {
-		t.Errorf("expected 1 entry remaining, got %d", len(rt.history))
-	}
-	if _, exists := rt.history["active-guest"]; !exists {
-		t.Error("active-guest should still exist after cleanup")
-	}
-	if _, exists := rt.history["stale-guest"]; exists {
-		t.Error("stale-guest should be removed after cleanup")
+	if _, ok := tracker.history["idle"]; !ok {
+		t.Fatal("idle resource was removed despite a recent sample")
 	}
 }

-func TestRateTrackerCleanupEmpty(t *testing.T) {
-	rt := NewRateTracker()
-	cutoff := time.Now().Add(-24 * time.Hour)
-
-	removed := rt.Cleanup(cutoff)
-	if removed != 0 {
-		t.Errorf("expected 0 entries removed from empty tracker, got %d", removed)
+func TestRateTrackerDiskBusyUsesElapsedMilliseconds(t *testing.T) {
+	tracker := NewRateTracker()
+	start := time.Unix(1_700_000_000, 0)
+	tracker.CalculateRatesWithBusy("disk", models.IOMetrics{DiskBusy: 100, Timestamp: start})
+	_, _, busy, _, _ := tracker.CalculateRatesWithBusy("disk", models.IOMetrics{
+		DiskBusy:  15_100,
+		Timestamp: start.Add(30 * time.Second),
+	})
+	if busy != 50 {
+		t.Fatalf("busy = %v, want 50%%", busy)
+	}
+}
+
+func TestRateTrackerClearRestoresFirstSampleUnknown(t *testing.T) {
+	tracker := NewRateTracker()
+	now := time.Now()
+	tracker.CalculateRates("guest", fullCounterSample(now, 1, 2, 3, 4))
+	tracker.Clear()
+	read, write, in, out := tracker.CalculateRates("guest", fullCounterSample(now.Add(time.Second), 2, 3, 4, 5))
+	if read != -1 || write != -1 || in != -1 || out != -1 {
+		t.Fatalf("first rates after clear = (%v, %v, %v, %v)", read, write, in, out)
 	}
 }
@@ -39,6 +39,10 @@ func (m *Monitor) resourceStaleThresholds() map[unifiedresources.DataSource]time
 	return ResourceStaleThresholdsForConfig(m.config)
 }

+func (m *Monitor) pveNodeOfflineGracePeriod() time.Duration {
+	return m.resourceStaleThresholds()[unifiedresources.SourceProxmox]
+}
+
 func effectivePVEPollingIntervalForConfig(cfg *config.Config) time.Duration {
 	const minInterval = 10 * time.Second
 	const maxInterval = time.Hour
@@ -59,6 +59,7 @@ package unifiedresources
 // banned direct-state access patterns.

 import (
+	"encoding/json"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -70,6 +71,22 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/models"
 )

+func TestMetricValueAlwaysCarriesNumericValueWhenObjectExists(t *testing.T) {
+	payload, err := json.Marshal(MetricValue{Unit: "bytes/s", Source: SourceProxmox})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var decoded map[string]any
+	if err := json.Unmarshal(payload, &decoded); err != nil {
+		t.Fatal(err)
+	}
+	value, ok := decoded["value"].(float64)
+	if !ok || value != 0 {
+		t.Fatalf("MetricValue JSON must contain numeric zero, got %s", payload)
+	}
+}
+
 func TestActionTruthTypesStayUnifiedResourceOwned(t *testing.T) {
 	repoRoot := filepath.Join("..", "..")
 	roots := []string{"internal/api", "internal/ai", "internal/agentexec", "internal/hostagent", "internal/dockeragent", "internal/relay", "internal/workflow", "internal/workflows"}
@@ -159,6 +159,7 @@ func metricsFromVM(vm models.VM) *ResourceMetrics {
 		vm.NetworkOut,
 		vm.DiskRead,
 		vm.DiskWrite,
+		vm.IORateValidity,
 		SourceProxmox,
 	)
 }
@@ -172,6 +173,7 @@ func metricsFromContainer(ct models.Container) *ResourceMetrics {
 		ct.NetworkOut,
 		ct.DiskRead,
 		ct.DiskWrite,
+		ct.IORateValidity,
 		SourceProxmox,
 	)
 }
@@ -181,6 +183,7 @@ func buildVMMetricPayload(
 	memory models.Memory,
 	disk models.Disk,
 	netIn, netOut, diskRead, diskWrite int64,
+	rateValidity models.IORateValidity,
 	source DataSource,
 ) *ResourceMetrics {
 	metrics := &ResourceMetrics{}
@@ -194,7 +197,7 @@ func buildVMMetricPayload(
 		percent := percentFromUsage(disk.Usage)
 		metrics.Disk = &MetricValue{Used: &disk.Used, Total: &disk.Total, Percent: percent, Unit: "bytes", Source: source}
 	}
-	setNetworkAndDiskIOMetricsVM(metrics, netIn, netOut, diskRead, diskWrite, source)
+	setNetworkAndDiskIOMetricsVM(metrics, netIn, netOut, diskRead, diskWrite, rateValidity, source)
 	return metrics
 }

@@ -822,17 +825,18 @@ func setNetworkAndDiskIOMetricsHost(metrics *ResourceMetrics, netIn, netOut, dis
 	}
 }

-func setNetworkAndDiskIOMetricsVM(metrics *ResourceMetrics, netIn, netOut, diskRead, diskWrite int64, source DataSource) {
-	if netIn != 0 {
+func setNetworkAndDiskIOMetricsVM(metrics *ResourceMetrics, netIn, netOut, diskRead, diskWrite int64, validity models.IORateValidity, source DataSource) {
+	effective := validity.EffectiveForRates(diskRead, diskWrite, netIn, netOut)
+	if effective.NetworkIn {
 		metrics.NetIn = &MetricValue{Value: float64(netIn), Unit: "bytes/s", Source: source}
 	}
-	if netOut != 0 {
+	if effective.NetworkOut {
 		metrics.NetOut = &MetricValue{Value: float64(netOut), Unit: "bytes/s", Source: source}
 	}
-	if diskRead != 0 {
+	if effective.DiskRead {
 		metrics.DiskRead = &MetricValue{Value: float64(diskRead), Unit: "bytes/s", Source: source}
 	}
-	if diskWrite != 0 {
+	if effective.DiskWrite {
 		metrics.DiskWrite = &MetricValue{Value: float64(diskWrite), Unit: "bytes/s", Source: source}
 	}
 }
@@ -33,6 +33,31 @@ func TestMetricsFromDockerHostIncludesIORates(t *testing.T) {
 	}
 }

+func TestProxmoxGuestMetricsDistinguishValidZeroFromUnknown(t *testing.T) {
+	validZero := metricsFromVM(models.VM{
+		Status: "running",
+		IORateValidity: models.IORateValidity{
+			Explicit:   true,
+			DiskRead:   true,
+			DiskWrite:  true,
+			NetworkIn:  true,
+			NetworkOut: true,
+		},
+	})
+	if validZero.DiskRead == nil || validZero.DiskRead.Value != 0 ||
+		validZero.DiskWrite == nil || validZero.NetIn == nil || validZero.NetOut == nil {
+		t.Fatalf("valid zero rates were not projected: %+v", validZero)
+	}
+
+	unknown := metricsFromVM(models.VM{
+		Status:         "running",
+		IORateValidity: models.IORateValidity{Explicit: true},
+	})
+	if unknown.DiskRead != nil || unknown.DiskWrite != nil || unknown.NetIn != nil || unknown.NetOut != nil {
+		t.Fatalf("unknown rates were projected as measurements: %+v", unknown)
+	}
+}
+
 func TestUnavailableMemoryDoesNotProjectOrOverwriteTrustedCrossSourceMetric(t *testing.T) {
 	unavailable := models.UnavailableMemory(8 * 1024 * 1024 * 1024)

@@ -321,7 +321,7 @@ type ResourceMetrics struct {

 // MetricValue represents a metric value, optionally with totals.
 type MetricValue struct {
-	Value   float64    `json:"value,omitempty"`
+	Value   float64    `json:"value"`
 	Used    *int64     `json:"used,omitempty"`
 	Total   *int64     `json:"total,omitempty"`
 	Percent float64    `json:"percent,omitempty"`
@@ -943,59 +943,63 @@ func (c *Client) GetVMRRDData(ctx context.Context, node string, vmid int, timefr

 // VM represents a Proxmox VE virtual machine
 type VM struct {
-	VMID      int     `json:"vmid"`
-	Name      string  `json:"name"`
-	Node      string  `json:"node"`
-	Pool      string  `json:"pool,omitempty"`
-	Status    string  `json:"status"`
-	CPU       float64 `json:"cpu"`
-	CPUs      int     `json:"cpus"`
-	Mem       uint64  `json:"mem"`
-	MaxMem    uint64  `json:"maxmem"`
-	Disk      uint64  `json:"disk"`
-	MaxDisk   uint64  `json:"maxdisk"`
-	NetIn     uint64  `json:"netin"`
-	NetOut    uint64  `json:"netout"`
-	DiskRead  uint64  `json:"diskread"`
-	DiskWrite uint64  `json:"diskwrite"`
-	Uptime    uint64  `json:"uptime"`
-	Template  int     `json:"template"`
-	Tags      string  `json:"tags"`
-	Lock      string  `json:"lock"`
-	Agent     int     `json:"agent"`
+	VMID       int               `json:"vmid"`
+	Name       string            `json:"name"`
+	Node       string            `json:"node"`
+	Pool       string            `json:"pool,omitempty"`
+	Status     string            `json:"status"`
+	CPU        float64           `json:"cpu"`
+	CPUs       int               `json:"cpus"`
+	Mem        uint64            `json:"mem"`
+	MaxMem     uint64            `json:"maxmem"`
+	Disk       uint64            `json:"disk"`
+	MaxDisk    uint64            `json:"maxdisk"`
+	NetIn      uint64            `json:"netin"`
+	NetOut     uint64            `json:"netout"`
+	DiskRead   uint64            `json:"diskread"`
+	DiskWrite  uint64            `json:"diskwrite"`
+	Uptime     uint64            `json:"uptime"`
+	Template   int               `json:"template"`
+	Tags       string            `json:"tags"`
+	Lock       string            `json:"lock"`
+	Agent      int               `json:"agent"`
+	IOCounters IOCounterPresence `json:"-"`
+	ObservedAt time.Time         `json:"-"`
 }

 // Container represents a Proxmox VE LXC container
 type Container struct {
-	VMID      FlexInt                           `json:"vmid"` // Changed to FlexInt to handle string VMIDs from some Proxmox versions
-	Name      string                            `json:"name"`
-	Node      string                            `json:"node"`
-	Pool      string                            `json:"pool,omitempty"`
-	Status    string                            `json:"status"`
-	CPU       float64                           `json:"cpu"`
-	CPUs      FlexInt                           `json:"cpus"`
-	Mem       uint64                            `json:"mem"`
-	MaxMem    uint64                            `json:"maxmem"`
-	Swap      uint64                            `json:"swap"`
-	MaxSwap   uint64                            `json:"maxswap"`
-	Disk      uint64                            `json:"disk"`
-	MaxDisk   uint64                            `json:"maxdisk"`
-	NetIn     uint64                            `json:"netin"`
-	NetOut    uint64                            `json:"netout"`
-	DiskRead  uint64                            `json:"diskread"`
-	DiskWrite uint64                            `json:"diskwrite"`
-	Uptime    uint64                            `json:"uptime"`
-	Template  int                               `json:"template"`
-	Tags      string                            `json:"tags"`
-	Lock      string                            `json:"lock"`
-	Hostname  string                            `json:"hostname,omitempty"`
-	IP        string                            `json:"ip,omitempty"`
-	IP6       string                            `json:"ip6,omitempty"`
-	IPv4      json.RawMessage                   `json:"ipv4,omitempty"`
-	IPv6      json.RawMessage                   `json:"ipv6,omitempty"`
-	Network   map[string]ContainerNetworkConfig `json:"network,omitempty"`
-	DiskInfo  map[string]ContainerDiskUsage     `json:"diskinfo,omitempty"`
-	RootFS    string                            `json:"rootfs,omitempty"`
+	VMID       FlexInt                           `json:"vmid"` // Changed to FlexInt to handle string VMIDs from some Proxmox versions
+	Name       string                            `json:"name"`
+	Node       string                            `json:"node"`
+	Pool       string                            `json:"pool,omitempty"`
+	Status     string                            `json:"status"`
+	CPU        float64                           `json:"cpu"`
+	CPUs       FlexInt                           `json:"cpus"`
+	Mem        uint64                            `json:"mem"`
+	MaxMem     uint64                            `json:"maxmem"`
+	Swap       uint64                            `json:"swap"`
+	MaxSwap    uint64                            `json:"maxswap"`
+	Disk       uint64                            `json:"disk"`
+	MaxDisk    uint64                            `json:"maxdisk"`
+	NetIn      uint64                            `json:"netin"`
+	NetOut     uint64                            `json:"netout"`
+	DiskRead   uint64                            `json:"diskread"`
+	DiskWrite  uint64                            `json:"diskwrite"`
+	Uptime     uint64                            `json:"uptime"`
+	Template   int                               `json:"template"`
+	Tags       string                            `json:"tags"`
+	Lock       string                            `json:"lock"`
+	Hostname   string                            `json:"hostname,omitempty"`
+	IP         string                            `json:"ip,omitempty"`
+	IP6        string                            `json:"ip6,omitempty"`
+	IPv4       json.RawMessage                   `json:"ipv4,omitempty"`
+	IPv6       json.RawMessage                   `json:"ipv6,omitempty"`
+	Network    map[string]ContainerNetworkConfig `json:"network,omitempty"`
+	DiskInfo   map[string]ContainerDiskUsage     `json:"diskinfo,omitempty"`
+	RootFS     string                            `json:"rootfs,omitempty"`
+	IOCounters IOCounterPresence                 `json:"-"`
+	ObservedAt time.Time                         `json:"-"`
 }

 // ContainerNetworkConfig captures basic container network status information.
@@ -1153,6 +1157,7 @@ func (c *Client) GetVMs(ctx context.Context, node string) ([]VM, error) {
 		return nil, err
 	}

+	stampVMObservation(result.Data, time.Now().UTC())
 	return result.Data, nil
 }

@@ -1172,6 +1177,7 @@ func (c *Client) GetContainers(ctx context.Context, node string) ([]Container, e
 		return nil, err
 	}

+	stampContainerObservation(result.Data, time.Now().UTC())
 	return result.Data, nil
 }

@@ -2290,6 +2296,7 @@ func (c *Client) GetVMStatus(ctx context.Context, node string, vmid int) (*VMSta
 		return nil, err
 	}

+	result.Data.ObservedAt = time.Now().UTC()
 	return &result.Data, nil
 }

@@ -2309,31 +2316,34 @@ func (c *Client) GetContainerStatus(ctx context.Context, node string, vmid int)
 		return nil, err
 	}

+	result.Data.ObservedAt = time.Now().UTC()
 	return &result.Data, nil
 }

 // ClusterResource represents a resource from /cluster/resources
 type ClusterResource struct {
-	ID        string  `json:"id"`
-	Type      string  `json:"type"`
-	Node      string  `json:"node"`
-	Pool      string  `json:"pool,omitempty"`
-	Status    string  `json:"status"`
-	Name      string  `json:"name,omitempty"`
-	VMID      int     `json:"vmid,omitempty"`
-	CPU       float64 `json:"cpu,omitempty"`
-	MaxCPU    int     `json:"maxcpu,omitempty"`
-	Mem       uint64  `json:"mem,omitempty"`
-	MaxMem    uint64  `json:"maxmem,omitempty"`
-	Disk      uint64  `json:"disk,omitempty"`
-	MaxDisk   uint64  `json:"maxdisk,omitempty"`
-	NetIn     uint64  `json:"netin,omitempty"`
-	NetOut    uint64  `json:"netout,omitempty"`
-	DiskRead  uint64  `json:"diskread,omitempty"`
-	DiskWrite uint64  `json:"diskwrite,omitempty"`
-	Uptime    uint64  `json:"uptime,omitempty"`
-	Template  int     `json:"template,omitempty"`
-	Tags      string  `json:"tags,omitempty"`
+	ID         string            `json:"id"`
+	Type       string            `json:"type"`
+	Node       string            `json:"node"`
+	Pool       string            `json:"pool,omitempty"`
+	Status     string            `json:"status"`
+	Name       string            `json:"name,omitempty"`
+	VMID       int               `json:"vmid,omitempty"`
+	CPU        float64           `json:"cpu,omitempty"`
+	MaxCPU     int               `json:"maxcpu,omitempty"`
+	Mem        uint64            `json:"mem,omitempty"`
+	MaxMem     uint64            `json:"maxmem,omitempty"`
+	Disk       uint64            `json:"disk,omitempty"`
+	MaxDisk    uint64            `json:"maxdisk,omitempty"`
+	NetIn      uint64            `json:"netin,omitempty"`
+	NetOut     uint64            `json:"netout,omitempty"`
+	DiskRead   uint64            `json:"diskread,omitempty"`
+	DiskWrite  uint64            `json:"diskwrite,omitempty"`
+	Uptime     uint64            `json:"uptime,omitempty"`
+	Template   int               `json:"template,omitempty"`
+	Tags       string            `json:"tags,omitempty"`
+	IOCounters IOCounterPresence `json:"-"`
+	ObservedAt time.Time         `json:"-"`
 }

 // GetClusterResources returns all resources (VMs, containers) across the cluster
@@ -2357,6 +2367,7 @@ func (c *Client) GetClusterResources(ctx context.Context, resourceType string) (
 		return nil, err
 	}

+	stampClusterResourceObservation(result.Data, time.Now().UTC())
 	return result.Data, nil
 }

@@ -2596,24 +2607,26 @@ func (a *VMAgentField) UnmarshalJSON(data []byte) error {

 // VMStatus represents detailed VM status returned by Proxmox.
 type VMStatus struct {
-	Status      string         `json:"status"`
-	CPU         float64        `json:"cpu"`
-	CPUs        int            `json:"cpus"`
-	Mem         uint64         `json:"mem"`
-	MaxMem      uint64         `json:"maxmem"`
-	Balloon     uint64         `json:"balloon"`
-	BalloonMin  uint64         `json:"balloon_min"`
-	BalloonInfo *VMBalloonInfo `json:"ballooninfo,omitempty"`
-	FreeMem     uint64         `json:"freemem"`
-	MemInfo     *VMMemInfo     `json:"meminfo,omitempty"`
-	Disk        uint64         `json:"disk"`
-	MaxDisk     uint64         `json:"maxdisk"`
-	DiskRead    uint64         `json:"diskread"`
-	DiskWrite   uint64         `json:"diskwrite"`
-	NetIn       uint64         `json:"netin"`
-	NetOut      uint64         `json:"netout"`
-	Uptime      uint64         `json:"uptime"`
-	Agent       VMAgentField   `json:"agent"`
+	Status      string            `json:"status"`
+	CPU         float64           `json:"cpu"`
+	CPUs        int               `json:"cpus"`
+	Mem         uint64            `json:"mem"`
+	MaxMem      uint64            `json:"maxmem"`
+	Balloon     uint64            `json:"balloon"`
+	BalloonMin  uint64            `json:"balloon_min"`
+	BalloonInfo *VMBalloonInfo    `json:"ballooninfo,omitempty"`
+	FreeMem     uint64            `json:"freemem"`
+	MemInfo     *VMMemInfo        `json:"meminfo,omitempty"`
+	Disk        uint64            `json:"disk"`
+	MaxDisk     uint64            `json:"maxdisk"`
+	DiskRead    uint64            `json:"diskread"`
+	DiskWrite   uint64            `json:"diskwrite"`
+	NetIn       uint64            `json:"netin"`
+	NetOut      uint64            `json:"netout"`
+	Uptime      uint64            `json:"uptime"`
+	Agent       VMAgentField      `json:"agent"`
+	IOCounters  IOCounterPresence `json:"-"`
+	ObservedAt  time.Time         `json:"-"`
 }

 // GetZFSPoolStatus gets the status of ZFS pools on a node
@@ -0,0 +1,124 @@
+package proxmox
+
+import (
+	"bytes"
+	"encoding/json"
+	"time"
+)
+
+// IOCounterPresence distinguishes explicit zero counters from fields omitted
+// by a Proxmox endpoint or permission-limited response.
+type IOCounterPresence struct {
+	Explicit   bool
+	DiskRead   bool
+	DiskWrite  bool
+	NetworkIn  bool
+	NetworkOut bool
+}
+
+// Effective keeps manually constructed fixtures and older producers
+// compatible while decoded API responses retain exact field presence.
+func (p IOCounterPresence) Effective() IOCounterPresence {
+	if p.Explicit {
+		return p
+	}
+	return IOCounterPresence{
+		Explicit:   true,
+		DiskRead:   true,
+		DiskWrite:  true,
+		NetworkIn:  true,
+		NetworkOut: true,
+	}
+}
+
+func counterPresence(raw map[string]json.RawMessage) IOCounterPresence {
+	return IOCounterPresence{
+		Explicit:   true,
+		DiskRead:   jsonFieldPresent(raw, "diskread"),
+		DiskWrite:  jsonFieldPresent(raw, "diskwrite"),
+		NetworkIn:  jsonFieldPresent(raw, "netin"),
+		NetworkOut: jsonFieldPresent(raw, "netout"),
+	}
+}
+
+func jsonFieldPresent(raw map[string]json.RawMessage, key string) bool {
+	value, ok := raw[key]
+	return ok && !bytes.Equal(bytes.TrimSpace(value), []byte("null"))
+}
+
+func decodeWithCounterPresence(data []byte, target any) (IOCounterPresence, error) {
+	var raw map[string]json.RawMessage
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return IOCounterPresence{}, err
+	}
+	if err := json.Unmarshal(data, target); err != nil {
+		return IOCounterPresence{}, err
+	}
+	return counterPresence(raw), nil
+}
+
+func (v *VM) UnmarshalJSON(data []byte) error {
+	type alias VM
+	var decoded alias
+	presence, err := decodeWithCounterPresence(data, &decoded)
+	if err != nil {
+		return err
+	}
+	*v = VM(decoded)
+	v.IOCounters = presence
+	return nil
+}
+
+func (c *Container) UnmarshalJSON(data []byte) error {
+	type alias Container
+	var decoded alias
+	presence, err := decodeWithCounterPresence(data, &decoded)
+	if err != nil {
+		return err
+	}
+	*c = Container(decoded)
+	c.IOCounters = presence
+	return nil
+}
+
+func (r *ClusterResource) UnmarshalJSON(data []byte) error {
+	type alias ClusterResource
+	var decoded alias
+	presence, err := decodeWithCounterPresence(data, &decoded)
+	if err != nil {
+		return err
+	}
+	*r = ClusterResource(decoded)
+	r.IOCounters = presence
+	return nil
+}
+
+func (s *VMStatus) UnmarshalJSON(data []byte) error {
+	type alias VMStatus
+	var decoded alias
+	presence, err := decodeWithCounterPresence(data, &decoded)
+	if err != nil {
+		return err
+	}
+	*s = VMStatus(decoded)
+	s.IOCounters = presence
+	return nil
+}
+
+func stampVMObservation(values []VM, observedAt time.Time) {
+	for i := range values {
+		values[i].ObservedAt = observedAt
+	}
+}
+
+func stampContainerObservation(values []Container, observedAt time.Time) {
+	for i := range values {
+		values[i].ObservedAt = observedAt
+	}
+}
+
+func stampClusterResourceObservation(values []ClusterResource, observedAt time.Time) {
+	for i := range values {
+		values[i].ObservedAt = observedAt
+	}
+}
@@ -0,0 +1,117 @@
+package proxmox
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+func TestClusterResourceCounterPresenceDistinguishesZeroNullAndMissing(t *testing.T) {
+	var resource ClusterResource
+	if err := json.Unmarshal([]byte(`{
+		"type":"qemu",
+		"diskread":0,
+		"diskwrite":null,
+		"netin":42
+	}`), &resource); err != nil {
+		t.Fatal(err)
+	}
+
+	presence := resource.IOCounters.Effective()
+	if !presence.DiskRead || !presence.NetworkIn {
+		t.Fatalf("explicit zero/value fields were not present: %+v", presence)
+	}
+	if presence.DiskWrite || presence.NetworkOut {
+		t.Fatalf("null/missing fields were incorrectly present: %+v", presence)
+	}
+}
+
+func TestGuestStatusTypesRetainCounterPresence(t *testing.T) {
+	tests := []struct {
+		name string
+		read func() IOCounterPresence
+	}{
+		{
+			name: "vm listing",
+			read: func() IOCounterPresence {
+				var value VM
+				_ = json.Unmarshal([]byte(`{"diskread":0}`), &value)
+				return value.IOCounters
+			},
+		},
+		{
+			name: "lxc status",
+			read: func() IOCounterPresence {
+				var value Container
+				_ = json.Unmarshal([]byte(`{"diskread":0}`), &value)
+				return value.IOCounters
+			},
+		},
+		{
+			name: "qemu status",
+			read: func() IOCounterPresence {
+				var value VMStatus
+				_ = json.Unmarshal([]byte(`{"diskread":0}`), &value)
+				return value.IOCounters
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			presence := test.read().Effective()
+			if !presence.DiskRead || presence.DiskWrite || presence.NetworkIn || presence.NetworkOut {
+				t.Fatalf("presence = %+v, want only diskread", presence)
+			}
+		})
+	}
+}
+
+func TestObservationStampUsesOneReceiptTimeForResponse(t *testing.T) {
+	observedAt := time.Date(2026, time.July, 24, 8, 30, 0, 0, time.UTC)
+
+	vms := []VM{{VMID: 100}, {VMID: 101}}
+	stampVMObservation(vms, observedAt)
+	for _, vm := range vms {
+		if !vm.ObservedAt.Equal(observedAt) {
+			t.Fatalf("VM %d observedAt = %v", vm.VMID, vm.ObservedAt)
+		}
+	}
+
+	containers := []Container{{VMID: 200}, {VMID: 201}}
+	stampContainerObservation(containers, observedAt)
+	for _, container := range containers {
+		if !container.ObservedAt.Equal(observedAt) {
+			t.Fatalf("container %d observedAt = %v", container.VMID, container.ObservedAt)
+		}
+	}
+
+	resources := []ClusterResource{{VMID: 300}, {VMID: 301}}
+	stampClusterResourceObservation(resources, observedAt)
+	for _, resource := range resources {
+		if !resource.ObservedAt.Equal(observedAt) {
+			t.Fatalf("resource %d observedAt = %v", resource.VMID, resource.ObservedAt)
+		}
+	}
+}
+
+func TestInternalCounterMetadataNeverChangesProxmoxWireShape(t *testing.T) {
+	payload, err := json.Marshal(ClusterResource{
+		VMID:       100,
+		DiskRead:   0,
+		IOCounters: IOCounterPresence{Explicit: true, DiskRead: true},
+		ObservedAt: time.Now(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var raw map[string]json.RawMessage
+	if err := json.Unmarshal(payload, &raw); err != nil {
+		t.Fatal(err)
+	}
+	for _, key := range []string{"ioCounters", "IOCounters", "observedAt", "ObservedAt"} {
+		if _, ok := raw[key]; ok {
+			t.Fatalf("internal counter metadata %q leaked into JSON", key)
+		}
+	}
+}
@@ -216,6 +216,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
                    "exact_files": [
                        "internal/api/host_agent_removal_lifecycle_integration_test.go",
                        "internal/config/host_continuity_test.go",
+                        "internal/models/metrics_types_test.go",
                        "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
                        "internal/monitoring/monitor_host_agents_test.go",
                        "scripts/installtests/agent_state_dir_lifecycle_test.go",
@@ -258,6 +259,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
                "docker-swarm-runtime",
                "kubernetes-native-agent-runtime",
                "runtime-report-model",
+                "proxmox-guest-counter-runtime",
                "proxmox-zfs-runtime",
                "proxmox-cluster-client-runtime",
                "proxmox-ceph-runtime",
@@ -287,6 +289,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
                        "internal/monitoring/canonical_guardrails_test.go",
                        "internal/monitoring/issue1485_unraid_lifecycle_test.go",
                        "internal/monitoring/issue1595_collection_trust_test.go",
+                        "internal/monitoring/issue1613_contract_test.go",
                        "internal/monitoring/monitor_additional_test.go",
                        "internal/monitoring/monitor_alert_intent_test.go",
                        "internal/monitoring/monitor_alert_override_migration_test.go",
@@ -295,6 +298,8 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
                        "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
                        "internal/monitoring/monitor_host_agents_test.go",
                        "internal/monitoring/monitor_pve_cluster_refresh_test.go",
+                        "internal/monitoring/monitor_pve_guest_lxc_test.go",
+                        "internal/monitoring/ratetracker_test.go",
                        "internal/unifiedresources/code_standards_test.go",
                    ],
                }
@@ -392,6 +397,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
                    "exact_files": [
                        "internal/api/host_agent_removal_lifecycle_integration_test.go",
                        "internal/config/host_continuity_test.go",
+                        "internal/models/metrics_types_test.go",
                        "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
                        "internal/monitoring/monitor_host_agents_test.go",
                        "scripts/installtests/agent_state_dir_lifecycle_test.go",
@@ -4159,6 +4159,7 @@ class SubsystemLookupTest(unittest.TestCase):
            [
                "internal/api/host_agent_removal_lifecycle_integration_test.go",
                "internal/config/host_continuity_test.go",
+                "internal/models/metrics_types_test.go",
                "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
                "internal/monitoring/monitor_host_agents_test.go",
                "scripts/installtests/agent_state_dir_lifecycle_test.go",