Surface agent auth failures and staleness instead of a silent 401 loop

Refs #1515

When an upgraded or restored Pulse server no longer recognises an agent's
API token, the report endpoint returns 401. The agent buffered and retried
that report forever with only a generic warning, and the server kept the
node green at its last known agent version because a Proxmox node stays
online via the PVE API poll even after its agent dies.

- Agent: special-case 401 on /api/agents/agent/report. Drop the report
  instead of buffering it and log a throttled, actionable error pointing the
  operator at the install command to mint a fresh token.
- Installer: verify_agent_server_registration now tells a rejected token
  (401/403) apart from "agent has not reported yet" and prints the recovery
  steps at install time instead of a vague soft warning.
- Status layer: resourceFromHost flags a stale agent (its host marked
  offline by the staleness evaluator) and carries the agent's own last
  report time. Coalescing keeps a node online via the PVE source but the
  dead agent stays flagged, so its version is no longer presented as current.
  The Machines table renders such versions as "(stale)".

internal/hostagent/agent.go

@@ -109,12 +109,21 @@ type Agent struct {
 	collector              SystemCollector
 	newCommandClient       func(Config, string, string, string, string) *CommandClient
 	runCommandClient       func(*CommandClient, context.Context) error
+
+	// lastAuthFailureLog throttles the actionable 401 error so a permanently
+	// rejected token does not spam the log every report interval. Only touched
+	// from the single-threaded report loop (process/flushBuffer).
+	lastAuthFailureLog time.Time
 }
 
 const defaultInterval = 30 * time.Second
 const defaultStateDir = "/var/lib/pulse-agent"
 const agentReportEndpoint = "/api/agents/agent/report"
 
