Fix host agent clock-skew liveness

Refs #1519

internal/monitoring/monitor_agents.go

@@ -1644,6 +1644,9 @@ func normalizedSecurityStrings(values []string) []string {
 
 // ApplyHostReport ingests a host agent report into the shared state.
 func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.APITokenRecord) (models.Host, error) {
+	receivedAt := time.Now()
+	observedAt := receivedAt.UTC()
+
 	hostname := strings.TrimSpace(report.Host.Hostname)
 	if hostname == "" {
 		return models.Host{}, fmt.Errorf("host report missing hostname")
@@ -1783,11 +1786,6 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 		displayName = hostname
 	}
 
-	timestamp := report.Timestamp
-	if timestamp.IsZero() {
-		timestamp = time.Now().UTC()
-	}
-
 	memory := models.Memory{
 		Total:     report.Metrics.Memory.TotalBytes,
 		Used:      report.Metrics.Memory.UsedBytes,
@@ -1981,7 +1979,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 		Status:          "online",
 		UptimeSeconds:   report.Host.UptimeSeconds,
 		IntervalSeconds: report.Agent.IntervalSeconds,
-		LastSeen:        timestamp,
+		LastSeen:        observedAt,
 		AgentVersion:    strings.TrimSpace(report.Agent.Version),
 		MachineID:       strings.TrimSpace(report.Host.MachineID),
 		CommandsEnabled: report.Agent.CommandsEnabled,
@@ -2064,7 +2062,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 	// Compute I/O rates from cumulative counters before adding to state.
 	// Network and disk bytes from the agent are cumulative totals since boot;
 	// the RateTracker converts them to bytes/second, just like VMs and containers.
-	now := time.Now()
+	now := receivedAt
 
 	var totalRXBytes, totalTXBytes uint64
 	for _, nic := range host.NetworkInterfaces {
@@ -2114,7 +2112,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 
 	// If host reports Ceph data, also update the global CephClusters state
 	if report.Ceph != nil {
-		cephCluster := convertAgentCephToGlobalCluster(report.Ceph, hostname, identifier, timestamp)
+		cephCluster := convertAgentCephToGlobalCluster(report.Ceph, hostname, identifier, observedAt)
 		storedCephCluster := m.state.UpsertCephCluster(cephCluster)
 		log.Debug().
 			Str("hostId", identifier).
@@ -2196,7 +2194,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 	}
 
 	// Store cluster peer sensor data if present and evict stale entries
-	m.applyClusterSensors(report.ClusterSensors, timestamp)
+	m.applyClusterSensors(report.ClusterSensors, observedAt)
 	m.persistHostContinuity(host, report)
 
 	return host, nil

internal/monitoring/monitor_host_agents_test.go

@@ -706,6 +706,54 @@ func TestApplyHostReportPreservesThermalState(t *testing.T) {
 	}
 }
 
+func TestApplyHostReportUsesReceiptTimeForSkewedAgentClockLiveness(t *testing.T) {
+	monitor := newTestMonitor(t)
+
+	agentClock := time.Now().UTC().Add(-3 * time.Hour)
+	report := agentshost.Report{
+		Agent: agentshost.AgentInfo{
+			ID:              "clock-skew-agent",
+			Version:         "6.0.3",
+			IntervalSeconds: 30,
+		},
+		Host: agentshost.HostInfo{
+			ID:        "clock-skew-machine",
+			MachineID: "clock-skew-machine",
+			Hostname:  "clock-skew.local",
+			Platform:  "linux",
+		},
+		Timestamp: agentClock,
+		Metrics: agentshost.Metrics{
+			CPUUsagePercent: 12,
+		},
+	}
+
+	before := time.Now().UTC()
+	host, err := monitor.ApplyHostReport(report, &config.APITokenRecord{ID: "clock-skew-token"})
+	after := time.Now().UTC()
+	if err != nil {
+		t.Fatalf("ApplyHostReport: %v", err)
+	}
+	if host.LastSeen.Before(before) || host.LastSeen.After(after.Add(time.Second)) {
+		t.Fatalf("host LastSeen = %s, want server receipt between %s and %s", host.LastSeen, before, after)
+	}
+	if !host.LastSeen.After(agentClock.Add(2 * time.Hour)) {
+		t.Fatalf("host LastSeen followed skewed agent timestamp %s: got %s", agentClock, host.LastSeen)
+	}
+
+	readState := monitor.snapshotBackedUnifiedReadState()
+	if readState == nil {
+		t.Fatal("expected snapshot-backed read state")
+	}
+	hosts := readState.Hosts()
+	if len(hosts) != 1 {
+		t.Fatalf("host count = %d, want 1", len(hosts))
+	}
+	if hosts[0].Status() != unifiedresources.StatusOnline {
+		t.Fatalf("canonical host status = %q, want online", hosts[0].Status())
+	}
+}
+
 func TestApplyDockerReportPreservesNativeRuntimeInventory(t *testing.T) {
 	monitor := newTestMonitor(t)
 	now := time.Date(2026, 5, 24, 8, 0, 0, 0, time.UTC)