Judge resources API freshness by the monitor's thresholds (#2745)

Supersedes #2737, rebuilt on current main because its merge with main
conflicted in four files (`monitor.go`, `registry.go`,
`canonical_guardrails_test.go`, `code_standards_test.go`) and a
conflicted merge cannot pass the commit hooks.

The resources API rebuilt its registry from the monitor's unified seed
with the registry default stale thresholds, while the monitor judges
each source by its configured polling cadence (Proxmox/PBS/PMG: max(2x
polling interval, default)). Presentation coalescing used the defaults
too, in both the websocket broadcast and `/api/resources`.

Reproduced before the fix (Proxmox polled every two minutes, read 90
seconds after its poll, agent silent for 75 seconds):

| | websocket, before | `/api/resources`, before | both, after |
|---|---|---|---|
| Proxmox sighting | online | stale | online |
| node status | online | warning | online |
| CPU on a node whose agent keeps its own row (coalesced by hostname) |
70% (silent agent) | 70% (silent agent) | 20% (node's poll) |

## Change

- The seed carries the thresholds its registry generation was judged by.
`MonitorAdapter` records them per generation (`StaleThresholds`,
`GetAllWithMetricsTargetsAndStaleThresholds`). Thresholds set since the
last rebuild wait for the next one, and overlays and live supplemental
refreshes of a generation are judged by its thresholds. A view built
from mock fixtures or a bare snapshot reports none (the defaults it was
judged by).
- `Monitor.UnifiedResourceSnapshotWithStaleThresholds` and
`MultiTenantStateProvider.UnifiedResourceSnapshotWithStaleThresholdsForTenant`
hand seed, freshness and thresholds over from one view.
- `resourceapi` builds its registry with
`NewRegistryWithStaleThresholds` (ingests and stale passes that pass
none use them) and keys its cache on its own copy of the thresholds.
- The coalesce judges metric freshness with `metricSourceStale`, the
rule the stale pass and merge already share. The broadcast and
`ListForPresentation` pass their view's thresholds.
- Conflict resolution with main's exclusion-aware coalesce
(#2725/#2739): `ListForPresentation` passes both its exclusion filter
and its construction thresholds;
`MonitorAdapter.CoalesceForPresentation(resources, thresholds)` takes
the listing's thresholds next to the exclusions, and
`Monitor.coalesceResourcesForPresentation` passes the broadcast view's,
falling back to `CoalescePresentationHostResourcesWithStaleThresholds`.

Not changed here: Patrol's scoped runtime state still rebuilds
registries with the defaults, and PBS/PMG thresholds read the config
rather than runtime polling overrides. Both predate this change.

## Verification

- New: `TestContract_ResourceListJudgesFreshnessLikeTheMonitorBroadcast`
(default monitor, tenant monitor, monitor without a resource store, each
against the same monitor's broadcast),
`TestRegistrySeededFromMonitorJudgesFreshnessByItsThresholds`,
`TestMonitorAdapterReportsTheThresholdsItsGenerationWasJudgedBy`,
`TestUnifiedViewCarriesItsGenerationsStaleThresholds`,
`TestBroadcastCoalescedHostJudgesFreshnessByConfiguredThresholds`,
`TestResourceListRebuildsWhenSeedStaleThresholdsChange`, plus source
guardrails.
- Negative controls: each new test fails with its runtime piece reverted
(base `resources.go`, coalesce by defaults, providers dropping
thresholds, cache keeping the provider's map, adapter reporting pending
thresholds, overlay judged by pending thresholds).
- Full `internal/unifiedresources`, `internal/api/resourceapi` and
`internal/monitoring` suites pass. Re-run on the rebuilt tree: full
`internal/unifiedresources`, `internal/api/resourceapi` and
`internal/monitoring` suites pass, plus the stale-threshold,
resource-list and contract tests in `internal/api`. The full
`internal/api` suite passes except
`TestServerInfoEndpointReportsDevelopment`, which fails only because the
test tree had no `.git`.

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

@@ -8746,6 +8746,17 @@ nodes/controller members, and confirmed inventory removal.
 `TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata.
 This is synthetic runtime evidence, not USB hardware or reporter acceptance.
 
+### Coalesced host readings follow configured source freshness
+
+The websocket broadcast coalesces a Proxmox node with a Pulse agent that keeps
+its own registry row, and now judges which source's readings the row shows by
+the stale thresholds the view's registry used, the monitor's configured ones
+on the resource store (`internal/monitoring/monitor.go`).
+This changes which readings a coalesced row shows, not agent admission,
+registration, linking, heartbeat or removal. Proof:
+`TestBroadcastCoalescedHostJudgesFreshnessByConfiguredThresholds` in
+`internal/monitoring/monitor_host_agents_test.go`.
+
 ### Quick security setup preserves unrelated settings
 
 Authenticated force setup in `internal/api/security_setup_fix.go` retains the

docs/release-control/v6/internal/subsystems/api-contracts.md

@@ -9570,6 +9570,22 @@ That same resource-handler seed contract must also stay on canonical unified
 resource ownership for tenant-scoped requests: once a tenant state provider
 implements `UnifiedResourceSnapshotForTenant`, `/api/resources` may not fall
 back to raw tenant `StateSnapshot` seeding when that unified seed is empty.
+The seed contract carries source freshness too. A provider that implements
+`UnifiedResourceSnapshotWithStaleThresholds` (the default monitor) or
+`UnifiedResourceSnapshotWithStaleThresholdsForTenant` (the tenant state
+provider) returns the seed together with the stale thresholds its registry
+generation judged it by, captured with the listing, and the registry
+`/api/resources` rebuilds is built with them. Its stale pass and
+presentation coalesce then judge as the websocket broadcast does: a Proxmox node
+polled every two minutes stays online with a current sighting between polls
+instead of reading stale with warning status. The registry cache is keyed by
+its own copy of those thresholds as well as the freshness marker. A provider without those
+methods leaves the registry defaults, and the resources API reads them
+through optional interface checks, so `internal/api/router_helpers.go`
+asserts the monitor and the tenant state provider implement them. Proof:
+`TestContract_ResourceListJudgesFreshnessLikeTheMonitorBroadcast` (default
+and tenant monitors on a resource store, and a monitor without one, each
+against the same monitor's broadcast).
 That same mock/runtime contract now also governs chart payloads under
 `internal/api/chartapi/service.go`: when demo or mock presentation is enabled,
 `/api/charts`, `/api/charts/infrastructure`, and `/api/storage-charts` must

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

@@ -1193,6 +1193,18 @@ Monitoring owns source freshness cadence for Proxmox, PBS, and PMG resources:
 the stale threshold is derived from the configured polling interval with a
 minimum floor, so API-facing resource status must not degrade merely because a
 healthy source is between normal poll cycles.
+Each unified view carries the thresholds its listed registry generation was
+judged by, read with the listing from the same generation: the resource
+store's configured ones, or none (the defaults) for a view built from mock
+fixtures or a bare state snapshot.
+`UnifiedResourceSnapshotWithStaleThresholds` hands them, with the seed, to
+consumers that rebuild a registry from the monitor's resources (the resources
+API reads each tenant monitor's), and the websocket broadcast coalesces split
+host rows with them, so a coalesced row prefers readings by the same
+freshness rule and thresholds as the view's own merges. Regression coverage:
+`TestBroadcastCoalescedHostJudgesFreshnessByConfiguredThresholds` and
+`TestUnifiedViewCarriesItsGenerationsStaleThresholds` in
+`internal/monitoring/monitor_host_agents_test.go`.
 PBS and PMG configured instances also have one monitoring-owned runtime
 resource identity constructor. Poll publication, connection status, setup and
 auto-registration checks, canonical alias resolution, and alert-policy bridges

docs/release-control/v6/internal/subsystems/storage-recovery.md

@@ -6506,6 +6506,15 @@ unchanged and older builds ignore the fields. `internal/api/ai_handlers.go`
 reads them for the usage export only. No new file, directory, retention
 window, backup, migration or recovery authority is introduced.
 
+### Resource freshness thresholds open no storage or recovery path
+
+`internal/api/resourceapi/resources.go` builds the `/api/resources` registry
+with the stale thresholds its seed was judged by, and
+`internal/api/router_helpers.go` hands a tenant monitor's seed and thresholds
+to it.
+Both are read-path changes to source freshness and presentation; no storage,
+retention, backup or recovery path is added or moved.
+
 ### Webhook configuration never persists the API mask
 
 `internal/api/alerting/notifications.go` resolves each `***REDACTED***`

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

@@ -1615,6 +1615,45 @@ the stale pass's own threshold rule, so the merge's freshness gate never
 calls a source current that the stale pass marks stale, or the reverse. The
 gate decides only where no stronger rule does: a hypervisor-managed guest
 keeps its platform CPU even while that source is stale.
+The resources API's registry rebuild is resource seeding too, and the seed
+carries the thresholds its own registry generation judged it by. The monitor
+adapter records them per generation (`MonitorAdapter.StaleThresholds`):
+thresholds set since a rebuild wait for the next one, and the read-state
+overlays and live supplemental refreshes of a generation are judged by that
+generation's thresholds, since a listing already marked stale cannot be
+cleared by re-judging it. A monitor view lists the store's current
+generation together with those thresholds
+(`GetAllWithMetricsTargetsAndStaleThresholds` reads both from one registry
+pointer); a view the monitor builds from mock fixtures or a bare state
+snapshot was judged by the defaults and carries none. `UnifiedResourceSnapshotWithStaleThresholds`
+(per tenant, `UnifiedResourceSnapshotWithStaleThresholdsForTenant`) hands the
+view's listing, freshness marker and thresholds to the resources API, which
+builds its registry with `NewRegistryWithStaleThresholds`: every ingest or
+stale pass on that registry that passes no thresholds of its own uses them,
+and a cached registry built with other thresholds is rebuilt even at the same
+freshness marker. Presentation coalescing picks the source of a merged host
+row's readings with the metric merge's own rule (`metricSourceStale`):
+`ListForPresentation` passes the thresholds its registry was constructed
+with, and the websocket broadcast passes its view's, so on mock and snapshot
+views both follow the defaults and still agree. Before this the resources API judged by
+the defaults everywhere, so a Proxmox node polled every two minutes read
+stale, with warning status, in `/api/resources` ninety seconds after its poll
+while the broadcast called it online, and both views showed a silent agent's
+readings on a node whose agent keeps its own registry row. Patrol's scoped
+runtime state still rebuilds registries with the defaults. Regression
+coverage: `TestRegistrySeededFromMonitorJudgesFreshnessByItsThresholds` and
+`TestRegistryIngestThresholdsOverrideConstructionThresholds` in
+`registry_merge_policy_test.go`,
+`TestMonitorAdapterReportsTheThresholdsItsGenerationWasJudgedBy` in
+`monitor_adapter_read_state_test.go`,
+`TestUnifiedViewCarriesItsGenerationsStaleThresholds` in
+`internal/monitoring/monitor_host_agents_test.go`,
+`TestResourceAPIRebuildsWithTheSeedingMonitorStaleThresholds` in
+`code_standards_test.go`,
+`TestResourceListRebuildsWhenSeedStaleThresholdsChange` in
+`internal/api/resourceapi/resources_test.go`, and
+`TestContract_ResourceListJudgesFreshnessLikeTheMonitorBroadcast` in
+`internal/api/contract_test.go`.
 
 Service-discovery readiness is a unified-resource payload contract, not a
 drawer-local decoration. Resource list/detail payloads that expose a

internal/api/contract_test.go

@@ -14983,6 +14983,182 @@ func TestContract_ResourceListUsesTenantStateProviderAtStartup(t *testing.T) {
 	}
 }
 
+// The resources API rebuilds its registry from a monitor's unified seed, so it
+// judges source freshness by the thresholds that seed's registry judged it by,
+// for the default monitor and a tenant monitor alike. A Proxmox estate polled
+// every two minutes, read ninety seconds after its last poll, lists as the
+// websocket broadcast shows it: every node online on a current poll, with the
+// node's own readings over those of an agent silent for seventy-five seconds.
+// With the registry defaults the API called each poll stale (warning), and
+// both views showed the silent agent's readings on a node whose agent keeps
+// its own row. A monitor without a resource store builds its view with the
+// defaults, which call the ninety-second polls stale, and both views then
+// follow the defaults: no row takes a reading from a source its own sighting
+// calls stale over a live one.
+func TestContract_ResourceListJudgesFreshnessLikeTheMonitorBroadcast(t *testing.T) {
+	now := time.Now().UTC()
+	polledMonitor := func(t *testing.T, withStore bool) *monitoring.Monitor {
+		t.Helper()
+		monitor, state, _ := newTestMonitor(t)
+		setUnexportedField(t, monitor, "config", &config.Config{PVEPollingInterval: 2 * time.Minute})
+		polled := now.Add(-90 * time.Second)
+		agentSilent := now.Add(-75 * time.Second)
+		state.UpdateNodesForInstance("homelab", []models.Node{
+			{
+				ID: "homelab-pve1", Name: "pve1", Instance: "homelab", Status: "online",
+				LinkedAgentID: "host-pve1", CPU: 0.12,
+				Memory:   models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25},
+				LastSeen: polled,
+			},
+			{
+				ID: "homelab-pve2", Name: "pve2", Instance: "homelab", Status: "online", CPU: 0.30,
+				Memory:   models.Memory{Total: 64 << 30, Used: 32 << 30, Free: 32 << 30, Usage: 50},
+				LastSeen: polled,
+			},
+			{
+				ID: "homelab-pve3", Name: "pve3", Instance: "homelab", Status: "online", CPU: 0.20,
+				Memory:   models.Memory{Total: 64 << 30, Used: 8 << 30, Free: 56 << 30, Usage: 12.5},
+				LastSeen: polled,
+			},
+		})
+		state.UpsertHost(models.Host{
+			ID: "host-pve1", MachineID: "machine-pve1", Hostname: "pve1", LinkedNodeID: "homelab-pve1",
+			Status: "online", CPUUsage: 80,
+			Memory:   models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75},
+			LastSeen: agentSilent, IntervalSeconds: 30,
+		})
+		state.UpsertHost(models.Host{
+			ID: "host-pve3", MachineID: "machine-pve3", Hostname: "pve3",
+			Status: "online", CPUUsage: 70,
+			Memory:   models.Memory{Total: 64 << 30, Used: 40 << 30, Free: 24 << 30, Usage: 62.5},
+			LastSeen: agentSilent, IntervalSeconds: 30,
+		})
+		if withStore {
+			adapter := unifiedresources.NewMonitorAdapter(nil)
+			monitor.SetResourceStore(adapter)
+			adapter.PopulateFromSnapshot(state.GetSnapshot())
+		}
+		return monitor
+	}
+
+	type nodeRow struct {
+		status      string
+		cpu, memory float64
+	}
+	nodeNames := map[string]bool{"pve1": true, "pve2": true, "pve3": true}
+	broadcastRows := func(t *testing.T, monitor *monitoring.Monitor) map[string]nodeRow {
+		t.Helper()
+		rows := make(map[string]nodeRow)
+		for _, resource := range monitor.BuildBroadcastFrontendState().Resources {
+			if !nodeNames[resource.Name] {
+				continue
+			}
+			if _, dup := rows[resource.Name]; dup {
+				t.Fatalf("broadcast shows node %s twice", resource.Name)
+			}
+			row := nodeRow{status: resource.Status}
+			if row.status == "degraded" {
+				// The broadcast spells a host's warning status as degraded.
+				row.status = string(unifiedresources.StatusWarning)
+			}
+			if resource.CPU != nil {
+				row.cpu = resource.CPU.Current
+			}
+			if resource.Memory != nil {
+				row.memory = resource.Memory.Current
+			}
+			rows[resource.Name] = row
+		}
+		return rows
+	}
+	// listedRows also returns each row's Proxmox sighting, which the
+	// broadcast payload does not carry.
+	listedRows := func(t *testing.T, h *ResourceHandlers, orgID string) (map[string]nodeRow, map[string]string) {
+		t.Helper()
+		req := httptest.NewRequest(http.MethodGet, "/api/resources?type=agent", nil)
+		if orgID != "" {
+			req = req.WithContext(context.WithValue(req.Context(), OrgIDContextKey, orgID))
+		}
+		rec := httptest.NewRecorder()
+		h.HandleListResources(rec, req)
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
+		}
+		var response ResourcesResponse
+		if err := json.NewDecoder(rec.Body).Decode(&response); err != nil {
+			t.Fatalf("decode response: %v", err)
+		}
+		rows := make(map[string]nodeRow)
+		sightings := make(map[string]string)
+		for _, resource := range response.Data {
+			if !nodeNames[resource.Name] {
+				continue
+			}
+			if _, dup := rows[resource.Name]; dup {
+				t.Fatalf("resources API lists node %s twice", resource.Name)
+			}
+			row := nodeRow{status: string(resource.Status)}
+			if resource.Metrics != nil && resource.Metrics.CPU != nil {
+				row.cpu = resource.Metrics.CPU.Percent
+			}
+			if resource.Metrics != nil && resource.Metrics.Memory != nil {
+				row.memory = resource.Metrics.Memory.Percent
+			}
+			rows[resource.Name] = row
+			sightings[resource.Name] = resource.SourceStatus[unifiedresources.SourceProxmox].Status
+		}
+		return rows, sightings
+	}
+	current := map[string]nodeRow{
+		"pve1": {status: "online", cpu: 12, memory: 25},
+		"pve2": {status: "online", cpu: 30, memory: 50},
+		"pve3": {status: "online", cpu: 20, memory: 12.5},
+	}
+	check := func(t *testing.T, monitor *monitoring.Monitor, h *ResourceHandlers, orgID string, want map[string]nodeRow, wantSighting string) {
+		t.Helper()
+		broadcast := broadcastRows(t, monitor)
+		if !reflect.DeepEqual(broadcast, want) {
+			t.Fatalf("broadcast node rows = %+v, want %+v", broadcast, want)
+		}
+		listed, sightings := listedRows(t, h, orgID)
+		for name, sighting := range sightings {
+			if sighting != wantSighting {
+				t.Errorf("resources API %s Proxmox sighting = %q, want %q", name, sighting, wantSighting)
+			}
+		}
+		if !reflect.DeepEqual(listed, broadcast) {
+			t.Fatalf("resources API node rows = %+v, want the broadcast's %+v", listed, broadcast)
+		}
+	}
+
+	t.Run("default monitor", func(t *testing.T) {
+		monitor := polledMonitor(t, true)
+		h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()})
+		h.SetStateProvider(monitor)
+		check(t, monitor, h, "", current, "online")
+	})
+	t.Run("tenant monitor", func(t *testing.T) {
+		tenant := polledMonitor(t, true)
+		defaultMonitor, _, _ := newTestMonitor(t)
+		mtm := &monitoring.MultiTenantMonitor{}
+		setUnexportedField(t, mtm, "monitors", map[string]*monitoring.Monitor{"acme": tenant})
+		h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()})
+		h.SetStateProvider(defaultMonitor)
+		h.SetTenantStateProvider(NewMultiTenantStateProvider(mtm, defaultMonitor))
+		check(t, tenant, h, "acme", current, "online")
+	})
+	t.Run("monitor without a resource store", func(t *testing.T) {
+		monitor := polledMonitor(t, false)
+		h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()})
+		h.SetStateProvider(monitor)
+		check(t, monitor, h, "", map[string]nodeRow{
+			"pve1": {status: "warning", cpu: 80, memory: 75},
+			"pve2": {status: "warning", cpu: 30, memory: 50},
+			"pve3": {status: "warning", cpu: 70, memory: 62.5},
+		}, "stale")
+	})
+}
+
 func TestContract_ResourceCapabilitiesJSONSnapshot(t *testing.T) {
 	payload := struct {
 		ResourceID   string                                `json:"resourceId"`

internal/api/resourceapi/resources.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"maps"
 	"net/http"
 	"sort"
 	"strconv"
@@ -66,6 +67,22 @@ type UnifiedResourceSnapshotProvider interface {
 	UnifiedResourceSnapshot() ([]unified.Resource, time.Time)
 }
 
+// UnifiedResourceSnapshotWithStaleThresholdsProvider supplies the default
+// tenant's unified-resource seed with the stale thresholds the seed's registry
+// judged its sightings by (nil: the registry defaults). The registry rebuilt
+// from the seed runs its own stale pass, merges and presentation coalesce,
+// which must judge by the same thresholds or a source the monitor calls
+// current between configured polls reads stale here.
+type UnifiedResourceSnapshotWithStaleThresholdsProvider interface {
+	UnifiedResourceSnapshotWithStaleThresholds() ([]unified.Resource, time.Time, map[unified.DataSource]time.Duration)
+}
+
+// TenantUnifiedResourceSnapshotWithStaleThresholdsProvider is the
+// tenant-scoped form of UnifiedResourceSnapshotWithStaleThresholdsProvider.
+type TenantUnifiedResourceSnapshotWithStaleThresholdsProvider interface {
+	UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID string) ([]unified.Resource, time.Time, map[unified.DataSource]time.Duration)
+}
+
 // ResourceDiscoveryReadinessProvider projects service-discovery state onto a
 // unified resource without coupling the resource API to discovery storage.
 type ResourceDiscoveryReadinessProvider interface {
@@ -77,6 +94,9 @@ type registrySeed struct {
 	resources     []unified.Resource
 	lastUpdate    time.Time
 	unifiedSource bool
+	// staleThresholds are those the seed's registry judged it by; nil
+	// means the registry defaults.
+	staleThresholds map[unified.DataSource]time.Duration
 }
 
 // SupplementalRecordsProvider provides out-of-band ingest records for a specific source.
@@ -1087,7 +1107,8 @@ func (h *QueryService) buildRegistry(orgID string) (*unified.ResourceRegistry, e
 
 	h.cacheMu.Lock()
 	entry, ok := h.registryCache[key]
-	if ok && entry.registry != nil && !seed.lastUpdate.IsZero() && entry.lastUpdate.Equal(seed.lastUpdate) {
+	if ok && entry.registry != nil && !seed.lastUpdate.IsZero() && entry.lastUpdate.Equal(seed.lastUpdate) &&
+		maps.Equal(entry.staleThresholds, seed.staleThresholds) {
 		h.cacheMu.Unlock()
 		return entry.registry, nil
 	}
@@ -1100,7 +1121,11 @@ func (h *QueryService) buildRegistry(orgID string) (*unified.ResourceRegistry, e
 	}
 	h.supplementalMu.RUnlock()
 
-	registry := unified.NewRegistry(store)
+	// The rebuild runs its own stale pass, merges and presentation coalesce,
+	// so it judges freshness by the thresholds the seed was judged by; with
+	// the defaults a Proxmox node polled every two minutes read stale here
+	// between polls while the monitor called it current.
+	registry := unified.NewRegistryWithStaleThresholds(store, seed.staleThresholds)
 	ownedSources := supplementalSnapshotOwnedSources(supplementalProviders, orgID)
 	supplementalSources := sortedSupplementalSources(supplementalProviders)
 	if seed.unifiedSource {
@@ -1143,7 +1168,7 @@ func (h *QueryService) buildRegistry(orgID string) (*unified.ResourceRegistry, e
 	if h.registryCache == nil {
 		h.registryCache = make(map[string]registryCacheEntry)
 	}
-	h.registryCache[key] = registryCacheEntry{registry: registry, lastUpdate: seed.lastUpdate}
+	h.registryCache[key] = registryCacheEntry{registry: registry, lastUpdate: seed.lastUpdate, staleThresholds: seed.staleThresholds}
 	h.cacheMu.Unlock()
 
 	return registry, nil
@@ -1182,14 +1207,28 @@ func (h *QueryService) registrySeed(orgID string) (registrySeed, error) {
 		if h.tenantStateProvider == nil {
 			return seed, errors.New("tenant state provider unavailable")
 		}
-		resources, lastUpdate := h.tenantStateProvider.UnifiedResourceSnapshotForTenant(orgID)
-		seed.resources = resources
-		seed.lastUpdate = lastUpdate
+		if provider, ok := any(h.tenantStateProvider).(TenantUnifiedResourceSnapshotWithStaleThresholdsProvider); ok {
+			var thresholds map[unified.DataSource]time.Duration
+			seed.resources, seed.lastUpdate, thresholds = provider.UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID)
+			seed.staleThresholds = maps.Clone(thresholds)
+		} else {
+			seed.resources, seed.lastUpdate = h.tenantStateProvider.UnifiedResourceSnapshotForTenant(orgID)
+		}
 		seed.unifiedSource = true
 		return seed, nil
 	}
 
-	if provider, ok := any(h.stateProvider).(UnifiedResourceSnapshotProvider); ok {
+	if provider, ok := any(h.stateProvider).(UnifiedResourceSnapshotWithStaleThresholdsProvider); ok {
+		resources, lastUpdate, thresholds := provider.UnifiedResourceSnapshotWithStaleThresholds()
+		if len(resources) > 0 || !lastUpdate.IsZero() {
+			seed.resources = resources
+			seed.lastUpdate = lastUpdate
+			seed.unifiedSource = true
+			// Owned copy: the cache compares it with the next seed's.
+			seed.staleThresholds = maps.Clone(thresholds)
+			return seed, nil
+		}
+	} else if provider, ok := any(h.stateProvider).(UnifiedResourceSnapshotProvider); ok {
 		resources, lastUpdate := provider.UnifiedResourceSnapshot()
 		if len(resources) > 0 || !lastUpdate.IsZero() {
 			seed.resources = resources
@@ -1655,6 +1694,9 @@ type StorageIncidentSection struct {
 type registryCacheEntry struct {
 	registry   *unified.ResourceRegistry
 	lastUpdate time.Time
+	// staleThresholds are those the registry was built with; a seed judged
+	// by others needs a rebuild even at the same lastUpdate.
+	staleThresholds map[unified.DataSource]time.Duration
 	// rawList and presentation are lazily built once per registry generation
 	// and SHARED between requests. Both the slices and the nested data of
 	// their elements are read-only: take a flat copy (flatCopyResources)

internal/api/resourceapi/resources_test.go

@@ -694,6 +694,72 @@ func TestResourceListInvalidatesUnifiedSeedCacheOnFreshnessChange(t *testing.T)
 	}
 }
 
+type staleThresholdSeedProvider struct {
+	resources  []unified.Resource
+	freshness  time.Time
+	thresholds map[unified.DataSource]time.Duration
+}
+
+func (p *staleThresholdSeedProvider) ReadSnapshot() models.StateSnapshot {
+	return models.StateSnapshot{}
+}
+
+func (p *staleThresholdSeedProvider) UnifiedResourceSnapshotWithStaleThresholds() ([]unified.Resource, time.Time, map[unified.DataSource]time.Duration) {
+	out := make([]unified.Resource, len(p.resources))
+	copy(out, p.resources)
+	return out, p.freshness, p.thresholds
+}
+
+// A cached registry was built with one seed's stale thresholds; a seed judged
+// by others needs a rebuild even when its freshness marker is unchanged. The
+// provider changes its threshold map in place, so the cache must keep its own
+// copy to notice.
+func TestResourceListRebuildsWhenSeedStaleThresholdsChange(t *testing.T) {
+	now := time.Now().UTC()
+	polled := now.Add(-90 * time.Second)
+	provider := &staleThresholdSeedProvider{
+		resources: []unified.Resource{{
+			ID: "agent-pve2", Type: unified.ResourceTypeAgent, Name: "pve2", Status: unified.StatusOnline,
+			LastSeen: polled, UpdatedAt: polled,
+			Sources:      []unified.DataSource{unified.SourceProxmox},
+			SourceStatus: map[unified.DataSource]unified.SourceStatus{unified.SourceProxmox: {Status: "online", LastSeen: polled}},
+			Proxmox:      &unified.ProxmoxData{NodeName: "pve2", Instance: "homelab"},
+			Identity:     unified.ResourceIdentity{Hostnames: []string{"pve2"}},
+		}},
+		freshness:  now,
+		thresholds: map[unified.DataSource]time.Duration{unified.SourceProxmox: 4 * time.Minute},
+	}
+	h := NewQueryService(&config.Config{DataPath: t.TempDir()})
+	h.SetStateProvider(provider)
+
+	list := func(t *testing.T) unified.Resource {
+		t.Helper()
+		rec := httptest.NewRecorder()
+		h.HandleListResources(rec, httptest.NewRequest(http.MethodGet, "/api/resources?type=agent", nil))
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
+		}
+		var response ResourcesResponse
+		if err := json.NewDecoder(rec.Body).Decode(&response); err != nil {
+			t.Fatalf("decode response: %v", err)
+		}
+		if len(response.Data) != 1 {
+			t.Fatalf("resources = %#v, want the one node", response.Data)
+		}
+		return response.Data[0]
+	}
+
+	if got := list(t); got.Status != unified.StatusOnline || got.SourceStatus[unified.SourceProxmox].Status != "online" {
+		t.Fatalf("under a four-minute threshold: status %q, Proxmox sighting %q; want online for both",
+			got.Status, got.SourceStatus[unified.SourceProxmox].Status)
+	}
+	provider.thresholds[unified.SourceProxmox] = 60 * time.Second
+	if got := list(t); got.SourceStatus[unified.SourceProxmox].Status != "stale" {
+		t.Fatalf("under sixty seconds at the same freshness: Proxmox sighting %q, want stale; the cached registry was reused",
+			got.SourceStatus[unified.SourceProxmox].Status)
+	}
+}
+
 func TestResourceListMergesOneSidedLinkedHostWhenHostnameCorroborates(t *testing.T) {
 	now := time.Now().UTC()
 	node := models.Node{

internal/api/router_helpers.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/rcourtman/pulse-go-rewrite/internal/api/resourceapi"
 	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
 	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
 	"github.com/rs/zerolog/log"
@@ -80,6 +81,25 @@ func (p *MultiTenantStateProvider) UnifiedResourceSnapshotForTenant(orgID string
 	return monitor.UnifiedResourceSnapshot()
 }
 
+// The resources API reads both seeds-with-thresholds through optional
+// interface checks, so a signature drift would silently restore the registry
+// defaults.
+var (
+	_ resourceapi.UnifiedResourceSnapshotWithStaleThresholdsProvider       = (*monitoring.Monitor)(nil)
+	_ resourceapi.TenantUnifiedResourceSnapshotWithStaleThresholdsProvider = (*MultiTenantStateProvider)(nil)
+)
+
+// UnifiedResourceSnapshotWithStaleThresholdsForTenant returns a tenant's
+// unified-resource seed with the stale thresholds its registry judged it by,
+// read from one monitor in one call.
+func (p *MultiTenantStateProvider) UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID string) ([]unifiedresources.Resource, time.Time, map[unifiedresources.DataSource]time.Duration) {
+	monitor := p.monitorForTenant(orgID)
+	if monitor == nil {
+		return nil, time.Time{}, nil
+	}
+	return monitor.UnifiedResourceSnapshotWithStaleThresholds()
+}
+
 // ResourceHealthAlertsForTenant returns only the minimal alert facts needed by
 // the resource health projection, resolved from the same tenant monitor as the
 // resource seed.

internal/monitoring/canonical_guardrails_test.go

@@ -650,8 +650,9 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) {
 	for _, snippet := range []string{
 		"metricsTargetResolver := unifiedView.metricsTargets",
 		"metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)",
-		"broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources)",
-		"if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok {",
+		"broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources, unifiedView.staleThresholds)",
+		"if coalesced, ok := coalescer.CoalesceForPresentation(resources, staleThresholds); ok {",
+		"return unifiedresources.CoalescePresentationHostResourcesWithStaleThresholds(resources, staleThresholds)",
 		"broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources)",
 		"attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver)",
 		"frontendState.CapabilityCatalog = broadcastCatalogs.capabilities",

internal/monitoring/monitor.go

@@ -4479,7 +4479,9 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot mode
 	if metricsTargetResolver == nil {
 		metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)
 	}
-	broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources)
+	// Coalescing picks each merged metric's source by freshness, judged by
+	// the thresholds the view's own registry judged its sightings by.
+	broadcastResources := m.coalesceResourcesForPresentation(unifiedView.readState, unifiedView.resources, unifiedView.staleThresholds)
 	// Coalescing owns the outer slice. Decorate that one projection in place,
 	// not three full-resource copies; nested store data is still read-only.
 	healthAlerts := resourceHealthAlerts(frontendState.ActiveAlerts)
@@ -4505,7 +4507,7 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot mode
 // presentationCoalescingResourceStore is implemented by resource stores whose
 // registry can hold the operator's merge exclusions (MonitorAdapter).
 type presentationCoalescingResourceStore interface {
-	CoalesceForPresentation(resources []unifiedresources.Resource) ([]unifiedresources.Resource, bool)
+	CoalesceForPresentation(resources []unifiedresources.Resource, thresholds map[unifiedresources.DataSource]time.Duration) ([]unifiedresources.Resource, bool)
 }
 
 // coalesceResourcesForPresentation applies the presentation host coalesce with
@@ -4515,10 +4517,16 @@ type presentationCoalescingResourceStore interface {
 // the rows supplies them when its registry is store-backed, as the resource
 // store's adapter and a host-continuity overlay are; the mock view and other
 // views built from an unified list carry none, so the resource store's
-// adapter supplies them instead.
-func (m *Monitor) coalesceResourcesForPresentation(readState unifiedresources.ReadState, resources []unifiedresources.Resource) []unifiedresources.Resource {
+// adapter supplies them instead. staleThresholds are those the listing's
+// registry generation judged its sightings by; the coalesce judges each
+// merged metric's freshness by them.
+func (m *Monitor) coalesceResourcesForPresentation(
+	readState unifiedresources.ReadState,
+	resources []unifiedresources.Resource,
+	staleThresholds map[unifiedresources.DataSource]time.Duration,
+) []unifiedresources.Resource {
 	if coalescer, ok := readState.(presentationCoalescingResourceStore); ok {
-		if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok {
+		if coalesced, ok := coalescer.CoalesceForPresentation(resources, staleThresholds); ok {
 			return coalesced
 		}
 	}
@@ -4527,12 +4535,12 @@ func (m *Monitor) coalesceResourcesForPresentation(readState unifiedresources.Re
 		store := m.resourceStore
 		m.mu.RUnlock()
 		if coalescer, ok := store.(presentationCoalescingResourceStore); ok {
-			if coalesced, ok := coalescer.CoalesceForPresentation(resources); ok {
+			if coalesced, ok := coalescer.CoalesceForPresentation(resources, staleThresholds); ok {
 				return coalesced
 			}
 		}
 	}
-	return unifiedresources.CoalescePresentationHostResources(resources)
+	return unifiedresources.CoalescePresentationHostResourcesWithStaleThresholds(resources, staleThresholds)
 }
 
 // GetLiveStateSnapshot returns the underlying monitor state snapshot without
@@ -5086,6 +5094,9 @@ type monitorUnifiedStateView struct {
 	readState      unifiedresources.ReadState
 	freshness      time.Time
 	metricsTargets MetricsTargetResourceStore
+	// staleThresholds are those the listed resources' registry generation
+	// was judged by; nil means the registry defaults.
+	staleThresholds map[unifiedresources.DataSource]time.Duration
 }
 
 type unifiedResourceReadStateLister interface {
@@ -5155,9 +5166,10 @@ func (m *Monitor) unifiedStateViewWithStandaloneHostContinuity(view monitorUnifi
 		return view
 	}
 
-	resources, targets := unifiedProjectionResources(lister)
+	resources, targets, thresholds := unifiedProjectionResourcesWithStaleThresholds(lister)
 	view.resources = resources
 	view.metricsTargets = targets
+	view.staleThresholds = thresholds
 	if view.freshness.IsZero() {
 		view.freshness = latestUnifiedResourceLastSeen(resources)
 	}
@@ -5263,11 +5275,11 @@ func unifiedResourceFreshness(store ResourceStoreInterface, state *models.State)
 // associated freshness marker. In mock mode it returns the shared mock
 // unified-resource fixture graph rather than the live resource store.
 func (m *Monitor) UnifiedResourceSnapshot() ([]unifiedresources.Resource, time.Time) {
-	view := m.currentUnifiedStateView()
 	// REST /api/resources seeds its registry from this snapshot. Apply the
 	// same user-metadata hydration (container customUrl) as the websocket
 	// broadcast path, so the two payload shapes cannot drift.
-	return m.applyPersistedMetadataToUnifiedResources(view.resources), view.freshness
+	resources, freshness, _ := m.UnifiedResourceSnapshotWithStaleThresholds()
+	return resources, freshness
 }
 
 // GetUnifiedReadState returns a typed unified read-state provider when the

internal/monitoring/monitor_host_agents_test.go

@@ -279,6 +279,83 @@ func TestBroadcastSlimmingKeepsHostAgentSupersededIdentityResolvable(t *testing.
 	}
 }
 
+// The broadcast coalesces an agent that keeps its own registry row into its
+// Proxmox node by hostname, and picks each reading's source by freshness
+// judged by the configured thresholds, as the registry's own merges are. A
+// node polled every two minutes, read ninety seconds after its poll, shows its
+// own readings rather than those of its agent, silent for seventy-five
+// seconds. Polled at the default cadence, the node is stale at ninety seconds
+// too, so the agent's readings stand.
+func TestBroadcastCoalescedHostJudgesFreshnessByConfiguredThresholds(t *testing.T) {
+	now := time.Now().UTC()
+	for _, tc := range []struct {
+		name        string
+		interval    time.Duration
+		cpu, memory float64
+	}{
+		{"two-minute polls", 2 * time.Minute, 20, 12.5},
+		{"default polls", 10 * time.Second, 70, 62.5},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m := &Monitor{state: models.NewState(), config: &config.Config{PVEPollingInterval: tc.interval}}
+			m.state.UpdateNodesForInstance("homelab", []models.Node{{
+				ID: "homelab-pve3", Name: "pve3", Instance: "homelab", Status: "online", CPU: 0.20,
+				Memory:   models.Memory{Total: 64 << 30, Used: 8 << 30, Free: 56 << 30, Usage: 12.5},
+				LastSeen: now.Add(-90 * time.Second),
+			}})
+			m.state.UpsertHost(models.Host{
+				ID: "host-pve3", MachineID: "machine-pve3", Hostname: "pve3", Status: "online", CPUUsage: 70,
+				Memory:   models.Memory{Total: 64 << 30, Used: 40 << 30, Free: 24 << 30, Usage: 62.5},
+				LastSeen: now.Add(-75 * time.Second), IntervalSeconds: 30,
+			})
+			m.SetResourceStore(unifiedresources.NewMonitorAdapter(nil))
+
+			var rows []models.ResourceFrontend
+			for _, resource := range m.BuildBroadcastFrontendState().Resources {
+				if resource.Name == "pve3" {
+					rows = append(rows, resource)
+				}
+			}
+			if len(rows) != 1 {
+				t.Fatalf("broadcast shows %d pve3 rows, want the node and its agent coalesced into one", len(rows))
+			}
+			if rows[0].CPU == nil || rows[0].Memory == nil {
+				t.Fatalf("pve3 lost its readings: cpu %+v memory %+v", rows[0].CPU, rows[0].Memory)
+			}
+			if rows[0].CPU.Current != tc.cpu || rows[0].Memory.Current != tc.memory {
+				t.Fatalf("pve3 cpu %.1f%%, memory %.1f%%; want %.1f%% and %.1f%%",
+					rows[0].CPU.Current, rows[0].Memory.Current, tc.cpu, tc.memory)
+			}
+		})
+	}
+}
+
+// A monitor view pairs its listing with the thresholds that registry
+// generation was judged by. Thresholds set on the store since its last
+// rebuild belong to the next generation, so the view and the seed it hands
+// the resources API keep the ones the listing was judged by.
+func TestUnifiedViewCarriesItsGenerationsStaleThresholds(t *testing.T) {
+	now := time.Now().UTC()
+	m := &Monitor{state: models.NewState(), config: &config.Config{PVEPollingInterval: 2 * time.Minute}}
+	m.state.UpdateNodesForInstance("homelab", []models.Node{{
+		ID: "homelab-pve2", Name: "pve2", Instance: "homelab", Status: "online", CPU: 0.30,
+		Memory:   models.Memory{Total: 64 << 30, Used: 32 << 30, Free: 32 << 30, Usage: 50},
+		LastSeen: now.Add(-90 * time.Second),
+	}})
+	adapter := unifiedresources.NewMonitorAdapter(nil)
+	m.SetResourceStore(adapter)
+	adapter.PopulateFromSnapshot(m.state.GetSnapshot())
+	adapter.SetStaleThresholds(map[unifiedresources.DataSource]time.Duration{unifiedresources.SourceProxmox: time.Minute})
+
+	resources, _, thresholds := m.UnifiedResourceSnapshotWithStaleThresholds()
+	if got := thresholds[unifiedresources.SourceProxmox]; got != 4*time.Minute {
+		t.Fatalf("seed thresholds = %v, want the four minutes its generation was judged by", thresholds)
+	}
+	if len(resources) != 1 || resources[0].SourceStatus[unifiedresources.SourceProxmox].Status != "online" {
+		t.Fatalf("seed = %+v, want pve2 with the Proxmox sighting its generation judged online", resources)
+	}
+}
+
 func TestHostAgentCorrelationKeepsSameNamedProxmoxProvidersDistinct(t *testing.T) {
 	now := time.Now()
 	state := &models.State{
@@ -6926,6 +7003,10 @@ func (s *broadcastProjectionCountingStore) GetAllWithMetricsTargets() ([]unified
 	s.reads++
 	return s.MonitorAdapter.GetAllWithMetricsTargets()
 }
+func (s *broadcastProjectionCountingStore) GetAllWithMetricsTargetsAndStaleThresholds() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget, map[unifiedresources.DataSource]time.Duration) {
+	s.reads++
+	return s.MonitorAdapter.GetAllWithMetricsTargetsAndStaleThresholds()
+}
 func (*broadcastProjectionCountingStore) TryReplaceRegistryForRead(models.StateSnapshot, time.Duration, func() map[unifiedresources.DataSource][]unifiedresources.IngestRecord) bool {
 	return false
 }
@@ -7079,10 +7160,10 @@ func TestBroadcastPresentationCoalescePrefersTheListingReadState(t *testing.T) {
 	overlay := unifiedresources.NewRegistry(store)
 	overlay.IngestResources(resources)
 	listing := unifiedresources.NewMonitorAdapter(overlay)
-	if got := len(m.coalesceResourcesForPresentation(listing, listing.GetAll())); got != 2 {
+	if got := len(m.coalesceResourcesForPresentation(listing, listing.GetAll(), nil)); got != 2 {
 		t.Fatalf("coalesced %d rows with the listing's exclusion, want 2", got)
 	}
-	if got := len(m.coalesceResourcesForPresentation(nil, listing.GetAll())); got != 1 {
+	if got := len(m.coalesceResourcesForPresentation(nil, listing.GetAll(), nil)); got != 1 {
 		t.Fatalf("without a listing read state the store generation decides: %d rows, want 1", got)
 	}
 
@@ -7090,7 +7171,7 @@ func TestBroadcastPresentationCoalescePrefersTheListingReadState(t *testing.T) {
 	// operator decisions, so the resource store's adapter supplies them.
 	mockView := monitorUnifiedStateViewFromResources(resources, now)
 	withSplit := &Monitor{resourceStore: listing}
-	if got := len(withSplit.coalesceResourcesForPresentation(mockView.readState, mockView.resources)); got != 2 {
+	if got := len(withSplit.coalesceResourcesForPresentation(mockView.readState, mockView.resources, mockView.staleThresholds)); got != 2 {
 		t.Fatalf("mock-style view coalesced %d rows, want the store's split to keep 2", got)
 	}
 }

internal/monitoring/resource_stale_thresholds.go

@@ -57,6 +57,38 @@ func resourceStaleThresholdsForConfig(
 	return thresholds
 }
 
+// UnifiedResourceSnapshotWithStaleThresholds is UnifiedResourceSnapshot
+// plus the stale thresholds the snapshot's registry judged its sightings by.
+// A consumer that rebuilds a registry from the snapshot, as the resources API
+// does, judges by the same thresholds, so its stale pass and presentation
+// agree with the monitor's views.
+func (m *Monitor) UnifiedResourceSnapshotWithStaleThresholds() ([]unifiedresources.Resource, time.Time, map[unifiedresources.DataSource]time.Duration) {
+	view := m.currentUnifiedStateView()
+	return m.applyPersistedMetadataToUnifiedResources(view.resources), view.freshness, view.staleThresholds
+}
+
+// unifiedResourceThresholdProjectionLister lists a read state with the stale
+// thresholds the listed registry generation was judged by (MonitorAdapter,
+// and any wrapper embedding one).
+type unifiedResourceThresholdProjectionLister interface {
+	GetAllWithMetricsTargetsAndStaleThresholds() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget, map[unifiedresources.DataSource]time.Duration)
+}
+
+// unifiedProjectionResourcesWithStaleThresholds lists a read state with the
+// thresholds the listed generation was judged by: the resource store's
+// configured ones, or nil (the registry defaults) for a read state that
+// reports none, such as a view built from mock fixtures or a bare snapshot.
+func unifiedProjectionResourcesWithStaleThresholds(
+	lister unifiedResourceReadStateLister,
+) ([]unifiedresources.Resource, MetricsTargetResourceStore, map[unifiedresources.DataSource]time.Duration) {
+	if projection, ok := lister.(unifiedResourceThresholdProjectionLister); ok {
+		resources, targets, thresholds := projection.GetAllWithMetricsTargetsAndStaleThresholds()
+		return resources, projectionMetricsTargets(targets), thresholds
+	}
+	resources, targets := unifiedProjectionResources(lister)
+	return resources, targets, nil
+}
+
 func (m *Monitor) resourceStaleThresholds() map[unifiedresources.DataSource]time.Duration {
 	if m == nil {
 		return ResourceStaleThresholdsForConfig(nil)

internal/unifiedresources/code_standards_test.go

@@ -729,6 +729,34 @@ func TestResourceAPIUsesCanonicalTenantUnifiedSeed(t *testing.T) {
 	}
 }
 
+// The resources API rebuilds its registry from a monitor's seed, so the
+// rebuild must judge source freshness by the thresholds that seed was judged
+// by, delivered with the seed in one call, and must not reuse a registry built
+// with other thresholds. A bare NewRegistry falls back to the defaults and
+// marks a slowly polled source stale between its configured polls.
+func TestResourceAPIRebuildsWithTheSeedingMonitorStaleThresholds(t *testing.T) {
+	data, err := os.ReadFile(filepath.Join("..", "api", "resourceapi", "resources.go"))
+	if err != nil {
+		t.Fatalf("failed to read resources.go: %v", err)
+	}
+	source := string(data)
+
+	for _, snippet := range []string{
+		"registry := unified.NewRegistryWithStaleThresholds(store, seed.staleThresholds)",
+		"seed.resources, seed.lastUpdate, thresholds = provider.UnifiedResourceSnapshotWithStaleThresholdsForTenant(orgID)",
+		"resources, lastUpdate, thresholds := provider.UnifiedResourceSnapshotWithStaleThresholds()",
+		"seed.staleThresholds = maps.Clone(thresholds)",
+		"maps.Equal(entry.staleThresholds, seed.staleThresholds)",
+	} {
+		if !strings.Contains(source, snippet) {
+			t.Fatalf("internal/api/resourceapi/resources.go must contain %q", snippet)
+		}
+	}
+	if strings.Contains(source, "unified.NewRegistry(store)") {
+		t.Fatal("internal/api/resourceapi/resources.go must not rebuild a tenant registry with the default stale thresholds")
+	}
+}
+
 // Report-merge undoes an operator link through the registry's fold record. A
 // candidate ID derived from the merged resource's type names neither side of
 // a link, and the resources API's registry, seeded from the monitor's

internal/unifiedresources/monitor_adapter.go

@@ -26,7 +26,11 @@ type MonitorAdapter struct {
 	activeAlerts    []models.Alert
 	lastRebuiltAt   time.Time
 	staleThresholds map[DataSource]time.Duration
-	legacyHistory   legacyHistoryBackfill
+	// registryStaleThresholds are the thresholds the current registry
+	// generation was built and judged with. staleThresholds may already hold
+	// newer ones, which reach the registry at the next rebuild.
+	registryStaleThresholds map[DataSource]time.Duration
+	legacyHistory           legacyHistoryBackfill
 
 	overlays overlayReadStateCache
 }
@@ -64,8 +68,9 @@ func NewMonitorAdapterWithStaleThresholds(registry *ResourceRegistry, staleThres
 	}
 
 	return &MonitorAdapter{
-		registry:        registry,
-		staleThresholds: cloneStaleThresholds(staleThresholds),
+		registry:                registry,
+		staleThresholds:         cloneStaleThresholds(staleThresholds),
+		registryStaleThresholds: cloneStaleThresholds(staleThresholds),
 	}
 }
 
@@ -80,6 +85,20 @@ func (a *MonitorAdapter) SetStaleThresholds(staleThresholds map[DataSource]time.
 	a.mu.Unlock()
 }
 
+// StaleThresholds returns the source freshness thresholds the current
+// registry generation was judged by; nil means the registry defaults.
+// Thresholds set since that rebuild are not reported until the next one
+// judges a generation by them.
+func (a *MonitorAdapter) StaleThresholds() map[DataSource]time.Duration {
+	if a == nil {
+		return nil
+	}
+	a.mu.RLock()
+	thresholds := cloneStaleThresholds(a.registryStaleThresholds)
+	a.mu.RUnlock()
+	return thresholds
+}
+
 func (a *MonitorAdapter) currentStaleThresholds() map[DataSource]time.Duration {
 	if a == nil {
 		return nil
@@ -117,7 +136,10 @@ func readStateWithRecords(readState ReadState, source DataSource, records []Inge
 		return readState
 	}
 
-	registry := adapter.currentRegistry()
+	// The overlay re-judges the generation's listing, so it takes the
+	// thresholds that generation was judged by, not ones set since: those
+	// cannot clear a stale verdict the listing already carries.
+	registry, thresholds := adapter.currentRegistryAndStaleThresholds()
 	if registry == nil {
 		return readState
 	}
@@ -133,7 +155,6 @@ func readStateWithRecords(readState ReadState, source DataSource, records []Inge
 	}
 
 	cloned := NewRegistry(registry.store)
-	thresholds := adapter.currentStaleThresholds()
 	cloned.IngestResourcesWithStaleThresholds(registry.List(), thresholds)
 	cloned.ingestRecords(source, records, onlyMissing, thresholds)
 	overlay := NewMonitorAdapterWithStaleThresholds(cloned, thresholds)
@@ -184,6 +205,19 @@ func (c *overlayReadStateCache) store(registry *ResourceRegistry, rebuiltAt time
 	c.overlay = overlay
 }
 
+// currentRegistryAndStaleThresholds reads the current registry generation
+// together with the thresholds it was judged by.
+func (a *MonitorAdapter) currentRegistryAndStaleThresholds() (*ResourceRegistry, map[DataSource]time.Duration) {
+	if a == nil {
+		return nil, nil
+	}
+	a.mu.RLock()
+	registry := a.registry
+	thresholds := cloneStaleThresholds(a.registryStaleThresholds)
+	a.mu.RUnlock()
+	return registry, thresholds
+}
+
 func (a *MonitorAdapter) currentRegistry() *ResourceRegistry {
 	if a == nil {
 		return nil
@@ -496,6 +530,7 @@ func (a *MonitorAdapter) replaceRegistryLocked(snapshot models.StateSnapshot, re
 
 	a.mu.Lock()
 	a.registry = rebuilt
+	a.registryStaleThresholds = staleThresholds
 	a.activeAlerts = append([]models.Alert(nil), snapshot.ActiveAlerts...)
 	a.lastRebuiltAt = rebuiltAt
 	a.mu.Unlock()
@@ -571,6 +606,19 @@ func (a *MonitorAdapter) GetAllWithMetricsTargets() ([]Resource, map[string]Metr
 	return registry.ListWithMetricsTargets()
 }
 
+// GetAllWithMetricsTargetsAndStaleThresholds is GetAllWithMetricsTargets
+// plus the stale thresholds the listed registry generation was judged by,
+// read with the registry pointer, so a rebuild publishing meanwhile cannot
+// pair one generation's listing with another's thresholds.
+func (a *MonitorAdapter) GetAllWithMetricsTargetsAndStaleThresholds() ([]Resource, map[string]MetricsTarget, map[DataSource]time.Duration) {
+	registry, thresholds := a.currentRegistryAndStaleThresholds()
+	if registry == nil {
+		return nil, nil, nil
+	}
+	resources, targets := registry.ListWithMetricsTargets()
+	return resources, targets, thresholds
+}
+
 func (a *MonitorAdapter) unifiedAIAdapter() *UnifiedAIAdapter {
 	registry := a.currentRegistry()
 	if registry == nil {
@@ -684,12 +732,14 @@ func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records
 	a.mutationMu.Lock()
 	defer a.mutationMu.Unlock()
 
-	registry := a.currentRegistry()
+	// Records merge into the current generation, so they are judged by the
+	// thresholds it was; thresholds set since wait for the next rebuild.
+	registry, thresholds := a.currentRegistryAndStaleThresholds()
 	if registry == nil || len(records) == 0 || strings.TrimSpace(string(source)) == "" {
 		return
 	}
 	before := registry.List()
-	registry.IngestRecordsWithStaleThresholds(source, records, a.currentStaleThresholds())
+	registry.IngestRecordsWithStaleThresholds(source, records, thresholds)
 	rebuiltAt := time.Now().UTC()
 	recordRegistryChanges(registry.store, before, registry.List(), rebuiltAt, nil, SourcePlatformEvent, changeSourceAdapterForDataSource(source))
 	registry.PersistIdentityPins()

internal/unifiedresources/monitor_adapter_read_state_test.go

@@ -628,6 +628,61 @@ func TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds(t *testing.T) {
 	})
 }
 
+// The adapter reports the thresholds its current registry generation was
+// judged by. Thresholds set since then wait for the rebuild that judges a
+// generation by them, so a consumer rebuilding from the current listing never
+// re-judges it by thresholds it was not judged by. A read-state overlay of
+// that generation is judged by the same thresholds: re-judging its listing by
+// newer, shorter ones would mark polls stale that the generation calls
+// current.
+func TestMonitorAdapterReportsTheThresholdsItsGenerationWasJudgedBy(t *testing.T) {
+	snapshot := staleThresholdEstateSnapshot(time.Now().UTC())
+	configured := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute}
+	adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), configured)
+	adapter.PopulateFromSnapshot(snapshot)
+	if got := adapter.StaleThresholds(); got[SourceProxmox] != 4*time.Minute {
+		t.Fatalf("after the rebuild StaleThresholds = %v, want the four minutes it was judged by", got)
+	}
+
+	adapter.SetStaleThresholds(map[DataSource]time.Duration{SourceProxmox: 60 * time.Second})
+	if got := adapter.StaleThresholds(); got[SourceProxmox] != 4*time.Minute {
+		t.Fatalf("before the next rebuild StaleThresholds = %v, want the current generation's four minutes", got)
+	}
+	for name, row := range presentedNodeRows(t, adapter.GetAll()) {
+		if poll := row.SourceStatus[SourceProxmox]; poll.Status != "online" {
+			t.Fatalf("current generation %s Proxmox sighting = %q, want online as judged", name, poll.Status)
+		}
+	}
+	overlay, ok := ReadStateWithRecords(adapter, SourceAgent, []IngestRecord{{
+		SourceID: "host-elsewhere",
+		Resource: Resource{
+			Type: ResourceTypeAgent, Name: "elsewhere", Status: StatusOnline, LastSeen: time.Now().UTC(),
+			Agent: &AgentData{AgentID: "host-elsewhere", Hostname: "elsewhere"},
+		},
+		Identity: ResourceIdentity{MachineID: "machine-elsewhere", Hostnames: []string{"elsewhere"}},
+	}}).(*MonitorAdapter)
+	if !ok || overlay == adapter {
+		t.Fatal("ReadStateWithRecords did not build an overlay")
+	}
+	if got := overlay.StaleThresholds(); got[SourceProxmox] != 4*time.Minute {
+		t.Fatalf("overlay StaleThresholds = %v, want the generation's four minutes", got)
+	}
+	overlayRows := presentedNodeRows(t, overlay.GetAll())
+	if len(overlayRows) != 3 {
+		t.Fatalf("overlay lists %d Proxmox node rows, want 3", len(overlayRows))
+	}
+	for name, row := range overlayRows {
+		if poll := row.SourceStatus[SourceProxmox]; poll.Status != "online" {
+			t.Fatalf("overlay %s Proxmox sighting = %q, want online as the generation judged it", name, poll.Status)
+		}
+	}
+
+	adapter.PopulateFromSnapshot(snapshot)
+	if got := adapter.StaleThresholds(); got[SourceProxmox] != 60*time.Second {
+		t.Fatalf("after the next rebuild StaleThresholds = %v, want the sixty seconds it was judged by", got)
+	}
+}
+
 func TestHostContinuityCannotOverwriteCurrentResource(t *testing.T) {
 	now := time.Now().UTC()
 	for _, mode := range []string{"machine-identity", "source-identity", "future-timestamp", "provider-only"} {

internal/unifiedresources/presentation_coalesce.go

@@ -11,7 +11,18 @@ import (
 // monitored host when a runtime/platform view and the Pulse agent view share a
 // canonical hostname.
 func CoalescePresentationHostResources(resources []Resource) []Resource {
-	return CoalescePresentationHostResourcesWithExclusions(resources, nil)
+	return coalescePresentationHostResources(resources, nil, nil)
+}
+
+// CoalescePresentationHostResourcesWithStaleThresholds applies the
+// presentation host coalesce, judging which source's metrics are current by
+// the caller's stale thresholds: those the registry that produced the
+// resources judged its sightings by.
+func CoalescePresentationHostResourcesWithStaleThresholds(
+	resources []Resource,
+	thresholds map[DataSource]time.Duration,
+) []Resource {
+	return coalescePresentationHostResources(resources, nil, thresholds)
 }
 
 // CoalescePresentationHostResourcesWithExclusions applies the presentation
@@ -20,9 +31,17 @@ func CoalescePresentationHostResourcesWithExclusions(
 	resources []Resource,
 	excluded func(left, right Resource) bool,
 ) []Resource {
-	coalesced := coalescePresentationHostResourcesOnce(resources, excluded)
+	return coalescePresentationHostResources(resources, excluded, nil)
+}
+
+func coalescePresentationHostResources(
+	resources []Resource,
+	excluded func(left, right Resource) bool,
+	thresholds map[DataSource]time.Duration,
+) []Resource {
+	coalesced := coalescePresentationHostResourcesOnce(resources, excluded, thresholds)
 	for len(coalesced) < len(resources) {
-		next := coalescePresentationHostResourcesOnce(coalesced, excluded)
+		next := coalescePresentationHostResourcesOnce(coalesced, excluded, thresholds)
 		if len(next) == len(coalesced) {
 			return refreshPresentationProxmoxChildActionAgents(next)
 		}
@@ -35,6 +54,7 @@ func CoalescePresentationHostResourcesWithExclusions(
 func coalescePresentationHostResourcesOnce(
 	resources []Resource,
 	excluded func(left, right Resource) bool,
+	thresholds map[DataSource]time.Duration,
 ) []Resource {
 	if len(resources) == 0 {
 		return resources
@@ -74,7 +94,7 @@ func coalescePresentationHostResourcesOnce(
 			if !shouldMergePresentationHostResources(existing, resource) {
 				continue
 			}
-			mergedResource := mergePresentationHostResources(existing, resource)
+			mergedResource := mergePresentationHostResources(existing, resource, thresholds)
 			coalesced[existingIndex] = mergedResource
 			addPresentationParentRedirect(parentRedirects, existing.ID, mergedResource.ID)
 			addPresentationParentRedirect(parentRedirects, resource.ID, mergedResource.ID)
@@ -414,7 +434,7 @@ func mergePresentationSources(left, right []DataSource) []DataSource {
 	return merged
 }
 
-func mergePresentationHostResources(left, right Resource) Resource {
+func mergePresentationHostResources(left, right Resource, thresholds map[DataSource]time.Duration) Resource {
 	primary, secondary := left, right
 	if preferPresentationHostPrimary(right, left) {
 		primary, secondary = right, left
@@ -467,8 +487,12 @@ func mergePresentationHostResources(left, right Resource) Resource {
 		for source, status := range primary.SourceStatus {
 			combinedStatus[source] = status
 		}
+		// Judged by the producing registry's thresholds and the rule its
+		// stale pass and metric merge read, so where both rows carry a
+		// reading this merge prefers a current source over a stale one as
+		// the registry's own merges do.
 		stale := func(source DataSource) bool {
-			return presentationSourceStale(now, combinedStatus, source)
+			return metricSourceStale(now, combinedStatus, source, thresholds)
 		}
 		merged.Metrics = mergePresentationMetrics(merged.Metrics, secondary.Metrics, stale)
 	}
@@ -568,24 +592,6 @@ func uniquePresentationStrings(values []string) []string {
 	return unique
 }
 
-// presentationSourceStale reports whether a source's most recent report is
-// older than its stale threshold. A zero/unknown last-seen is treated as NOT
-// stale so coalescing never demotes a source on missing information.
-func presentationSourceStale(now time.Time, status map[DataSource]SourceStatus, source DataSource) bool {
-	if status == nil {
-		return false
-	}
-	st, ok := status[source]
-	if !ok || st.LastSeen.IsZero() {
-		return false
-	}
-	threshold, ok := defaultStaleThresholds[source]
-	if !ok {
-		threshold = 60 * time.Second
-	}
-	return now.Sub(st.LastSeen) > threshold
-}
-
 // mergePresentationMetric keeps the primary (left) value, except that a metric
 // from a stale source must never win over a metric from a live one. Without
 // this, a Proxmox node whose Pulse Agent has gone offline shows the agent's

internal/unifiedresources/presentation_coalesce_test.go

@@ -390,7 +390,7 @@ func TestMonitorAdapterCoalesceForPresentationHonoursExclusions(t *testing.T) {
 		if split {
 			want = 2
 		}
-		rows, ok := adapter.CoalesceForPresentation(adapter.GetAll())
+		rows, ok := adapter.CoalesceForPresentation(adapter.GetAll(), nil)
 		if !ok || len(rows) != want {
 			t.Fatalf("split=%v: adapter coalesced %d rows (ok=%v), want %d", split, len(rows), ok, want)
 		}
@@ -403,7 +403,7 @@ func TestMonitorAdapterCoalesceForPresentationHonoursExclusions(t *testing.T) {
 	// rather than coalescing as if none existed.
 	storeless := NewRegistry(nil)
 	storeless.IngestResources(resources)
-	if _, ok := NewMonitorAdapter(storeless).CoalesceForPresentation(resources); ok {
+	if _, ok := NewMonitorAdapter(storeless).CoalesceForPresentation(resources, nil); ok {
 		t.Fatal("a store-less adapter claimed to carry operator decisions")
 	}
 }

internal/unifiedresources/presentation_exclusions.go

@@ -1,5 +1,7 @@
 package unifiedresources
 
+import "time"
+
 // presentationExclusionFilter reports whether the presentation host coalesce
 // must keep two resources apart because the operator split the pair (unlink
 // or report-merge). It reads a copy of the registry's exclusions, so callers
@@ -31,14 +33,15 @@ func (rr *ResourceRegistry) presentationExclusionFilter() func(left, right Resou
 // listed from this adapter's current registry generation, honouring the
 // operator's merge exclusions through the same filter that
 // ResourceRegistry.ListForPresentation uses, so the websocket broadcast keeps
-// a pair the operator split apart as the resources API does. It reports
-// false, leaving resources untouched, when the registry has no store and so
-// carries no operator decisions: the mock view and other read states built
-// from an already-unified list.
-func (a *MonitorAdapter) CoalesceForPresentation(resources []Resource) ([]Resource, bool) {
+// a pair the operator split apart as the resources API does. Metric
+// freshness is judged by thresholds: those the listing's registry generation
+// judged its sightings by. It reports false, leaving resources untouched,
+// when the registry has no store and so carries no operator decisions: the
+// mock view and other read states built from an already-unified list.
+func (a *MonitorAdapter) CoalesceForPresentation(resources []Resource, thresholds map[DataSource]time.Duration) ([]Resource, bool) {
 	registry := a.currentRegistry()
 	if registry == nil || registry.store == nil {
 		return resources, false
 	}
-	return CoalescePresentationHostResourcesWithExclusions(resources, registry.presentationExclusionFilter()), true
+	return coalescePresentationHostResources(resources, registry.presentationExclusionFilter(), thresholds), true
 }

internal/unifiedresources/registry.go

@@ -177,6 +177,12 @@ type ResourceRegistry struct {
 	// stale pass do. nil outside an ingest and for callers without
 	// thresholds, where all three use the defaults.
 	ingestStaleThresholds map[DataSource]time.Duration
+
+	// staleThresholds are the freshness thresholds of the monitor whose
+	// resources this registry holds, fixed at construction. An ingest or
+	// stale pass without thresholds of its own and the presentation listing
+	// judge sightings by them; nil means the defaults.
+	staleThresholds map[DataSource]time.Duration
 }
 
 // agentNodeCandidate pairs a resources-map key with its entry so the indexed
@@ -215,6 +221,13 @@ func (rr *ResourceRegistry) ensureCanonicalMetadataLocked() {
 
 // NewRegistry creates a new registry using the provided store for overrides.
 func NewRegistry(store ResourceStore) *ResourceRegistry {
+	return NewRegistryWithStaleThresholds(store, nil)
+}
+
+// NewRegistryWithStaleThresholds creates a registry that judges source
+// freshness by the given thresholds wherever a caller passes none, for a
+// registry rebuilt from a monitor's resources, such as the resources API's.
+func NewRegistryWithStaleThresholds(store ResourceStore, thresholds map[DataSource]time.Duration) *ResourceRegistry {
 	rr := &ResourceRegistry{
 		resources:              make(map[string]*Resource),
 		bySource:               make(map[DataSource]map[string]string),
@@ -222,6 +235,7 @@ func NewRegistry(store ResourceStore) *ResourceRegistry {
 		store:                  store,
 		exclusions:             make(map[string]struct{}),
 		canonicalMetadataDirty: true,
+		staleThresholds:        cloneStaleThresholds(thresholds),
 	}
 
 	rr.bySource[SourceProxmox] = make(map[string]string)
@@ -275,6 +289,7 @@ func (rr *ResourceRegistry) IngestSnapshotWithStaleThresholds(snapshot models.St
 }
 
 func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresholds map[DataSource]time.Duration) {
+	thresholds = rr.thresholdsOrOwn(thresholds)
 	rr.mu.Lock()
 	rr.ingestStaleThresholds = thresholds
 	rr.mu.Unlock()
@@ -658,6 +673,7 @@ func (rr *ResourceRegistry) IngestRecordsWithStaleThresholds(source DataSource,
 }
 
 func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool, thresholds map[DataSource]time.Duration) {
+	thresholds = rr.thresholdsOrOwn(thresholds)
 	rr.mu.Lock()
 	rr.ingestStaleThresholds = thresholds
 	rr.mu.Unlock()
@@ -916,6 +932,7 @@ func (rr *ResourceRegistry) IngestResourcesWithStaleThresholds(resources []Resou
 }
 
 func (rr *ResourceRegistry) ingestResources(resources []Resource, thresholds map[DataSource]time.Duration) {
+	thresholds = rr.thresholdsOrOwn(thresholds)
 	seededIDs := make([]string, 0, len(resources))
 	for _, incoming := range resources {
 		resource := cloneResourcePtr(&incoming)
@@ -1503,10 +1520,11 @@ func (rr *ResourceRegistry) listMaterialized(withTargets bool) ([]Resource, map[
 
 // ListForPresentation returns resources in the canonical API/broadcast
 // presentation shape, including top-level host coalescing that respects manual
-// merge exclusions.
+// merge exclusions and judges source freshness by the thresholds the registry
+// was constructed with (not those an individual ingest passed).
 func (rr *ResourceRegistry) ListForPresentation() []Resource {
 	resources := rr.List()
-	return CoalescePresentationHostResourcesWithExclusions(resources, rr.presentationExclusionFilter())
+	return coalescePresentationHostResources(resources, rr.presentationExclusionFilter(), rr.staleThresholds)
 }
 
 // ListByType returns all resources of the provided type.
@@ -1865,13 +1883,22 @@ func (rr *ResourceRegistry) Stats() ResourceStats {
 }
 
 // MarkStale marks sources as stale based on last seen timestamps.
-// If thresholds is nil, default thresholds are used.
+// If thresholds is nil, the registry's own thresholds are used.
 func (rr *ResourceRegistry) MarkStale(now time.Time, thresholds map[DataSource]time.Duration) {
+	thresholds = rr.thresholdsOrOwn(thresholds)
 	rr.mu.Lock()
 	defer rr.mu.Unlock()
 	rr.markStaleLocked(now, thresholds)
 }
 
+// thresholdsOrOwn returns a caller's thresholds, else the registry's own.
+func (rr *ResourceRegistry) thresholdsOrOwn(thresholds map[DataSource]time.Duration) map[DataSource]time.Duration {
+	if thresholds != nil {
+		return thresholds
+	}
+	return rr.staleThresholds
+}
+
 // sourceSightingStatus derives the per-source delivery status from the last
 // time that source actually delivered the resource. A zero sighting means the
 // source has never delivered it (synthesized offline placeholders, instances

internal/unifiedresources/registry_merge_policy_test.go

@@ -356,6 +356,158 @@ func TestMetricMergeFreshnessMatchesTheStalePass(t *testing.T) {
 	}
 }
 
+// staleThresholdEstateSnapshot is a Proxmox estate polled every two minutes,
+// read ninety seconds after its last poll: a node merged with its auto-linked
+// agent, silent for seventy-five seconds; a node with no agent; and a node
+// whose agent the registry keeps as its own row, which presentation coalesces
+// into the node by hostname.
+func staleThresholdEstateSnapshot(now time.Time) models.StateSnapshot {
+	polled := now.Add(-90 * time.Second)
+	agentSilent := now.Add(-75 * time.Second)
+	return models.StateSnapshot{
+		Nodes: []models.Node{
+			{
+				ID: "homelab-pve1", Name: "pve1", Instance: "homelab", Status: "online",
+				LinkedAgentID: "host-pve1", CPU: 0.12,
+				Memory:   models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25},
+				LastSeen: polled,
+			},
+			{
+				ID: "homelab-pve2", Name: "pve2", Instance: "homelab", Status: "online", CPU: 0.30,
+				Memory:   models.Memory{Total: 64 << 30, Used: 32 << 30, Free: 32 << 30, Usage: 50},
+				LastSeen: polled,
+			},
+			{
+				ID: "homelab-pve3", Name: "pve3", Instance: "homelab", Status: "online", CPU: 0.20,
+				Memory:   models.Memory{Total: 64 << 30, Used: 8 << 30, Free: 56 << 30, Usage: 12.5},
+				LastSeen: polled,
+			},
+		},
+		Hosts: []models.Host{
+			{
+				ID: "host-pve1", MachineID: "machine-pve1", Hostname: "pve1", LinkedNodeID: "homelab-pve1",
+				Status: "online", CPUUsage: 80,
+				Memory:   models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75},
+				LastSeen: agentSilent, IntervalSeconds: 30,
+			},
+			{
+				ID: "host-pve3", MachineID: "machine-pve3", Hostname: "pve3",
+				Status: "online", CPUUsage: 70,
+				Memory:   models.Memory{Total: 64 << 30, Used: 40 << 30, Free: 24 << 30, Usage: 62.5},
+				LastSeen: agentSilent, IntervalSeconds: 30,
+			},
+		},
+		LastUpdate: now,
+	}
+}
+
+// presentedNodeRows indexes the coalesced rows that carry a Proxmox sighting
+// by name.
+func presentedNodeRows(t *testing.T, resources []Resource) map[string]Resource {
+	t.Helper()
+	rows := make(map[string]Resource)
+	for _, resource := range resources {
+		if _, ok := resource.SourceStatus[SourceProxmox]; !ok {
+			continue
+		}
+		if _, dup := rows[resource.Name]; dup {
+			t.Fatalf("two presented rows carry Proxmox node %q", resource.Name)
+		}
+		rows[resource.Name] = resource
+	}
+	return rows
+}
+
+// A registry rebuilt from the monitor's resources, as the resources API
+// rebuilds one, judges source freshness by the monitor's configured
+// thresholds: its stale pass, and the presentation coalesce of both views,
+// agree with the monitor's own registry. With the defaults the rebuild called
+// every node's poll stale at ninety seconds (warning status), and both views
+// showed the silent agent's readings on the node whose agent stays its own
+// row.
+func TestRegistrySeededFromMonitorJudgesFreshnessByItsThresholds(t *testing.T) {
+	now := time.Now().UTC()
+	thresholds := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute}
+	snapshot := staleThresholdEstateSnapshot(now)
+
+	adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), thresholds)
+	adapter.PopulateFromSnapshot(snapshot)
+	adapter.PopulateFromSnapshot(snapshot)
+
+	broadcast := presentedNodeRows(t, CoalescePresentationHostResourcesWithStaleThresholds(adapter.GetAll(), thresholds))
+	seeded := NewRegistryWithStaleThresholds(nil, thresholds)
+	seeded.IngestResources(adapter.GetAll())
+	api := presentedNodeRows(t, seeded.ListForPresentation())
+
+	want := map[string]struct {
+		cpu, memory float64
+		agent       bool
+	}{
+		"pve1": {cpu: 12, memory: 25, agent: true},
+		"pve2": {cpu: 30, memory: 50},
+		"pve3": {cpu: 20, memory: 12.5, agent: true},
+	}
+	for view, rows := range map[string]map[string]Resource{"broadcast": broadcast, "resources API": api} {
+		if len(rows) != len(want) {
+			t.Fatalf("%s presents %d node rows, want %d", view, len(rows), len(want))
+		}
+		for name, expected := range want {
+			row, ok := rows[name]
+			if !ok {
+				t.Fatalf("%s lost node %s", view, name)
+			}
+			if poll := row.SourceStatus[SourceProxmox]; poll.Status != "online" {
+				t.Errorf("%s %s Proxmox sighting = %q, want online: ninety seconds is within the configured four minutes", view, name, poll.Status)
+			}
+			if row.Status != StatusOnline {
+				t.Errorf("%s %s status = %q, want online", view, name, row.Status)
+			}
+			if agent, ok := row.SourceStatus[SourceAgent]; ok != expected.agent || (ok && agent.Status != "stale") {
+				t.Errorf("%s %s agent sighting = %+v (present %v), want a stale one present %v", view, name, agent, ok, expected.agent)
+			}
+			if row.Metrics == nil || row.Metrics.CPU == nil || row.Metrics.Memory == nil {
+				t.Fatalf("%s %s lost its readings: %+v", view, name, row.Metrics)
+			}
+			if cpu, memory := row.Metrics.CPU, row.Metrics.Memory; cpu.Source != SourceProxmox || cpu.Percent != expected.cpu ||
+				memory.Source != SourceProxmox || memory.Percent != expected.memory {
+				t.Errorf("%s %s cpu %.1f%% from %s, memory %.1f%% from %s; want the current poll's %.1f%% and %.1f%%",
+					view, name, cpu.Percent, cpu.Source, memory.Percent, memory.Source, expected.cpu, expected.memory)
+			}
+		}
+	}
+
+	// The fixture sits where the thresholds decide: a rebuild judged by the
+	// defaults calls the same polls stale.
+	defaults := NewRegistry(nil)
+	defaults.IngestResources(adapter.GetAll())
+	defaultRows := presentedNodeRows(t, defaults.ListForPresentation())
+	if len(defaultRows) != len(want) {
+		t.Fatalf("default-threshold rebuild presents %d node rows, want %d", len(defaultRows), len(want))
+	}
+	for name, row := range defaultRows {
+		if poll := row.SourceStatus[SourceProxmox]; poll.Status != "stale" {
+			t.Fatalf("default-threshold rebuild %s Proxmox sighting = %q, want stale; the fixture no longer exercises the thresholds", name, poll.Status)
+		}
+	}
+}
+
+// An ingest that passes thresholds of its own keeps them over the registry's.
+func TestRegistryIngestThresholdsOverrideConstructionThresholds(t *testing.T) {
+	now := time.Now().UTC()
+	snapshot := staleThresholdEstateSnapshot(now)
+	registry := NewRegistryWithStaleThresholds(nil, map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute})
+	registry.IngestSnapshotWithStaleThresholds(snapshot, map[DataSource]time.Duration{SourceProxmox: 60 * time.Second})
+	rows := presentedNodeRows(t, registry.List())
+	if len(rows) != 3 {
+		t.Fatalf("registry lists %d Proxmox node rows, want 3", len(rows))
+	}
+	for name, row := range rows {
+		if poll := row.SourceStatus[SourceProxmox]; poll.Status != "stale" {
+			t.Fatalf("%s Proxmox sighting = %q, want stale under the ingest's own sixty seconds", name, poll.Status)
+		}
+	}
+}
+
 // Docker hosts hold a shorter lease than the registry's Docker stale
 // threshold, so a silent Docker host was offline for a while and then
 // flipped to warning once its sighting went stale.