+// authFailureLogInterval bounds how often the agent re-emits the actionable
+// "token rejected" error while the server keeps returning 401.
+const authFailureLogInterval = 5 * time.Minute
+
 type reportHTTPStatusError struct {
 	Endpoint   string
 	Status     string
@@ -553,6 +562,16 @@ func (a *Agent) process(ctx context.Context) error {
 				Msg("Failed to send Unified Agent report (403 Forbidden). API token may lack 'Unified Agent reporting' scope. Set PULSE_ENABLE_HOST=false if host monitoring is not needed.")
 			return nil
 		}
+		if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusUnauthorized {
+			// 401 means the server does not recognise this agent's API token
+			// (for example after a Pulse restore or a v5 -> v6 upgrade that did
+			// not carry the token across). Retrying with the same token loops
+			// forever, so drop the report instead of buffering it and surface an
+			// actionable error the operator can act on. Throttle it so a
+			// permanently rejected token does not flood the log.
+			a.logAuthFailure(statusErr)
+			return nil
+		}
 
 		a.reportBuffer.Push(report)
 		event := a.logger.Warn().
@@ -567,6 +586,10 @@ func (a *Agent) process(ctx context.Context) error {
 		return nil
 	}
 
+	// A successful report means the token is accepted again; reset the auth
+	// failure throttle so a later rejection is reported promptly.
+	a.lastAuthFailureLog = time.Time{}
+
 	// If successful, try to flush buffer
 	a.flushBuffer(ctx)
 
@@ -577,6 +600,31 @@ func (a *Agent) process(ctx context.Context) error {
 	return nil
 }
 
+// logAuthFailure emits an actionable, throttled error when the server rejects
+// the agent's API token with 401. It is only called from the single-threaded
+// report loop, so lastAuthFailureLog needs no additional synchronisation.
+func (a *Agent) logAuthFailure(statusErr *reportHTTPStatusError) {
+	if time.Since(a.lastAuthFailureLog) < authFailureLogInterval {
+		return
+	}
+	a.lastAuthFailureLog = time.Now()
+
+	endpoint := agentReportEndpoint
+	statusCode := http.StatusUnauthorized
+	if statusErr != nil {
+		if statusErr.Endpoint != "" {
+			endpoint = statusErr.Endpoint
+		}
+		statusCode = statusErr.StatusCode
+	}
+
+	a.logger.Error().
+		Str("endpoint", endpoint).
+		Int("status_code", statusCode).
+		Str("pulse_url", a.trimmedPulseURL).
+		Msg("Pulse rejected this agent's API token (401 Unauthorized). The token is no longer valid on the server, which usually means Pulse was restored/reinstalled or upgraded (for example v5 -> v6) without carrying the token across. Re-run the agent install command from the Pulse UI (Settings > Agents) to mint a fresh token. Reports are dropped until the token is replaced.")
+}
+
 func (a *Agent) flushBuffer(ctx context.Context) {
 	if a.reportBuffer.IsEmpty() {
 		return
@@ -593,6 +641,18 @@ func (a *Agent) flushBuffer(ctx context.Context) {
 
 		if err := a.sendReport(ctx, report); err != nil {
 			var statusErr *reportHTTPStatusError
+			if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusUnauthorized {
+				// The token is rejected; buffered reports will never be
+				// accepted with it. Drop them so the buffer cannot grow
+				// unbounded across restarts, and surface the actionable error.
+				a.logAuthFailure(statusErr)
+				for {
+					if _, ok := a.reportBuffer.Pop(); !ok {
+						break
+					}
+				}
+				return
+			}
 			event := a.logger.Warn().
 				Err(err).
 				Str("endpoint", agentReportEndpoint).

internal/hostagent/send_report_test.go

@@ -12,6 +12,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
 	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
@@ -239,6 +240,107 @@ func TestAgentProcess_ForbiddenResponseDoesNotBuffer(t *testing.T) {
 	}
 }
 
+func TestAgentProcess_UnauthorizedResponseDoesNotBuffer(t *testing.T) {
+	t.Parallel()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer server.Close()
+
+	var logBuf bytes.Buffer
+	agent := &Agent{
+		cfg:             Config{APIToken: "stale-token"},
+		logger:          zerolog.New(&logBuf),
+		httpClient:      server.Client(),
+		trimmedPulseURL: server.URL,
+		reportBuffer:    utils.New[agentshost.Report](8),
+		collector:       &mockCollector{},
+	}
+
+	if err := agent.process(context.Background()); err != nil {
+		t.Fatalf("process: %v", err)
+	}
+	// A rejected token loops forever; the report must be dropped, not buffered.
+	if !agent.reportBuffer.IsEmpty() {
+		t.Fatalf("buffer should stay empty for unauthorized response, len=%d", agent.reportBuffer.Len())
+	}
+	if !strings.Contains(logBuf.String(), "rejected this agent's API token") {
+		t.Fatalf("expected 401 log to be actionable, got %q", logBuf.String())
+	}
+}
+
+func TestAgentProcess_UnauthorizedLogThrottled(t *testing.T) {
+	t.Parallel()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer server.Close()
+
+	var logBuf bytes.Buffer
+	agent := &Agent{
+		cfg:             Config{APIToken: "stale-token"},
+		logger:          zerolog.New(&logBuf),
+		httpClient:      server.Client(),
+		trimmedPulseURL: server.URL,
+		reportBuffer:    utils.New[agentshost.Report](8),
+		collector:       &mockCollector{},
+	}
+
+	for i := 0; i < 3; i++ {
+		if err := agent.process(context.Background()); err != nil {
+			t.Fatalf("process %d: %v", i, err)
+		}
+	}
+	if got := strings.Count(logBuf.String(), "rejected this agent's API token"); got != 1 {
+		t.Fatalf("expected actionable 401 log exactly once (throttled), got %d in %q", got, logBuf.String())
+	}
+
+	// Resetting the throttle (which a successful report does) re-emits the error.
+	agent.lastAuthFailureLog = time.Time{}
+	if err := agent.process(context.Background()); err != nil {
+		t.Fatalf("process after reset: %v", err)
+	}
+	if got := strings.Count(logBuf.String(), "rejected this agent's API token"); got != 2 {
+		t.Fatalf("expected actionable 401 log to re-emit after throttle reset, got %d", got)
+	}
+}
+
+func TestAgentFlushBuffer_UnauthorizedDropsBuffer(t *testing.T) {
+	t.Parallel()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer server.Close()
+
+	var logBuf bytes.Buffer
+	buffer := utils.New[agentshost.Report](8)
+	buffer.Push(agentshost.Report{Agent: agentshost.AgentInfo{ID: "a1"}})
+	buffer.Push(agentshost.Report{Agent: agentshost.AgentInfo{ID: "a2"}})
+
+	agent := &Agent{
+		cfg:             Config{APIToken: "stale-token"},
+		logger:          zerolog.New(&logBuf),
+		httpClient:      server.Client(),
+		trimmedPulseURL: server.URL,
+		reportBuffer:    buffer,
+		collector:       &mockCollector{},
+	}
+
+	agent.flushBuffer(context.Background())
+
+	// Buffered reports will never be accepted with a rejected token; drop them
+	// so the buffer cannot grow unbounded across restarts.
+	if !agent.reportBuffer.IsEmpty() {
+		t.Fatalf("buffer should be drained after unauthorized flush, len=%d", agent.reportBuffer.Len())
+	}
+	if !strings.Contains(logBuf.String(), "rejected this agent's API token") {
+		t.Fatalf("expected flush 401 log to be actionable, got %q", logBuf.String())
+	}
+}
+
 func TestAgentProcess_SuccessLogsUnifiedAgentReport(t *testing.T) {
 	t.Parallel()
 

internal/unifiedresources/adapters.go

@@ -165,6 +165,20 @@ func resourceFromHost(host models.Host) (Resource, ResourceIdentity) {
 		LinkedVMID:        host.LinkedVMID,
 		LinkedContainerID: host.LinkedContainerID,
 	}
+
+	// Surface agent staleness so a row that stays online via another source
+	// (e.g. a Proxmox node still reachable over the PVE API) does not present a
+	// dead agent's version as if it were current. The staleness evaluator marks
+	// a host "offline" once it stops reporting; carry that onto the agent
+	// payload along with the agent's own last report time.
+	if !host.LastSeen.IsZero() {
+		lastReport := host.LastSeen
+		agent.LastReportAt = &lastReport
+	}
+	if strings.EqualFold(strings.TrimSpace(host.Status), "offline") {
+		agent.Stale = true
+	}
+
 	storageAssessments := make([]storagehealth.Assessment, 0, len(host.RAID)+1)
 
 	// Populate sensors

internal/unifiedresources/adapters_test.go

@@ -579,6 +579,54 @@ func TestResourceFromHostProjectsAgentHostProfile(t *testing.T) {
 	}
 }
 
+func TestResourceFromHostMarksOfflineAgentStale(t *testing.T) {
+	lastReport := time.Now().Add(-30 * time.Minute).UTC()
+	host := models.Host{
+		ID:           "omv-host",
+		Hostname:     "omv",
+		Platform:     "linux",
+		Status:       "offline", // set by the staleness evaluator when reports stop
+		AgentVersion: "6.0.2",
+		LastSeen:     lastReport,
+	}
+
+	resource, _ := resourceFromHost(host)
+	if resource.Agent == nil {
+		t.Fatal("expected agent payload")
+	}
+	if !resource.Agent.Stale {
+		t.Fatalf("expected offline agent to be marked stale, got %+v", resource.Agent)
+	}
+	if resource.Agent.AgentVersion != "6.0.2" {
+		t.Fatalf("agent version = %q, want the last reported 6.0.2 retained", resource.Agent.AgentVersion)
+	}
+	if resource.Agent.LastReportAt == nil || !resource.Agent.LastReportAt.Equal(lastReport) {
+		t.Fatalf("expected agent LastReportAt = %v, got %v", lastReport, resource.Agent.LastReportAt)
+	}
+}
+
+func TestResourceFromHostOnlineAgentNotStale(t *testing.T) {
+	host := models.Host{
+		ID:           "live-host",
+		Hostname:     "live",
+		Platform:     "linux",
+		Status:       "online",
+		AgentVersion: "6.0.5",
+		LastSeen:     time.Now().UTC(),
+	}
+
+	resource, _ := resourceFromHost(host)
+	if resource.Agent == nil {
+		t.Fatal("expected agent payload")
+	}
+	if resource.Agent.Stale {
+		t.Fatalf("online agent should not be marked stale, got %+v", resource.Agent)
+	}
+	if resource.Agent.LastReportAt == nil {
+		t.Fatalf("expected LastReportAt to be populated for a reporting agent")
+	}
+}
+
 func TestResourceFromHostProjectsPressureOnlyThermalState(t *testing.T) {
 	warningLevel := 1
 	host := models.Host{

internal/unifiedresources/presentation_coalesce_test.go

@@ -58,6 +58,55 @@ func TestCoalescePresentationHostResourcesMergesSplitRuntimeAndPlatformHost(t *t
 	}
 }
 
+func TestCoalescePresentationHostResourcesSurfacesStaleAgentOnLiveNode(t *testing.T) {
+	// A Proxmox node whose Pulse Agent has stopped reporting (401) stays online
+	// via the PVE API poll, but the dead agent must remain flagged stale so the
+	// UI does not present its last version as if it were current (issue #1515).
+	now := time.Date(2026, 7, 8, 10, 30, 0, 0, time.UTC)
+	resources := []Resource{
+		{
+			ID:       "agent-runtime-pve1",
+			Type:     ResourceTypeAgent,
+			Name:     "pve1",
+			Status:   StatusOffline,
+			LastSeen: now.Add(-30 * time.Minute),
+			Sources:  []DataSource{SourceAgent},
+			Identity: ResourceIdentity{Hostnames: []string{"pve1"}},
+			Agent: &AgentData{
+				AgentID:      "agent-machine-pve1",
+				Hostname:     "pve1",
+				AgentVersion: "6.0.2",
+				Stale:        true,
+			},
+		},
+		{
+			ID:       "proxmox-node-pve1",
+			Type:     ResourceTypeAgent,
+			Name:     "pve1",
+			Status:   StatusOnline,
+			LastSeen: now,
+			Sources:  []DataSource{SourceProxmox},
+			Identity: ResourceIdentity{Hostnames: []string{"pve1"}},
+			Proxmox:  &ProxmoxData{NodeName: "pve1", ClusterName: "homelab"},
+		},
+	}
+
+	coalesced := CoalescePresentationHostResources(resources)
+	if len(coalesced) != 1 {
+		t.Fatalf("expected split host resources to coalesce into 1 resource, got %d", len(coalesced))
+	}
+	resource := coalesced[0]
+	if resource.Status != StatusOnline {
+		t.Fatalf("node should stay online via the PVE source, got %q", resource.Status)
+	}
+	if resource.Agent == nil || !resource.Agent.Stale {
+		t.Fatalf("expected the dead agent to remain flagged stale after merge, got %+v", resource.Agent)
+	}
+	if resource.Agent.AgentVersion != "6.0.2" {
+		t.Fatalf("expected stale agent version retained, got %q", resource.Agent.AgentVersion)
+	}
+}
+
 func TestCoalescePresentationHostResourcesRedirectsProxmoxChildrenToAgentBackedParent(t *testing.T) {
 	now := time.Date(2026, 5, 22, 10, 30, 0, 0, time.UTC)
 	proxmoxParentID := "agent-proxmox-delly"

internal/unifiedresources/types.go

@@ -774,8 +774,19 @@ type AgentMemoryMeta struct {
 
 // AgentData contains host agent-specific data.
 type AgentData struct {
-	AgentID               string             `json:"agentId,omitempty"`
-	AgentVersion          string             `json:"agentVersion,omitempty"`
+	AgentID      string `json:"agentId,omitempty"`
+	AgentVersion string `json:"agentVersion,omitempty"`
+	// Stale is set when the agent has stopped reporting (its host was marked
+	// offline by the staleness evaluator) even though the row itself may stay
+	// online via another source such as the Proxmox API poll. It lets the UI
+	// present the agent and its version as not-reporting instead of a
+	// healthy-looking stale value.
+	Stale bool `json:"stale,omitempty"`
+	// LastReportAt is the agent's own last successful report time. On a
+	// multi-source row (for example a Proxmox node also polled over the PVE
+	// API) this differs from the row's LastSeen, which reflects the freshest
+	// source rather than the agent.
+	LastReportAt          *time.Time         `json:"lastReportAt,omitempty"`
 	Hostname              string             `json:"hostname,omitempty"`
 	MachineID             string             `json:"machineId,omitempty"`
 	TokenID               string             `json:"tokenId,omitempty"`