Refresh the resource store at most once per 2s of agent reports (#2242)

Every accepted agent report refreshed the canonical resource store, and
every refresh covers the whole estate, so refresh work grew with agent
count × estate size (#2199). #2235 made each pass linear and cheaper.
This PR bounds how many passes reports can trigger.

**Behaviour.** Accepted host, Docker and Kubernetes reports, including
rejected-report liveness, refresh through
`refreshUnifiedResourceStoreAfterAgentReport`:

- The first report after a quiet window is published before its handler
returns, the same as today.
- Reports within 2 s of the previous refresh fold into one trailing
refresh at the window's end. It reads the latest state and broadcasts it
to websocket clients.
- 2 s matches `readPathRegistryFreshness`, the staleness the read paths
already accept.
- Agent removal and host-agent evaluation still refresh immediately, so
removed inventory retires at once.
- `Monitor.Stop` cancels a pending trailing refresh and waits for a
running one before closing stores.
- Monitors built without `New` keep a zero window and refresh on every
report.

**Cost bound.** At 2,080 resources one refresh costs about 82 ms (after
#2235). 200 agents reporting every 30 s drive 6.7 refreshes/s, roughly
55% of a core; coalesced, reports trigger at most one refresh per 2 s,
about 4%, however many agents report.

**Contract change.** The agent-lifecycle contract promised that an
accepted report is visible when its handler returns. It is now visible
within the 2 s window. The founder approved that trade on 24 September
2026, and the contract records it.

**Tests**

- `TestAgentReportRefreshFoldsReportsWithinWindow`: the leading report
is published synchronously; a burst inside the window is deferred; one
trailing refresh publishes the burst's latest value; after a quiet
window the next report is synchronous again. It fails when the throttle
is removed.
- `TestStopCancelsPendingAgentReportRefresh`: fails if stop does not
cancel the pending refresh.
- The issue-1753 end-to-end test posts two reports back to back, so it
now waits for the published rows instead of reading them at once.

**Verification on pulse-dev**

- `internal/monitoring` runs in 112 s against 107 s on #2235's head. No
test is slower by more than 1.2 s.
- `internal/unifiedresources`, `internal/websocket` and
`internal/api/...` pass. `TestServerInfoEndpointReportsDevelopment`
fails in that source export with and without this change, because the
export has no git checkout.

Refs #2199

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

@@ -53,17 +53,25 @@ is exposed as unavailable, rather than replaced with a synthetic healthy state.
 
 Delivery-log diagnostic text is a notification-attempt projection, not an agent admission or liveness verdict. Masking embedded destination credentials preserves failure context without changing agent identities, credentials, session replacement or removal policy. Consumers must use the retained failureClass and outcome as delivery evidence only; a transport error does not establish that the monitored agent is offline.
 
-### Accepted reports refresh the canonical store synchronously — issue #2199
+### Accepted reports refresh the canonical store at most once per window — issue #2199
 
-An accepted host, Docker or Kubernetes report, and an agent removal, refresh the
-canonical resource store before the handler returns, so ReadState consumers see
-agent-backed truth without waiting for a broadcast. The #2199 reductions keep
-that contract: they make each refresh pass cheaper (node-indexed guest parents
-in the unified-resources contract, one registry snapshot per pass in the
-monitoring contract) without deferring or coalescing reports. A pass is still
-estate-wide, so its cost scales with agent report volume times estate size;
-coalescing reports would trade this immediate visibility and needs its own
-decision here first.
+Every canonical store refresh is estate-wide, so refreshing on each accepted
+report made total cost grow with agent count times estate size. Accepted host,
+Docker and Kubernetes reports, including rejected-report liveness, therefore
+refresh through `refreshUnifiedResourceStoreAfterAgentReport`, throttled to one
+refresh per `agentReportRefreshInterval` (2 seconds, the same staleness the
+read paths accept through `readPathRegistryFreshness`). The first report after
+a quiet window is published before the handler returns. Reports inside the
+window fold into one trailing refresh at the window's end, which reads the
+latest state and broadcasts it, so a report reaches ReadState consumers within
+the window rather than immediately. Agent removal and host-agent evaluation
+still refresh immediately, so removed inventory retires at once. `Monitor.Stop`
+cancels a pending trailing refresh and waits for a running one before closing
+stores. Monitors built without `New` keep a zero window and refresh on every
+report. `TestAgentReportRefreshFoldsReportsWithinWindow` and
+`TestStopCancelsPendingAgentReportRefresh` pin the leading, trailing and
+shutdown behaviour. The founder approved trading immediate visibility for this
+bound on 24 September 2026.
 
 ### Canonical Patrol and Assistant continuation, 2026-09-07
 

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

@@ -42,7 +42,8 @@ checks unchanged. `TestAgentReportRefreshClonesRegistryOnce` requires one clone
 per agent-report refresh and `TestResourceSnapshotStoreClonesOncePerPass` pins
 the wrapper. With node-indexed guest parents, a synthetic agent-report refresh
 at 2,080 resources moved from 188 to 82 ms; the pass still covers the whole
-estate, and it stays synchronous (see the agent-lifecycle contract).
+estate, and report-driven passes run at most once per 2-second window (see
+the agent-lifecycle contract).
 
 ### Linked Pulse agent memory for Proxmox LXC — issue #2148 (22 September 2026)
 

internal/api/issue1753_agent_ingest_integration_test.go

@@ -98,23 +98,43 @@ func TestIssue1753SameNameProxmoxAgentsAuthenticateAndStayDistinctEndToEnd(t *te
 		t.Fatalf("provider links = %+v, want each site linked to its own authenticated agent", linkedByInstance)
 	}
 
-	resources, _ := monitor.UnifiedResourceSnapshot()
-	byInstance := make(map[string]unifiedresources.Resource)
-	for _, resource := range resources {
-		if resource.Proxmox != nil && resource.Proxmox.NodeName != "" {
-			byInstance[resource.Proxmox.Instance] = resource
-		}
-	}
-	if len(byInstance) != 2 {
-		t.Fatalf("presentation provider rows = %d, want 2: %+v", len(byInstance), resources)
-	}
-	for instance, want := range map[string]struct {
+	wantRows := map[string]struct {
 		agentID string
 		name    string
 	}{
 		"staging":    {agentID: "machine-staging", name: "Staging"},
 		"production": {agentID: "machine-production", name: "Production"},
-	} {
+	}
+	presentationRows := func() ([]unifiedresources.Resource, map[string]unifiedresources.Resource) {
+		resources, _ := monitor.UnifiedResourceSnapshot()
+		byInstance := make(map[string]unifiedresources.Resource)
+		for _, resource := range resources {
+			if resource.Proxmox != nil && resource.Proxmox.NodeName != "" {
+				byInstance[resource.Proxmox.Instance] = resource
+			}
+		}
+		return resources, byInstance
+	}
+	rowsLinked := func(byInstance map[string]unifiedresources.Resource) bool {
+		for instance, want := range wantRows {
+			resource := byInstance[instance]
+			if resource.Agent == nil || resource.Agent.AgentID != want.agentID {
+				return false
+			}
+		}
+		return true
+	}
+	// Reports inside one refresh window are published together by a trailing
+	// refresh (agent-lifecycle contract, #2199), so wait for the canonical rows.
+	resources, byInstance := presentationRows()
+	for deadline := time.Now().Add(5 * time.Second); !rowsLinked(byInstance) && time.Now().Before(deadline); {
+		time.Sleep(20 * time.Millisecond)
+		resources, byInstance = presentationRows()
+	}
+	if len(byInstance) != 2 {
+		t.Fatalf("presentation provider rows = %d, want 2: %+v", len(byInstance), resources)
+	}
+	for instance, want := range wantRows {
 		resource := byInstance[instance]
 		if resource.Agent == nil || resource.Agent.AgentID != want.agentID || resource.Name != want.name {
 			t.Fatalf("%s presentation row = %+v, want agent %q and name %q", instance, resource, want.agentID, want.name)

internal/monitoring/kubernetes_agents.go

@@ -344,7 +344,7 @@ func (m *Monitor) ApplyKubernetesReport(report agentsk8s.Report, tokenRecord *co
 	m.state.UpsertKubernetesCluster(cluster)
 	m.state.SetConnectionHealth(kubernetesConnectionPrefix+identifier, true)
 	m.recordKubernetesPodMetrics(cluster, receivedAt)
-	m.refreshUnifiedResourceStoreAfterAgentStateChange()
+	m.refreshUnifiedResourceStoreAfterAgentReport()
 
 	return cluster, nil
 }

internal/monitoring/monitor.go

@@ -1301,6 +1301,11 @@ type Monitor struct {
 	// data, which is the dominant write amplification in #1966.
 	unifiedMetricSyncMu   sync.Mutex
 	unifiedMetricSyncLast map[unifiedMetricSampleKey]unifiedMetricSample
+
+	// agentReportRefreshWindow folds accepted agent reports into at most one
+	// canonical store refresh per window; zero refreshes on every report.
+	agentReportRefreshWindow time.Duration
+	agentReportRefresh       agentReportRefreshState
 }
 
 func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) {
@@ -1738,6 +1743,7 @@ func New(cfg *config.Config) (*Monitor, error) {
 	m := &Monitor{
 		config:                     cfg,
 		state:                      models.NewState(),
+		agentReportRefreshWindow:   agentReportRefreshInterval,
 		pveClients:                 make(map[string]PVEClientInterface),
 		pbsClients:                 make(map[string]*pbs.Client),
 		pmgClients:                 make(map[string]*pmg.Client),
@@ -5238,8 +5244,9 @@ func (s *resourceSnapshotStore) MetricsTargetForResource(resourceID string) *uni
 	return s.targets.MetricsTargetForResource(resourceID)
 }
 
-// refreshUnifiedResourceStoreAfterAgentStateChange makes accepted agent
-// ingest and removal immediately visible to canonical ReadState consumers.
+// refreshUnifiedResourceStoreAfterAgentStateChange makes agent removal and
+// host-agent evaluation immediately visible to canonical ReadState consumers;
+// accepted reports go through refreshUnifiedResourceStoreAfterAgentReport.
 // WebSocket broadcasts may also rebuild the store for their own hydrate path,
 // but client presence must never be the trigger that publishes agent-backed
 // runtime truth or retires removed inventory.
@@ -5250,6 +5257,105 @@ func (m *Monitor) refreshUnifiedResourceStoreAfterAgentStateChange() {
 	m.updateResourceStore(m.GetState())
 }
 
+// agentReportRefreshInterval bounds how often accepted agent reports refresh
+// the canonical store. It matches readPathRegistryFreshness, the staleness the
+// read paths already accept.
+const agentReportRefreshInterval = readPathRegistryFreshness
+
+// agentReportRefreshState throttles report-driven store refreshes to a leading
+// refresh plus at most one trailing refresh per window.
+type agentReportRefreshState struct {
+	mu       sync.Mutex
+	lastRun  time.Time
+	running  bool
+	pending  *time.Timer
+	stopped  bool
+	inFlight sync.WaitGroup
+}
+
+// refreshUnifiedResourceStoreAfterAgentReport publishes an accepted agent
+// report to the canonical store. Every refresh is estate-wide, so refreshing
+// on each report made total cost grow with agents times resources (#2199).
+// The first report after a quiet window still refreshes before the handler
+// returns. Reports within agentReportRefreshInterval of the previous refresh
+// fold into one trailing refresh at the window's end, which reads the latest
+// state and broadcasts it. Agent removal and host-agent evaluation keep
+// refreshing immediately through refreshUnifiedResourceStoreAfterAgentStateChange.
+func (m *Monitor) refreshUnifiedResourceStoreAfterAgentReport() {
+	if m == nil || m.state == nil {
+		return
+	}
+	window := m.agentReportRefreshWindow
+	if window <= 0 {
+		m.updateResourceStore(m.GetState())
+		return
+	}
+
+	r := &m.agentReportRefresh
+	r.mu.Lock()
+	if r.stopped || r.pending != nil {
+		// A due trailing refresh reads state after this report was applied.
+		r.mu.Unlock()
+		return
+	}
+	now := time.Now()
+	if !r.running && now.Sub(r.lastRun) >= window {
+		r.running = true
+		r.lastRun = now
+		r.mu.Unlock()
+		m.updateResourceStore(m.GetState())
+		r.mu.Lock()
+		r.running = false
+		r.mu.Unlock()
+		return
+	}
+	r.inFlight.Add(1)
+	r.pending = time.AfterFunc(max(window-now.Sub(r.lastRun), 0), m.runTrailingAgentReportRefresh)
+	r.mu.Unlock()
+}
+
+func (m *Monitor) runTrailingAgentReportRefresh() {
+	r := &m.agentReportRefresh
+	defer r.inFlight.Done()
+	r.mu.Lock()
+	r.pending = nil
+	if r.stopped {
+		r.mu.Unlock()
+		return
+	}
+	r.running = true
+	r.lastRun = time.Now()
+	r.mu.Unlock()
+
+	m.updateResourceStore(m.GetState())
+
+	r.mu.Lock()
+	r.running = false
+	r.mu.Unlock()
+
+	// No handler follows a trailing refresh, so publish it to clients here.
+	m.mu.RLock()
+	hub := m.wsHub
+	m.mu.RUnlock()
+	if hub != nil {
+		m.broadcastCurrentState(hub)
+	}
+}
+
+// stopAgentReportRefresh cancels a pending trailing refresh and waits for one
+// already running, so none touches stores Stop is about to close.
+func (m *Monitor) stopAgentReportRefresh() {
+	r := &m.agentReportRefresh
+	r.mu.Lock()
+	r.stopped = true
+	if r.pending != nil && r.pending.Stop() {
+		r.pending = nil
+		r.inFlight.Done()
+	}
+	r.mu.Unlock()
+	r.inFlight.Wait()
+}
+
 func recordSupplementalResourceChanges(store ResourceStoreInterface, changes []unifiedresources.ResourceChange) {
 	if store == nil || len(changes) == 0 {
 		return
@@ -7447,6 +7553,9 @@ const guestMetadataDrainTimeout = 2 * time.Second
 func (m *Monitor) Stop() {
 	log.Info().Msg("stopping monitor")
 
+	// A trailing agent-report refresh must not run against stores closed below.
+	m.stopAgentReportRefresh()
+
 	if m.deadMan != nil {
 		m.deadMan.stop(time.Now().UTC(), m.alertManager)
 	}

internal/monitoring/monitor_agents.go

@@ -1743,7 +1743,7 @@ func (m *Monitor) applyRejectedHostReportLiveness(
 	// clear an offline alert, or renew the accepted telemetry lease.
 	m.state.SetConnectionHealth(hostConnectionPrefix+host.ID, true)
 	m.persistHostContinuity(host, report, order)
-	m.refreshUnifiedResourceStoreAfterAgentStateChange()
+	m.refreshUnifiedResourceStoreAfterAgentReport()
 
 	log.Debug().
 		Str("hostID", host.ID).
@@ -2206,7 +2206,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con
 		if existing.ID != "" {
 			m.state.SetConnectionHealth(dockerConnectionPrefix+existing.ID, true)
 		}
-		m.refreshUnifiedResourceStoreAfterAgentStateChange()
+		m.refreshUnifiedResourceStoreAfterAgentReport()
 		log.Debug().
 			Str("dockerHostID", identifier).
 			Str("sequenceId", report.SequenceID).
@@ -2604,7 +2604,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con
 			Str("dockerHost", host.Hostname).
 			Str("sequenceId", report.SequenceID).
 			Msg("Docker helper status processed without refreshing inventory-derived alerts or metrics")
-		m.refreshUnifiedResourceStoreAfterAgentStateChange()
+		m.refreshUnifiedResourceStoreAfterAgentReport()
 		return host, nil
 	}
 
@@ -2700,7 +2700,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con
 		Int("containers", len(containers)).
 		Msg("Docker host report processed")
 
-	m.refreshUnifiedResourceStoreAfterAgentStateChange()
+	m.refreshUnifiedResourceStoreAfterAgentReport()
 
 	return host, nil
 }
@@ -3800,7 +3800,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 	// committed, because ownership is checked against that host ID.
 	m.ApplyProbeAvailabilityResults(host.ID, probeAvailabilityResultsFromReport(report.AvailabilityResults))
 	m.persistHostContinuity(host, report, reportOrder)
-	m.refreshUnifiedResourceStoreAfterAgentStateChange()
+	m.refreshUnifiedResourceStoreAfterAgentReport()
 
 	return host, nil
 }

internal/monitoring/monitor_host_agents_test.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"os"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -6412,11 +6413,11 @@ func TestApplyHostReportHonoursRemovalBlockWhenIdentityWouldFork(t *testing.T) {
 // countingResourceStore counts registry clones handed out by the live store.
 type countingResourceStore struct {
 	*unifiedresources.MonitorAdapter
-	getAll int
+	getAll atomic.Int32
 }
 
 func (c *countingResourceStore) GetAll() []unifiedresources.Resource {
-	c.getAll++
+	c.getAll.Add(1)
 	return c.MonitorAdapter.GetAll()
 }
 
@@ -6440,8 +6441,8 @@ func TestResourceSnapshotStoreClonesOncePerPass(t *testing.T) {
 	store := newResourceSnapshotStore(counting)
 	first := store.GetAll()
 	second := store.GetAll()
-	if counting.getAll != 1 {
-		t.Fatalf("underlying GetAll calls = %d, want 1", counting.getAll)
+	if got := counting.getAll.Load(); got != 1 {
+		t.Fatalf("underlying GetAll calls = %d, want 1", got)
 	}
 	if len(first) != 1 || len(second) != 1 || &first[0] != &second[0] {
 		t.Fatalf("pass consumers did not share one snapshot: %d and %d resources", len(first), len(second))
@@ -6480,8 +6481,98 @@ func TestAgentReportRefreshClonesRegistryOnce(t *testing.T) {
 	}
 
 	m.refreshUnifiedResourceStoreAfterAgentStateChange()
-	if counting.getAll != 1 {
-		t.Fatalf("registry clones for one agent report refresh = %d, want 1", counting.getAll)
+	if got := counting.getAll.Load(); got != 1 {
+		t.Fatalf("registry clones for one agent report refresh = %d, want 1", got)
+	}
+}
+
+// newAgentReportThrottleMonitor returns a monitor whose report-driven store
+// refreshes are throttled to window, with one online host agent.
+func newAgentReportThrottleMonitor(t *testing.T, window time.Duration) (*Monitor, *countingResourceStore) {
+	t.Helper()
+	state := models.NewState()
+	state.UpsertHost(models.Host{ID: "agent-1", Hostname: "pve1.example", Status: "online", CPUUsage: 1, LastSeen: time.Now().UTC()})
+	counting := &countingResourceStore{MonitorAdapter: unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil))}
+	m := &Monitor{
+		state:                    state,
+		resourceStore:            counting,
+		metricsHistory:           NewMetricsHistory(32, time.Hour),
+		agentReportRefreshWindow: window,
+	}
+	t.Cleanup(m.stopAgentReportRefresh)
+	return m, counting
+}
+
+func reportAgentCPU(m *Monitor, cpu float64) {
+	m.state.UpsertHost(models.Host{ID: "agent-1", Hostname: "pve1.example", Status: "online", CPUUsage: cpu, LastSeen: time.Now().UTC()})
+	m.refreshUnifiedResourceStoreAfterAgentReport()
+}
+
+func publishedAgentCPU(t *testing.T, store *countingResourceStore) float64 {
+	t.Helper()
+	for _, resource := range store.MonitorAdapter.GetAll() {
+		if resource.Type == unifiedresources.ResourceTypeAgent && resource.Metrics != nil && resource.Metrics.CPU != nil {
+			return resource.Metrics.CPU.Value
+		}
+	}
+	t.Fatal("no agent CPU in the canonical store")
+	return 0
+}
+
+// Accepted agent reports refresh the estate-wide store at most once per
+// window: the first report after a quiet window is published before the call
+// returns, and reports inside the window fold into one trailing refresh that
+// publishes the latest state (#2199).
+func TestAgentReportRefreshFoldsReportsWithinWindow(t *testing.T) {
+	const window = 500 * time.Millisecond
+	m, store := newAgentReportThrottleMonitor(t, window)
+
+	reportAgentCPU(m, 10)
+	if got := store.getAll.Load(); got != 1 {
+		t.Fatalf("refreshes after the leading report = %d, want 1", got)
+	}
+	if got := publishedAgentCPU(t, store); got != 10 {
+		t.Fatalf("leading report published CPU %v, want 10", got)
+	}
+
+	for cpu := 20.0; cpu <= 60; cpu += 10 {
+		reportAgentCPU(m, cpu)
+	}
+	if got := store.getAll.Load(); got != 1 {
+		t.Fatalf("refreshes during the window = %d, want the burst deferred", got)
+	}
+
+	deadline := time.Now().Add(5 * time.Second)
+	for store.getAll.Load() < 2 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if got := store.getAll.Load(); got != 2 {
+		t.Fatalf("refreshes after the window = %d, want one trailing refresh", got)
+	}
+	if got := publishedAgentCPU(t, store); got != 60 {
+		t.Fatalf("trailing refresh published CPU %v, want the latest report's 60", got)
+	}
+
+	time.Sleep(window + 100*time.Millisecond)
+	reportAgentCPU(m, 70)
+	if got := store.getAll.Load(); got != 3 {
+		t.Fatalf("refreshes after a quiet window = %d, want a synchronous leading refresh", got)
+	}
+	if got := publishedAgentCPU(t, store); got != 70 {
+		t.Fatalf("leading report after a quiet window published CPU %v, want 70", got)
+	}
+}
+
+func TestStopCancelsPendingAgentReportRefresh(t *testing.T) {
+	const window = 100 * time.Millisecond
+	m, store := newAgentReportThrottleMonitor(t, window)
+
+	reportAgentCPU(m, 10)
+	reportAgentCPU(m, 20)
+	m.stopAgentReportRefresh()
+	time.Sleep(3 * window)
+	if got := store.getAll.Load(); got != 1 {
+		t.Fatalf("refreshes after stop = %d, want the pending trailing refresh cancelled", got)
 	}
 }
 

tests/integration/tests/journeys/04-agent-install-registration.spec.ts

@@ -291,10 +291,25 @@ test.describe("Journey: Agent Install → Registration → Host Visible", () =>
       });
       expect(updateRes.ok()).toBeTruthy();
 
-      const stateRes2 = await apiRequest(page, "/api/state");
-      expect(stateRes2.ok()).toBeTruthy();
-      const state2 = (await stateRes2.json()) as Record<string, unknown>;
-      const host2 = findRegisteredAgentResource(state2);
+      // Reports within one refresh window are published together by a
+      // trailing store refresh (agent-lifecycle contract, #2199), so the
+      // update is visible within the window rather than at once.
+      let host2: StateResource | undefined;
+      await expect
+        .poll(
+          async () => {
+            const stateRes2 = await apiRequest(page, "/api/state");
+            expect(stateRes2.ok()).toBeTruthy();
+            const state2 = (await stateRes2.json()) as Record<string, unknown>;
+            host2 = findRegisteredAgentResource(state2);
+            return readNumber(asRecord(host2?.cpu)?.current) ?? 0;
+          },
+          {
+            message: "CPU usage should reflect updated report (75%)",
+            timeout: 10_000,
+          },
+        )
+        .toBeGreaterThanOrEqual(70);
       expect(host2, "Host should still exist after second report").toBeTruthy();
 
       const lastSeen2 = readNumber(host2!.lastSeen);