Scale connection stale cutoff by the adaptive planned poll interval

Adaptive polling deliberately stretches an instance's cadence toward its
max interval (5 minutes by default) while data is fresh, but the
connections aggregator judged staleness against the configured cadence
with a 2-minute floor. Any adaptive-enabled install therefore cycled
healthy PVE/PBS/PMG connections into stale for the back half of every
stretched poll gap: the Infrastructure page dropped the source badge
from API + Agent to Agent and connection-degraded alerts fired against
a schedule the poller was honoring.

The aggregator now scales the active-to-stale cutoff by the scheduler's
currently planned interval when that exceeds the configured cadence,
via Monitor.PlannedPollInterval and per-instance planned intervals in
the aggregator inputs. A plan tighter than the configured cadence never
tightens the cutoff, so genuine poll outages still trip the floor on
time. Connection-degraded alerts and the runtime inventory gate consume
the same derived state and inherit the fix.

Refs #1437

Contract-Neutral: behavioral fix: stale cutoff follows adaptive planned interval (#1437), no public contract delta

internal/api/connections_aggregator.go

@@ -139,6 +139,27 @@ type aggregatorInputs struct {
 	pvePollingInterval time.Duration
 	pbsPollingInterval time.Duration
 	pmgPollingInterval time.Duration
+
+	// plannedPollIntervals maps instanceHealth keys ("pve::<name>") to the
+	// adaptive scheduler's currently planned interval for that instance.
+	// Adaptive polling deliberately stretches cadence past the configured
+	// interval while data is fresh, so the active→stale cutoff must follow
+	// the schedule actually in force or a healthy connection reads as stale
+	// for the back half of every stretched cycle (#1437). Absent or zero
+	// entries fall back to the configured cadence.
+	plannedPollIntervals map[string]time.Duration
+}
+
+// effectivePollInterval picks the cadence the stale cutoff should scale by:
+// the adaptive scheduler's planned interval when it stretched beyond the
+// configured one, otherwise the configured interval. A planned interval that
+// shrank below the configured cadence never tightens the cutoff, so a genuine
+// poll outage still trips the connectionStaleThreshold floor on time.
+func effectivePollInterval(configured, planned time.Duration) time.Duration {
+	if planned > configured {
+		return planned
+	}
+	return configured
 }
 
 type connectionAgentDesiredConfig struct {
@@ -160,13 +181,16 @@ func buildConnections(in aggregatorInputs) []Connection {
 			len(in.vmwareInstances)+len(in.truenasInstances)+len(in.availabilityTargets)+len(in.hosts))
 
 	for _, pve := range in.pveInstances {
-		out = append(out, buildPVEConnection(pve, in.instanceHealth, now, in.pvePollingInterval))
+		interval := effectivePollInterval(in.pvePollingInterval, in.plannedPollIntervals["pve::"+pve.Name])
+		out = append(out, buildPVEConnection(pve, in.instanceHealth, now, interval))
 	}
 	for _, pbs := range in.pbsInstances {
-		out = append(out, buildPBSConnection(pbs, in.instanceHealth, now, in.pbsPollingInterval, in.pbsReportedNodeNames[pbs.Name]))
+		interval := effectivePollInterval(in.pbsPollingInterval, in.plannedPollIntervals["pbs::"+pbs.Name])
+		out = append(out, buildPBSConnection(pbs, in.instanceHealth, now, interval, in.pbsReportedNodeNames[pbs.Name]))
 	}
 	for _, pmg := range in.pmgInstances {
-		out = append(out, buildPMGConnection(pmg, in.instanceHealth, now, in.pmgPollingInterval))
+		interval := effectivePollInterval(in.pmgPollingInterval, in.plannedPollIntervals["pmg::"+pmg.Name])
+		out = append(out, buildPMGConnection(pmg, in.instanceHealth, now, interval))
 	}
 	for _, vmw := range in.vmwareInstances {
 		out = append(out, buildVMwareConnection(vmw, in.instanceHealth, in.vmwareSummaries, now))

internal/api/connections_aggregator_adaptive_stale_test.go

@@ -0,0 +1,58 @@
+package api
+
+import (
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
+)
+
+// Regression coverage for #1437: adaptive polling stretches an instance's
+// cadence past the configured interval while data is fresh, so the
+// active→stale cutoff has to follow the planned schedule. Before the fix a
+// healthy connection on a stretched (e.g. 5-minute) cadence read as stale for
+// the back half of every cycle and the platform page dropped to Agent-only.
+func TestBuildConnectionsAdaptiveIntervalScalesStaleCutoff(t *testing.T) {
+	now := time.Date(2026, 8, 11, 12, 0, 0, 0, time.UTC)
+	lastSuccess := now.Add(-4 * time.Minute)
+	base := aggregatorInputs{
+		pveInstances: []config.PVEInstance{{Name: "lab", Host: "https://lab:8006"}},
+		instanceHealth: map[string]monitoring.InstanceHealth{
+			"pve::lab": {PollStatus: monitoring.InstancePollStatus{LastSuccess: &lastSuccess}},
+		},
+		pvePollingInterval: 30 * time.Second,
+		now:                now,
+	}
+
+	pveState := func(in aggregatorInputs) ConnectionState {
+		t.Helper()
+		conns := buildConnections(in)
+		if len(conns) != 1 {
+			t.Fatalf("connections = %d, want 1", len(conns))
+		}
+		return conns[0].State
+	}
+
+	// No planned interval known: a 4-minute-old poll on a 30s configured
+	// cadence is past the floor and correctly reads stale.
+	if state := pveState(base); state != ConnectionStateStale {
+		t.Fatalf("state without plan = %q, want %q", state, ConnectionStateStale)
+	}
+
+	// The scheduler planned a 5-minute cadence, so 4 minutes old is on
+	// schedule and must stay active.
+	onSchedule := base
+	onSchedule.plannedPollIntervals = map[string]time.Duration{"pve::lab": 5 * time.Minute}
+	if state := pveState(onSchedule); state != ConnectionStateActive {
+		t.Fatalf("state with 5m plan = %q, want %q", state, ConnectionStateActive)
+	}
+
+	// A plan tighter than the configured cadence never tightens the cutoff:
+	// a genuine outage still trips against the configured interval and floor.
+	tightPlan := base
+	tightPlan.plannedPollIntervals = map[string]time.Duration{"pve::lab": 5 * time.Second}
+	if state := pveState(tightPlan); state != ConnectionStateStale {
+		t.Fatalf("state with 5s plan = %q, want %q", state, ConnectionStateStale)
+	}
+}

internal/api/connections_alerts.go

@@ -66,6 +66,7 @@ func buildAggregatorInputsWithRuntimeSources(
 		inputs.instanceHealth = instanceHealthByKey(monitor.SchedulerHealth())
 		inputs.availabilityStatuses = monitor.AvailabilityStatusSnapshot()
 		inputs.pbsReportedNodeNames = pbsReportedNodeNamesByInstance(monitor.PBSInstancesSnapshot())
+		inputs.plannedPollIntervals = plannedPollIntervalsForConfig(monitor, cfg)
 	} else {
 		inputs.hosts = []models.Host{}
 		inputs.instanceHealth = map[string]monitoring.InstanceHealth{}
@@ -143,3 +144,29 @@ func buildAlertConnectionSnapshotsWithRuntimeSources(
 	inputs := buildAggregatorInputsWithRuntimeSources(ctx, cfg, persistence, monitor, runtime)
 	return snapshotConnectionsForAlerts(buildConnections(inputs))
 }
+
+// plannedPollIntervalsForConfig collects the adaptive scheduler's currently
+// planned interval for every configured PVE/PBS/PMG instance, keyed the same
+// way as instanceHealth ("pve::<name>"). Instances the scheduler has no plan
+// for are omitted so the aggregator falls back to the configured cadence.
+func plannedPollIntervalsForConfig(monitor *monitoring.Monitor, cfg *config.Config) map[string]time.Duration {
+	if monitor == nil || cfg == nil {
+		return nil
+	}
+	out := make(map[string]time.Duration)
+	record := func(instanceType monitoring.InstanceType, keyPrefix, name string) {
+		if interval := monitor.PlannedPollInterval(instanceType, name); interval > 0 {
+			out[keyPrefix+name] = interval
+		}
+	}
+	for _, inst := range cfg.PVEInstances {
+		record(monitoring.InstanceTypePVE, "pve::", inst.Name)
+	}
+	for _, inst := range cfg.PBSInstances {
+		record(monitoring.InstanceTypePBS, "pbs::", inst.Name)
+	}
+	for _, inst := range cfg.PMGInstances {
+		record(monitoring.InstanceTypePMG, "pmg::", inst.Name)
+	}
+	return out
+}

internal/monitoring/monitor.go

@@ -2636,6 +2636,22 @@ func (m *Monitor) DeadLetterCount() int {
 	return m.deadLetterQueue.Size()
 }
 
+// PlannedPollInterval reports the adaptive scheduler's currently planned
+// interval for one instance, or zero when no plan exists (adaptive polling
+// disabled, unknown instance, or startup before the first plan). Consumers
+// use it to judge poll freshness against the schedule actually in force
+// rather than the configured cadence, which adaptive polling deliberately
+// stretches while data is fresh.
+func (m *Monitor) PlannedPollInterval(instanceType InstanceType, instanceName string) time.Duration {
+	if m == nil || m.scheduler == nil {
+		return 0
+	}
+	if task, ok := m.scheduler.LastScheduled(instanceType, instanceName); ok {
+		return task.Interval
+	}
+	return 0
+}
+
 func (m *Monitor) SchedulerHealth() SchedulerHealthResponse {
 	response := emptySchedulerHealthResponse(m.config != nil && m.config.AdaptivePollingEnabled)
 

internal/monitoring/monitor_planned_interval_test.go

@@ -0,0 +1,40 @@
+package monitoring
+
+import (
+	"testing"
+	"time"
+)
+
+// PlannedPollInterval feeds the connections aggregator the cadence the
+// adaptive scheduler actually promised, so the active→stale cutoff can track
+// stretched schedules (#1437).
+func TestPlannedPollIntervalReportsLastScheduled(t *testing.T) {
+	sched := NewAdaptiveScheduler(DefaultSchedulerConfig(), nil, nil, nil)
+	now := time.Now()
+	tasks := sched.BuildPlan(now, []InstanceDescriptor{{
+		Name:        "pve-a",
+		Type:        InstanceTypePVE,
+		LastSuccess: now,
+	}}, 0)
+	if len(tasks) != 1 {
+		t.Fatalf("planned tasks = %d, want 1", len(tasks))
+	}
+	if tasks[0].Interval <= 0 {
+		t.Fatalf("planned interval = %v, want > 0", tasks[0].Interval)
+	}
+
+	m := &Monitor{scheduler: sched}
+	if got := m.PlannedPollInterval(InstanceTypePVE, "pve-a"); got != tasks[0].Interval {
+		t.Fatalf("PlannedPollInterval = %v, want %v", got, tasks[0].Interval)
+	}
+	if got := m.PlannedPollInterval(InstanceTypePVE, "unknown"); got != 0 {
+		t.Fatalf("unknown instance interval = %v, want 0", got)
+	}
+	if got := (&Monitor{}).PlannedPollInterval(InstanceTypePVE, "pve-a"); got != 0 {
+		t.Fatalf("schedulerless monitor interval = %v, want 0", got)
+	}
+	var nilMonitor *Monitor
+	if got := nilMonitor.PlannedPollInterval(InstanceTypePVE, "pve-a"); got != 0 {
+		t.Fatalf("nil monitor interval = %v, want 0", got)
+	}
+}