Run assigned availability checks from the host agent

The unified agent gains an availability module: probe assignments
arrive through the signed remote-config channel (missing key clears
the schedule), each enabled target runs on its own clamped interval
through the shared probe core, and results queue in a bounded
drop-oldest buffer. A result is offered to the primary server until
one delivery succeeds and never again after - buffered offline
reports are stripped of availability results so the disk buffer
cannot replay observations the queue still holds. ApplyHostReport
feeds accepted reports into the probe ingestion path, where the
ownership check and failure accounting live, and the probe agent id
is projected onto unified availability resources for source
attribution in the UI.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

cmd/pulse-agent/main.go

@@ -28,6 +28,7 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentupdate"
+	pulseconfig "github.com/rcourtman/pulse-go-rewrite/internal/config"
 	"github.com/rcourtman/pulse-go-rewrite/internal/dockeragent"
 	"github.com/rcourtman/pulse-go-rewrite/internal/hostagent"
 	"github.com/rcourtman/pulse-go-rewrite/internal/kubernetesagent"
@@ -387,32 +388,33 @@ func run(ctx context.Context, args []string, getenv func(string) string) error {
 	// 8. Start Host Agent (if enabled)
 	if cfg.EnableHost {
 		hostCfg := hostagent.Config{
-			PulseURL:           cfg.PulseURL,
-			APIToken:           cfg.APIToken,
-			Interval:           cfg.Interval,
-			HostnameOverride:   cfg.HostnameOverride,
-			AgentID:            cfg.AgentID,
-			AgentType:          "unified",
-			AgentVersion:       Version,
-			Tags:               cfg.Tags,
-			InsecureSkipVerify: cfg.InsecureSkipVerify,
-			CACertPath:         cfg.CACertPath,
-			ServerFingerprint:  cfg.ServerFingerprint,
-			DeploySSHUser:      cfg.DeploySSHUser,
-			LogLevel:           cfg.LogLevel,
-			Logger:             &logger,
-			EnableProxmox:      cfg.EnableProxmox,
-			ProxmoxType:        cfg.ProxmoxType,
-			EnableCommands:     cfg.EnableCommands,
-			Enroll:             cfg.Enroll,
-			DiskExclude:        cfg.DiskExclude,
-			StateDir:           cfg.StateDir,
-			ReportIP:           cfg.ReportIP,
-			DisableCeph:        cfg.DisableCeph,
-			AppliedConfig:      cfg.AppliedConfig,
-			UpdateStatus:       updater.Snapshot,
-			ModuleStatus:       runtimeStatus.moduleStatuses,
-			Observers:          hostObserverTargets(cfg.Observers),
+			PulseURL:            cfg.PulseURL,
+			APIToken:            cfg.APIToken,
+			Interval:            cfg.Interval,
+			HostnameOverride:    cfg.HostnameOverride,
+			AgentID:             cfg.AgentID,
+			AgentType:           "unified",
+			AgentVersion:        Version,
+			Tags:                cfg.Tags,
+			InsecureSkipVerify:  cfg.InsecureSkipVerify,
+			CACertPath:          cfg.CACertPath,
+			ServerFingerprint:   cfg.ServerFingerprint,
+			DeploySSHUser:       cfg.DeploySSHUser,
+			LogLevel:            cfg.LogLevel,
+			Logger:              &logger,
+			EnableProxmox:       cfg.EnableProxmox,
+			ProxmoxType:         cfg.ProxmoxType,
+			EnableCommands:      cfg.EnableCommands,
+			Enroll:              cfg.Enroll,
+			DiskExclude:         cfg.DiskExclude,
+			StateDir:            cfg.StateDir,
+			ReportIP:            cfg.ReportIP,
+			DisableCeph:         cfg.DisableCeph,
+			AvailabilityTargets: cfg.AvailabilityTargets,
+			AppliedConfig:       cfg.AppliedConfig,
+			UpdateStatus:        updater.Snapshot,
+			ModuleStatus:        runtimeStatus.moduleStatuses,
+			Observers:           hostObserverTargets(cfg.Observers),
 
 			DockerContainerUpdater: dockerUpdaterBridge,
 		}
@@ -848,6 +850,10 @@ type Config struct {
 	DisableCeph bool   // Disable local Ceph status polling
 	SelfTest    bool   // Perform self-test and exit
 
+	// AvailabilityTargets are externally probed availability checks assigned to
+	// this agent by the server. Remote config is the only source.
+	AvailabilityTargets []pulseconfig.AvailabilityTarget
+
 	// Health/metrics server
 	HealthAddr    string
 	AppliedConfig *agentshost.ConfigFingerprint
@@ -1530,6 +1536,7 @@ func retryLogEvent(logger *zerolog.Logger, attempt int) *zerolog.Event {
 // - interval (string/duration)
 // - report_ip (string)
 // - disable_ceph (bool)
+// - availabilityTargets (array of assigned availability checks)
 func applyRemoteSettings(cfg *Config, settings map[string]interface{}, logger *zerolog.Logger) {
 	for k, v := range settings {
 		switch k {
@@ -1633,6 +1640,14 @@ func applyRemoteSettings(cfg *Config, settings map[string]interface{}, logger *z
 				cfg.DisableCeph = b
 				logger.Info().Bool("val", b).Msg("Remote config: disable_ceph")
 			}
+		case "availabilityTargets":
+			targets, err := hostagent.AvailabilityTargetsFromSetting(v)
+			if err != nil {
+				logger.Warn().Err(err).Msg("Remote config: ignoring unreadable availabilityTargets value")
+				continue
+			}
+			cfg.AvailabilityTargets = targets
+			logger.Info().Int("count", len(targets)).Msg("Remote config: availabilityTargets")
 		}
 	}
 }

cmd/pulse-agent/main_test.go

@@ -2354,3 +2354,35 @@ func TestAllowPlaintextHTTPFlagParsesAndDefaultsClosed(t *testing.T) {
 		t.Fatal("PULSE_AGENT_ALLOW_PLAINTEXT_HTTP env was not applied")
 	}
 }
+
+func TestApplyRemoteSettingsCarriesAvailabilityAssignmentsToStartup(t *testing.T) {
+	logger := zerolog.New(io.Discard)
+	cfg := &Config{}
+
+	applyRemoteSettings(cfg, map[string]interface{}{
+		"availabilityTargets": []interface{}{
+			map[string]interface{}{
+				"id":                  "remote-a",
+				"address":             "a.local",
+				"protocol":            "icmp",
+				"enabled":             true,
+				"pollIntervalSeconds": float64(30),
+			},
+		},
+	}, &logger)
+
+	if len(cfg.AvailabilityTargets) != 1 {
+		t.Fatalf("availability targets = %+v, want the assignment applied at boot", cfg.AvailabilityTargets)
+	}
+	if cfg.AvailabilityTargets[0].ID != "remote-a" {
+		t.Fatalf("availability target = %+v", cfg.AvailabilityTargets[0])
+	}
+
+	applyRemoteSettings(cfg, map[string]interface{}{
+		"availabilityTargets": "not-a-list",
+	}, &logger)
+
+	if len(cfg.AvailabilityTargets) != 1 {
+		t.Fatalf("availability targets = %+v, want an unreadable payload ignored", cfg.AvailabilityTargets)
+	}
+}

cmd/pulse-agent/service_windows.go

@@ -85,24 +85,25 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha
 	// Start Host Agent (if enabled)
 	if ws.cfg.EnableHost {
 		hostCfg := hostagent.Config{
-			PulseURL:           ws.cfg.PulseURL,
-			APIToken:           ws.cfg.APIToken,
-			Interval:           ws.cfg.Interval,
-			HostnameOverride:   ws.cfg.HostnameOverride,
-			AgentID:            ws.cfg.AgentID,
-			AgentType:          "unified",
-			AgentVersion:       Version,
-			Tags:               ws.cfg.Tags,
-			InsecureSkipVerify: ws.cfg.InsecureSkipVerify,
-			CACertPath:         ws.cfg.CACertPath,
-			ServerFingerprint:  ws.cfg.ServerFingerprint,
-			DeploySSHUser:      ws.cfg.DeploySSHUser,
-			LogLevel:           ws.cfg.LogLevel,
-			Logger:             &ws.logger,
-			AppliedConfig:      ws.cfg.AppliedConfig,
-			UpdateStatus:       updater.Snapshot,
-			ModuleStatus:       runtimeStatus.moduleStatuses,
-			Observers:          hostObserverTargets(ws.cfg.Observers),
+			PulseURL:            ws.cfg.PulseURL,
+			APIToken:            ws.cfg.APIToken,
+			Interval:            ws.cfg.Interval,
+			HostnameOverride:    ws.cfg.HostnameOverride,
+			AgentID:             ws.cfg.AgentID,
+			AgentType:           "unified",
+			AgentVersion:        Version,
+			Tags:                ws.cfg.Tags,
+			InsecureSkipVerify:  ws.cfg.InsecureSkipVerify,
+			CACertPath:          ws.cfg.CACertPath,
+			ServerFingerprint:   ws.cfg.ServerFingerprint,
+			DeploySSHUser:       ws.cfg.DeploySSHUser,
+			LogLevel:            ws.cfg.LogLevel,
+			Logger:              &ws.logger,
+			AppliedConfig:       ws.cfg.AppliedConfig,
+			AvailabilityTargets: ws.cfg.AvailabilityTargets,
+			UpdateStatus:        updater.Snapshot,
+			ModuleStatus:        runtimeStatus.moduleStatuses,
+			Observers:           hostObserverTargets(ws.cfg.Observers),
 		}
 		agent, err := hostagent.New(hostCfg)
 		if err != nil {

internal/hostagent/agent.go

@@ -26,6 +26,7 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agenttls"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentupdate"
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
 	"github.com/rcourtman/pulse-go-rewrite/internal/platformsupport"
 	"github.com/rcourtman/pulse-go-rewrite/internal/remoteconfig"
 	"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
@@ -78,6 +79,11 @@ type Config struct {
 	ReportIP    string // IP address to report instead of auto-detected (for multi-NIC systems)
 	DisableCeph bool   // If true, disables local Ceph status polling
 
+	// AvailabilityTargets are the externally probed availability checks the
+	// server assigned to this agent at startup. Later assignments arrive
+	// through ApplyRemoteConfig.
+	AvailabilityTargets []config.AvailabilityTarget
+
 	// AppliedConfig is the non-secret fingerprint of the managed config that
 	// was applied before this runtime started. UpdateStatus supplies live
 	// self-update state without coupling report collection to updater internals.
@@ -156,6 +162,7 @@ type Agent struct {
 	runCommandClient       func(*CommandClient, context.Context) error
 	packageUpdates         *packageUpdateManager
 	storageCleanup         *storageCleanupManager
+	availability           *availabilityProbeModule
 	reportStreamID         string
 	reportSequence         atomic.Uint64
 
@@ -387,6 +394,7 @@ func New(cfg Config) (*Agent, error) {
 		runCommandClient:    runCommandClientFn,
 		packageUpdates:      packageUpdates,
 		storageCleanup:      storageCleanup,
+		availability:        newAvailabilityProbeModule(logger, cfg.AvailabilityTargets),
 		reportStreamID:      reportStreamID,
 	}
 
@@ -541,6 +549,10 @@ func (a *Agent) Run(ctx context.Context) error {
 		a.startCommandClient(commandClient)
 	}
 
+	// Externally probed availability checks run on their own schedule, not on
+	// the report interval, and queue their results for the next report.
+	go a.availability.Run(ctx)
+
 	// Load any reports buffered from a previous shutdown
 	a.loadPersistedBuffer()
 	a.loadPersistedObserverBuffers()
@@ -643,10 +655,16 @@ func (a *Agent) currentUpdateStatus() *agentshost.UpdateStatus {
 }
 
 func (a *Agent) currentModuleStatus() []agentshost.ModuleStatus {
-	if a.cfg.ModuleStatus == nil {
-		return nil
+	var statuses []agentshost.ModuleStatus
+	if a.cfg.ModuleStatus != nil {
+		statuses = append(statuses, a.cfg.ModuleStatus()...)
 	}
-	return append([]agentshost.ModuleStatus(nil), a.cfg.ModuleStatus()...)
+	// The probe module is owned by the host agent rather than the runtime
+	// supervisor, so it reports itself and only while it has work assigned.
+	if status, ok := a.availability.moduleStatus(); ok {
+		statuses = append(statuses, status)
+	}
+	return statuses
 }
 
 func (a *Agent) signalRemoteConfigChanged() {
@@ -736,13 +754,14 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo
 	if !a.reportBuffer.IsEmpty() {
 		a.flushBuffer(ctx)
 		if !a.reportBuffer.IsEmpty() {
-			a.reportBuffer.Push(report)
+			a.bufferPrimaryReport(report)
 			return nil
 		}
 	}
 
 	if err := a.sendReport(ctx, report); err != nil {
 		agenttarget.MarkDelivery("host", "primary", "primary", false)
+		a.availability.discardInFlight()
 		var statusErr *reportHTTPStatusError
 		if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusForbidden {
 			a.logger.Error().
@@ -763,7 +782,7 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo
 			return nil
 		}
 
-		a.reportBuffer.Push(report)
+		a.bufferPrimaryReport(report)
 		event := a.logger.Warn().
 			Err(err).
 			Str("endpoint", agentReportEndpoint).
@@ -777,6 +796,10 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo
 	}
 	agenttarget.MarkDelivery("host", "primary", "primary", true)
 
+	// The server has counted every availability result in this report, so they
+	// must never be offered to it again.
+	a.availability.commitDelivered()
+
 	// 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{}
@@ -791,6 +814,17 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo
 	return nil
 }
 
+// bufferPrimaryReport queues a report the primary destination has not accepted
+// yet. Availability results are stripped from the buffered copy and left in the
+// probe module's queue, so the retry carries them exactly once: a buffered copy
+// that also held them could be delivered alongside a fresh report and make the
+// server count the same observation twice.
+func (a *Agent) bufferPrimaryReport(report agentshost.Report) {
+	a.availability.discardInFlight()
+	report.AvailabilityResults = nil
+	a.reportBuffer.Push(report)
+}
+
 func (a *Agent) deliverObserverReport(ctx context.Context, observer *observerReporter, report agentshost.Report) {
 	if !observer.reportBuffer.IsEmpty() {
 		a.flushObserverBuffer(ctx, observer)
@@ -1135,9 +1169,12 @@ func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) {
 		Unraid:         unraidData,
 		Ceph:           cephData,
 		ClusterSensors: clusterSensors,
-		Tags:           append([]string(nil), runtimeConfig.tags...),
-		Timestamp:      a.collector.Now(),
-		SequenceID:     a.nextReportSequenceID(),
+		// Results stay queued until the primary destination accepts this
+		// report, so a delivery failure retries them instead of losing them.
+		AvailabilityResults: a.availability.snapshotForReport(),
+		Tags:                append([]string(nil), runtimeConfig.tags...),
+		Timestamp:           a.collector.Now(),
+		SequenceID:          a.nextReportSequenceID(),
 	}
 
 	return report, nil
@@ -1322,6 +1359,7 @@ func (a *Agent) ApplyRemoteConfig(settings map[string]interface{}, commandsEnabl
 		a.configMu.Unlock()
 		a.logger.Info().Bool("disable_ceph", disableCeph).Msg("Applied remote Ceph collection setting")
 	}
+	a.applyRemoteAvailabilityTargets(settings)
 	if remoteConfigAppliedWithoutRestart(settings) && remoteconfig.HasAppliedDesiredConfig(commandsEnabled, settings) {
 		metadata, err := remoteconfig.BuildDesiredConfigMetadata(commandsEnabled, settings)
 		if err != nil {
@@ -1334,10 +1372,27 @@ func (a *Agent) ApplyRemoteConfig(settings map[string]interface{}, commandsEnabl
 	}
 }
 
+// applyRemoteAvailabilityTargets reconciles the externally probed availability
+// checks assigned to this agent. A payload the agent cannot decode leaves the
+// current assignment in place: results for targets it no longer owns are
+// rejected server-side, so keeping the schedule is safer than blindly
+// abandoning coverage over a malformed field.
+func (a *Agent) applyRemoteAvailabilityTargets(settings map[string]interface{}) {
+	targets, err := availabilityTargetsFromSettings(settings)
+	if err != nil {
+		a.logger.Warn().Err(err).Msg("Ignoring unreadable availability probe assignments")
+		return
+	}
+	if !a.availability.applyTargets(targets) {
+		return
+	}
+	a.logger.Info().Int("targets", len(targets)).Msg("Applied remote availability probe assignments")
+}
+
 func remoteConfigAppliedWithoutRestart(settings map[string]interface{}) bool {
 	for key := range settings {
 		switch key {
-		case "interval", "report_ip", "disable_ceph":
+		case "interval", "report_ip", "disable_ceph", availabilitySettingsKey:
 			continue
 		default:
 			return false

internal/hostagent/availability.go

@@ -0,0 +1,392 @@
+package hostagent
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe"
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
+	"github.com/rs/zerolog"
+)
+
+// availabilitySettingsKey is the remote-config key carrying the availability
+// targets the server assigned to this agent.
+const availabilitySettingsKey = "availabilityTargets"
+
+// availabilityModuleName is how an active probe assignment is surfaced in
+// AgentInfo.Modules, alongside host/docker/kubernetes.
+const availabilityModuleName = "availability"
+
+const (
+	availabilityModuleStateStarting = "starting"
+	availabilityModuleStateRunning  = "running"
+)
+
+// availabilityPendingCapacity bounds the results waiting for a report. It
+// mirrors the report buffer's spirit: an agent that cannot reach Pulse keeps
+// the newest observations and drops the oldest rather than growing without
+// limit.
+const availabilityPendingCapacity = 200
+
+// availabilityMinInterval and availabilityMaxInterval mirror the server-side
+// clamp so an assignment cannot make the agent probe faster (or rarer) than
+// the poller would have locally.
+const (
+	availabilityMinInterval = 10 * time.Second
+	availabilityMaxInterval = time.Hour
+)
+
+// availabilityErrorLimit bounds the failure text a single check contributes to
+// a report. The server only shows the latest error, so an unbounded message
+// buys nothing and inflates every report.
+const availabilityErrorLimit = 240
+
+// pendingAvailabilityResult tags each queued result with a monotonic sequence.
+// Delivery is confirmed by sequence rather than by count so a queue overflow
+// while a report is in flight cannot drop an observation that was never sent.
+type pendingAvailabilityResult struct {
+	sequence uint64
+	result   agentshost.AvailabilityProbeResult
+}
+
+// availabilityProbeModule runs the availability checks the server assigned to
+// this agent and queues their results for the next host report. It owns its
+// scheduling entirely: the server sends assignments, never a schedule.
+type availabilityProbeModule struct {
+	logger zerolog.Logger
+	now    func() time.Time
+	probe  func(context.Context, config.AvailabilityTarget) (availabilityprobe.Outcome, error)
+
+	mu        sync.Mutex
+	targets   []config.AvailabilityTarget
+	updatedAt time.Time
+	running   bool
+	sequence  uint64
+	inFlight  uint64
+
+	pending *utils.Queue[pendingAvailabilityResult]
+	reload  chan struct{}
+}
+
+func newAvailabilityProbeModule(logger zerolog.Logger, targets []config.AvailabilityTarget) *availabilityProbeModule {
+	module := &availabilityProbeModule{
+		logger:  logger.With().Str("module", availabilityModuleName).Logger(),
+		now:     time.Now,
+		probe:   availabilityprobe.Result,
+		pending: utils.New[pendingAvailabilityResult](availabilityPendingCapacity),
+		reload:  make(chan struct{}, 1),
+	}
+	module.applyTargets(targets)
+	return module
+}
+
+// applyTargets replaces the assignment set. It reports whether the schedule
+// actually changed: the remote config is re-fetched on a fixed interval and an
+// unchanged assignment must not restart (and therefore re-run) every check.
+func (m *availabilityProbeModule) applyTargets(targets []config.AvailabilityTarget) bool {
+	if m == nil {
+		return false
+	}
+	normalized := normalizeAvailabilityAssignments(targets)
+
+	m.mu.Lock()
+	if availabilityAssignmentsEqual(m.targets, normalized) {
+		m.mu.Unlock()
+		return false
+	}
+	m.targets = normalized
+	m.updatedAt = m.now().UTC()
+	m.mu.Unlock()
+
+	select {
+	case m.reload <- struct{}{}:
+	default:
+	}
+	return true
+}
+
+func normalizeAvailabilityAssignments(targets []config.AvailabilityTarget) []config.AvailabilityTarget {
+	normalized := make([]config.AvailabilityTarget, 0, len(targets))
+	for _, target := range targets {
+		// The ID is checked before normalization: defaults would mint a fresh
+		// one, and a target the server cannot recognise is unreportable.
+		if strings.TrimSpace(target.ID) == "" {
+			continue
+		}
+		normalized = append(normalized, config.NormalizeAvailabilityTarget(target))
+	}
+	sort.Slice(normalized, func(i, j int) bool { return normalized[i].ID < normalized[j].ID })
+	return normalized
+}
+
+func availabilityAssignmentsEqual(left, right []config.AvailabilityTarget) bool {
+	if len(left) != len(right) {
+		return false
+	}
+	for i := range left {
+		if left[i] != right[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func (m *availabilityProbeModule) assignments() []config.AvailabilityTarget {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return append([]config.AvailabilityTarget(nil), m.targets...)
+}
+
+// Run supervises the probe schedule until the context is cancelled. Each
+// assignment change stops the current workers and rebuilds them, which keeps a
+// removed target from ever running one more time.
+func (m *availabilityProbeModule) Run(ctx context.Context) {
+	if m == nil {
+		return
+	}
+	m.mu.Lock()
+	m.running = true
+	m.mu.Unlock()
+	defer func() {
+		m.mu.Lock()
+		m.running = false
+		m.mu.Unlock()
+	}()
+
+	for {
+		// Consume a reload queued before this pass: the assignment about to be
+		// read already contains it, and acting on it again would restart every
+		// worker and re-run each check immediately.
+		select {
+		case <-m.reload:
+		default:
+		}
+
+		runCtx, cancel := context.WithCancel(ctx)
+		var wg sync.WaitGroup
+		for _, target := range m.assignments() {
+			if !target.Enabled {
+				continue
+			}
+			wg.Add(1)
+			go func(target config.AvailabilityTarget) {
+				defer wg.Done()
+				m.runTarget(runCtx, target)
+			}(target)
+		}
+
+		select {
+		case <-ctx.Done():
+			cancel()
+			wg.Wait()
+			return
+		case <-m.reload:
+			cancel()
+			wg.Wait()
+		}
+	}
+}
+
+func (m *availabilityProbeModule) runTarget(ctx context.Context, target config.AvailabilityTarget) {
+	interval := availabilityProbeInterval(target)
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	// Probe immediately so a fresh assignment reports before the server's
+	// staleness window opens instead of after a full interval.
+	m.check(ctx, target)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			m.check(ctx, target)
+		}
+	}
+}
+
+func availabilityProbeInterval(target config.AvailabilityTarget) time.Duration {
+	interval := time.Duration(target.EffectivePollIntervalSecs()) * time.Second
+	if interval < availabilityMinInterval {
+		return availabilityMinInterval
+	}
+	if interval > availabilityMaxInterval {
+		return availabilityMaxInterval
+	}
+	return interval
+}
+
+func (m *availabilityProbeModule) check(ctx context.Context, target config.AvailabilityTarget) {
+	timeout := time.Duration(target.EffectiveTimeoutMillis()) * time.Millisecond
+	probeCtx, cancel := context.WithTimeout(ctx, timeout)
+	start := m.now()
+	outcome, err := m.probe(probeCtx, target)
+	latency := m.now().Sub(start)
+	cancel()
+
+	if ctx.Err() != nil {
+		// A shutdown or reassignment cancelled the check mid-flight. The
+		// failure it produced describes the agent, not the target.
+		return
+	}
+	if latency < 0 {
+		latency = 0
+	}
+
+	result := agentshost.AvailabilityProbeResult{
+		TargetID:      target.ID,
+		Outcome:       string(outcome),
+		LatencyMillis: latency.Milliseconds(),
+		CheckedAt:     m.now().UTC(),
+	}
+	if err != nil {
+		message := strings.TrimSpace(err.Error())
+		if len(message) > availabilityErrorLimit {
+			message = message[:availabilityErrorLimit]
+		}
+		result.Error = message
+	}
+
+	m.enqueue(result)
+
+	m.logger.Debug().
+		Str("targetID", target.ID).
+		Str("outcome", result.Outcome).
+		Int64("latencyMillis", result.LatencyMillis).
+		Msg("Completed assigned availability check")
+}
+
+// enqueue queues one completed observation for the next report. The queue is
+// bounded and drops the oldest entry when full, so a long outage costs the
+// earliest observations rather than unbounded memory.
+func (m *availabilityProbeModule) enqueue(result agentshost.AvailabilityProbeResult) {
+	m.mu.Lock()
+	m.sequence++
+	pending := pendingAvailabilityResult{sequence: m.sequence, result: result}
+	m.mu.Unlock()
+	m.pending.Push(pending)
+}
+
+// snapshotForReport returns every queued result without removing it and marks
+// the batch as in flight. Nothing leaves the queue until the primary
+// destination has accepted it.
+func (m *availabilityProbeModule) snapshotForReport() []agentshost.AvailabilityProbeResult {
+	if m == nil {
+		return nil
+	}
+	items := m.pending.Items()
+	if len(items) == 0 {
+		m.mu.Lock()
+		m.inFlight = 0
+		m.mu.Unlock()
+		return nil
+	}
+	results := make([]agentshost.AvailabilityProbeResult, 0, len(items))
+	for _, item := range items {
+		results = append(results, item.result)
+	}
+
+	m.mu.Lock()
+	m.inFlight = items[len(items)-1].sequence
+	m.mu.Unlock()
+	return results
+}
+
+// commitDelivered drops the results the primary destination accepted. Results
+// queued while the report was in flight carry a higher sequence and survive
+// for the next report, so no observation is delivered twice or lost.
+func (m *availabilityProbeModule) commitDelivered() {
+	if m == nil {
+		return
+	}
+	m.mu.Lock()
+	delivered := m.inFlight
+	m.inFlight = 0
+	m.mu.Unlock()
+	if delivered == 0 {
+		return
+	}
+	for {
+		item, ok := m.pending.Peek()
+		if !ok || item.sequence > delivered {
+			return
+		}
+		m.pending.Pop()
+	}
+}
+
+// discardInFlight forgets the in-flight marker when a report is buffered
+// instead of sent, so a later success cannot retire results the server never
+// received.
+func (m *availabilityProbeModule) discardInFlight() {
+	if m == nil {
+		return
+	}
+	m.mu.Lock()
+	m.inFlight = 0
+	m.mu.Unlock()
+}
+
+// moduleStatus surfaces the probe module in AgentInfo.Modules only while the
+// server has something assigned to this agent. An agent with no assignment has
+// no module to report on.
+func (m *availabilityProbeModule) moduleStatus() (agentshost.ModuleStatus, bool) {
+	if m == nil {
+		return agentshost.ModuleStatus{}, false
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if len(m.targets) == 0 {
+		return agentshost.ModuleStatus{}, false
+	}
+	state := availabilityModuleStateStarting
+	if m.running {
+		state = availabilityModuleStateRunning
+	}
+	return agentshost.ModuleStatus{
+		Name:      availabilityModuleName,
+		Enabled:   true,
+		State:     state,
+		UpdatedAt: m.updatedAt,
+	}, true
+}
+
+// AvailabilityTargetsFromSetting decodes the availability assignments carried
+// by a remote-config setting value. The payload is re-marshalled through JSON
+// so the agent shares the server's field names and ignores keys it does not
+// know.
+func AvailabilityTargetsFromSetting(value interface{}) ([]config.AvailabilityTarget, error) {
+	if value == nil {
+		return nil, nil
+	}
+	raw, err := json.Marshal(value)
+	if err != nil {
+		return nil, fmt.Errorf("encode availability targets: %w", err)
+	}
+	var targets []config.AvailabilityTarget
+	if err := json.Unmarshal(raw, &targets); err != nil {
+		return nil, fmt.Errorf("decode availability targets: %w", err)
+	}
+	return normalizeAvailabilityAssignments(targets), nil
+}
+
+// availabilityTargetsFromSettings reports the assignments in a remote-config
+// payload. The server omits the key entirely when it has nothing assigned to
+// this agent, so a missing key means "no assignments" rather than "unchanged".
+func availabilityTargetsFromSettings(settings map[string]interface{}) ([]config.AvailabilityTarget, error) {
+	if settings == nil {
+		return nil, nil
+	}
+	value, ok := settings[availabilitySettingsKey]
+	if !ok {
+		return nil, nil
+	}
+	return AvailabilityTargetsFromSetting(value)
+}

internal/hostagent/availability_test.go

@@ -0,0 +1,556 @@
+package hostagent
+
+import (
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe"
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
+	"github.com/rs/zerolog"
+)
+
+func testProbeModule(t *testing.T, probe func(context.Context, config.AvailabilityTarget) (availabilityprobe.Outcome, error)) *availabilityProbeModule {
+	t.Helper()
+	logger := zerolog.New(io.Discard)
+	module := newAvailabilityProbeModule(logger, nil)
+	if probe != nil {
+		module.probe = probe
+	}
+	return module
+}
+
+func availabilitySetting(entries ...map[string]interface{}) map[string]interface{} {
+	values := make([]interface{}, 0, len(entries))
+	for _, entry := range entries {
+		values = append(values, entry)
+	}
+	return map[string]interface{}{availabilitySettingsKey: values}
+}
+
+func TestAvailabilityTargetsFromSettingsDecodesServerPayload(t *testing.T) {
+	// Mirrors the payload shape built by the server for an assigned agent,
+	// including a key the agent does not know about.
+	settings := availabilitySetting(map[string]interface{}{
+		"id":                  "udp-check",
+		"name":                "UDP check",
+		"targetKind":          "device",
+		"address":             "sensor.local",
+		"protocol":            "udp",
+		"port":                float64(5353),
+		"udpMode":             "response-required",
+		"udpRequest":          "ping",
+		"udpExpectedResponse": "pong",
+		"enabled":             true,
+		"pollIntervalSeconds": float64(45),
+		"timeoutMillis":       float64(1500),
+		"unknownFutureField":  "ignored",
+	})
+
+	targets, err := availabilityTargetsFromSettings(settings)
+	if err != nil {
+		t.Fatalf("availabilityTargetsFromSettings() error = %v", err)
+	}
+	if len(targets) != 1 {
+		t.Fatalf("targets = %+v, want one assignment", targets)
+	}
+	target := targets[0]
+	if target.ID != "udp-check" || target.Address != "sensor.local" {
+		t.Fatalf("target = %+v", target)
+	}
+	if target.Protocol != config.AvailabilityProbeUDP || target.Port != 5353 {
+		t.Fatalf("target protocol/port = %v/%d", target.Protocol, target.Port)
+	}
+	if target.UDPRequest != "ping" || target.UDPExpected != "pong" {
+		t.Fatalf("target udp payloads = %q/%q", target.UDPRequest, target.UDPExpected)
+	}
+	if target.EffectivePollIntervalSecs() != 45 || target.EffectiveTimeoutMillis() != 1500 {
+		t.Fatalf("target schedule = %ds/%dms", target.EffectivePollIntervalSecs(), target.EffectiveTimeoutMillis())
+	}
+	if !target.Enabled {
+		t.Fatal("target should be enabled")
+	}
+	// Server-only concerns are never carried in an agent assignment.
+	if target.ProbeAgentID != "" || target.LinkedResourceID != "" {
+		t.Fatalf("target carried server-side fields: %+v", target)
+	}
+}
+
+func TestAvailabilityTargetsFromSettingsHandlesMissingAndUnusableValues(t *testing.T) {
+	targets, err := availabilityTargetsFromSettings(nil)
+	if err != nil || targets != nil {
+		t.Fatalf("nil settings = (%+v, %v), want no assignments", targets, err)
+	}
+
+	targets, err = availabilityTargetsFromSettings(map[string]interface{}{"interval": "30s"})
+	if err != nil || targets != nil {
+		t.Fatalf("settings without the key = (%+v, %v), want no assignments", targets, err)
+	}
+
+	targets, err = availabilityTargetsFromSettings(map[string]interface{}{availabilitySettingsKey: nil})
+	if err != nil || targets != nil {
+		t.Fatalf("null value = (%+v, %v), want no assignments", targets, err)
+	}
+
+	if _, err := availabilityTargetsFromSettings(map[string]interface{}{availabilitySettingsKey: "not-a-list"}); err == nil {
+		t.Fatal("garbage value error = nil, want a decode failure")
+	}
+
+	// An entry without an ID cannot be reported against and is dropped.
+	targets, err = availabilityTargetsFromSettings(availabilitySetting(
+		map[string]interface{}{"address": "orphan.local", "protocol": "icmp", "enabled": true},
+		map[string]interface{}{"id": "keep", "address": "keep.local", "protocol": "icmp", "enabled": true},
+	))
+	if err != nil {
+		t.Fatalf("availabilityTargetsFromSettings() error = %v", err)
+	}
+	if len(targets) != 1 || targets[0].ID != "keep" {
+		t.Fatalf("targets = %+v, want only the identified assignment", targets)
+	}
+}
+
+func TestApplyRemoteAvailabilityTargetsReconcilesAssignments(t *testing.T) {
+	agent := &Agent{
+		logger:       zerolog.New(io.Discard),
+		availability: testProbeModule(t, nil),
+	}
+
+	assigned := availabilitySetting(map[string]interface{}{
+		"id": "remote-a", "address": "a.local", "protocol": "icmp", "enabled": true,
+	})
+	agent.ApplyRemoteConfig(assigned, nil)
+	if got := agent.availability.assignments(); len(got) != 1 || got[0].ID != "remote-a" {
+		t.Fatalf("assignments = %+v, want remote-a", got)
+	}
+
+	// A repeated fetch of the same assignment must not restart the schedule.
+	if agent.availability.applyTargets(agent.availability.assignments()) {
+		t.Fatal("unchanged assignments reported a schedule change")
+	}
+
+	// An unreadable payload leaves the current schedule alone.
+	agent.ApplyRemoteConfig(map[string]interface{}{availabilitySettingsKey: "not-a-list"}, nil)
+	if got := agent.availability.assignments(); len(got) != 1 {
+		t.Fatalf("assignments after a garbage payload = %+v, want the previous schedule", got)
+	}
+
+	// The server omits the key once nothing is assigned: that unassigns.
+	agent.ApplyRemoteConfig(map[string]interface{}{"interval": "30s"}, nil)
+	if got := agent.availability.assignments(); len(got) != 0 {
+		t.Fatalf("assignments after unassignment = %+v, want none", got)
+	}
+}
+
+func TestNewSeedsAvailabilityAssignmentsFromStartupConfig(t *testing.T) {
+	recorder := &availabilityReportRecorder{}
+	agent := newAvailabilityDeliveryAgent(t, recorder, config.AvailabilityTarget{
+		ID:       "boot-target",
+		Address:  "boot.local",
+		Protocol: config.AvailabilityProbeICMP,
+		Enabled:  true,
+	})
+
+	// The assignment fetched before startup must be live without waiting for
+	// the first remote-config refresh.
+	if got := agent.availability.assignments(); len(got) != 1 || got[0].ID != "boot-target" {
+		t.Fatalf("assignments = %+v, want the startup assignment", got)
+	}
+	status, ok := agent.availability.moduleStatus()
+	if !ok || status.Name != availabilityModuleName {
+		t.Fatalf("module status = (%+v, %v), want the availability module", status, ok)
+	}
+}
+
+func TestAvailabilityProbeIntervalClampsToServerBounds(t *testing.T) {
+	tests := []struct {
+		seconds int
+		want    time.Duration
+	}{
+		{seconds: 0, want: time.Duration(config.DefaultAvailabilityPollIntervalSecs) * time.Second},
+		{seconds: 1, want: availabilityMinInterval},
+		{seconds: 45, want: 45 * time.Second},
+		{seconds: 86400, want: availabilityMaxInterval},
+	}
+	for _, test := range tests {
+		target := config.AvailabilityTarget{PollIntervalSecs: test.seconds}
+		if got := availabilityProbeInterval(target); got != test.want {
+			t.Fatalf("availabilityProbeInterval(%ds) = %v, want %v", test.seconds, got, test.want)
+		}
+	}
+}
+
+func TestAvailabilityModuleSchedulerRunsAssignedTargetsAndReschedules(t *testing.T) {
+	var (
+		mu      sync.Mutex
+		checked []string
+	)
+	module := testProbeModule(t, func(_ context.Context, target config.AvailabilityTarget) (availabilityprobe.Outcome, error) {
+		mu.Lock()
+		checked = append(checked, target.ID)
+		mu.Unlock()
+		if target.ID == "down" {
+			return availabilityprobe.OutcomeUnreachable, errors.New("icmp probe timed out")
+		}
+		return availabilityprobe.OutcomeReachable, nil
+	})
+	module.applyTargets([]config.AvailabilityTarget{
+		{ID: "up", Address: "up.local", Protocol: config.AvailabilityProbeICMP, Enabled: true},
+		{ID: "down", Address: "down.local", Protocol: config.AvailabilityProbeICMP, Enabled: true},
+		{ID: "paused", Address: "paused.local", Protocol: config.AvailabilityProbeICMP, Enabled: false},
+	})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		module.Run(ctx)
+	}()
+
+	waitForPending := func(want int) {
+		t.Helper()
+		deadline := time.After(5 * time.Second)
+		for {
+			if module.pending.Len() >= want {
+				return
+			}
+			select {
+			case <-time.After(5 * time.Millisecond):
+			case <-deadline:
+				t.Fatalf("timed out waiting for %d queued results, have %d", want, module.pending.Len())
+			}
+		}
+	}
+
+	// Enabled targets probe immediately rather than after a full interval.
+	waitForPending(2)
+	results := module.snapshotForReport()
+	if len(results) != 2 {
+		t.Fatalf("results = %+v, want one per enabled target", results)
+	}
+	byTarget := make(map[string]agentshost.AvailabilityProbeResult, len(results))
+	for _, result := range results {
+		byTarget[result.TargetID] = result
+	}
+	if got := byTarget["up"].Outcome; got != string(availabilityprobe.OutcomeReachable) {
+		t.Fatalf("up outcome = %q", got)
+	}
+	if got := byTarget["down"]; got.Outcome != string(availabilityprobe.OutcomeUnreachable) || got.Error == "" {
+		t.Fatalf("down result = %+v, want a failure with an explanation", got)
+	}
+	if byTarget["up"].CheckedAt.IsZero() {
+		t.Fatal("result is missing its observation time")
+	}
+	if _, ok := byTarget["paused"]; ok {
+		t.Fatalf("disabled target was probed: %+v", results)
+	}
+
+	module.commitDelivered()
+	if module.pending.Len() != 0 {
+		t.Fatalf("pending = %d after delivery, want 0", module.pending.Len())
+	}
+
+	// Reassignment stops the old workers and starts the new set at once.
+	mu.Lock()
+	checked = nil
+	mu.Unlock()
+	module.applyTargets([]config.AvailabilityTarget{
+		{ID: "replacement", Address: "new.local", Protocol: config.AvailabilityProbeICMP, Enabled: true},
+	})
+	waitForPending(1)
+	reassigned := module.snapshotForReport()
+	if len(reassigned) != 1 || reassigned[0].TargetID != "replacement" {
+		t.Fatalf("results after reassignment = %+v, want only the replacement", reassigned)
+	}
+	mu.Lock()
+	for _, id := range checked {
+		if id != "replacement" {
+			t.Fatalf("target %q ran after it was unassigned", id)
+		}
+	}
+	mu.Unlock()
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("scheduler did not stop on context cancellation")
+	}
+}
+
+func TestAvailabilityModuleQueueDropsOldestBeyondCapacity(t *testing.T) {
+	module := testProbeModule(t, nil)
+	for i := 0; i < availabilityPendingCapacity+50; i++ {
+		module.enqueue(agentshost.AvailabilityProbeResult{
+			TargetID:  "target",
+			Outcome:   string(availabilityprobe.OutcomeReachable),
+			CheckedAt: time.Unix(int64(i), 0).UTC(),
+		})
+	}
+
+	if got := module.pending.Len(); got != availabilityPendingCapacity {
+		t.Fatalf("pending = %d, want the capacity %d", got, availabilityPendingCapacity)
+	}
+	results := module.snapshotForReport()
+	if len(results) != availabilityPendingCapacity {
+		t.Fatalf("results = %d, want %d", len(results), availabilityPendingCapacity)
+	}
+	// The oldest 50 observations were dropped, not the newest.
+	if got := results[0].CheckedAt; !got.Equal(time.Unix(50, 0).UTC()) {
+		t.Fatalf("oldest retained result = %v, want the 51st observation", got)
+	}
+	if got := results[len(results)-1].CheckedAt; !got.Equal(time.Unix(availabilityPendingCapacity+49, 0).UTC()) {
+		t.Fatalf("newest retained result = %v", got)
+	}
+}
+
+func TestAvailabilityModuleStatusOnlyWhileAssigned(t *testing.T) {
+	module := testProbeModule(t, nil)
+	if _, ok := module.moduleStatus(); ok {
+		t.Fatal("module status reported without an assignment")
+	}
+
+	module.applyTargets([]config.AvailabilityTarget{
+		{ID: "remote", Address: "remote.local", Protocol: config.AvailabilityProbeICMP, Enabled: true},
+	})
+	status, ok := module.moduleStatus()
+	if !ok {
+		t.Fatal("module status missing for an assigned agent")
+	}
+	if status.Name != availabilityModuleName || !status.Enabled {
+		t.Fatalf("status = %+v", status)
+	}
+	if status.State != availabilityModuleStateStarting {
+		t.Fatalf("state = %q before the scheduler runs, want %q", status.State, availabilityModuleStateStarting)
+	}
+	if status.UpdatedAt.IsZero() {
+		t.Fatal("status is missing its assignment time")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		module.Run(ctx)
+	}()
+	deadline := time.After(5 * time.Second)
+	for {
+		status, _ = module.moduleStatus()
+		if status.State == availabilityModuleStateRunning {
+			break
+		}
+		select {
+		case <-time.After(5 * time.Millisecond):
+		case <-deadline:
+			t.Fatalf("state = %q, want %q once the scheduler is up", status.State, availabilityModuleStateRunning)
+		}
+	}
+	cancel()
+	<-done
+
+	module.applyTargets(nil)
+	if _, ok := module.moduleStatus(); ok {
+		t.Fatal("module status survived unassignment")
+	}
+}
+
+func TestAgentModuleStatusAppendsAvailabilityToRuntimeModules(t *testing.T) {
+	agent := &Agent{
+		logger:       zerolog.New(io.Discard),
+		availability: testProbeModule(t, nil),
+		cfg: Config{
+			ModuleStatus: func() []agentshost.ModuleStatus {
+				return []agentshost.ModuleStatus{{Name: "host", Enabled: true, State: "running"}}
+			},
+		},
+	}
+
+	statuses := agent.currentModuleStatus()
+	if len(statuses) != 1 || statuses[0].Name != "host" {
+		t.Fatalf("modules = %+v, want only the runtime modules", statuses)
+	}
+
+	agent.availability.applyTargets([]config.AvailabilityTarget{
+		{ID: "remote", Address: "remote.local", Protocol: config.AvailabilityProbeICMP, Enabled: true},
+	})
+	statuses = agent.currentModuleStatus()
+	if len(statuses) != 2 || statuses[1].Name != availabilityModuleName {
+		t.Fatalf("modules = %+v, want the availability module appended", statuses)
+	}
+}
+
+// availabilityReportRecorder is a Pulse report endpoint that can be switched
+// into failure so retention across a delivery outage is observable.
+type availabilityReportRecorder struct {
+	mu      sync.Mutex
+	reports []agentshost.Report
+	fail    bool
+}
+
+func (r *availabilityReportRecorder) setFail(fail bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.fail = fail
+}
+
+func (r *availabilityReportRecorder) received() []agentshost.Report {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([]agentshost.Report(nil), r.reports...)
+}
+
+func (r *availabilityReportRecorder) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	r.mu.Lock()
+	fail := r.fail
+	r.mu.Unlock()
+	if fail {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	var body io.Reader = req.Body
+	if req.Header.Get("Content-Encoding") == "gzip" {
+		gz, err := gzip.NewReader(req.Body)
+		if err != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		defer gz.Close()
+		body = gz
+	}
+	var report agentshost.Report
+	if err := json.NewDecoder(body).Decode(&report); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	r.mu.Lock()
+	r.reports = append(r.reports, report)
+	r.mu.Unlock()
+	_ = json.NewEncoder(w).Encode(map[string]any{"success": true})
+}
+
+func newAvailabilityDeliveryAgent(t *testing.T, recorder *availabilityReportRecorder, targets ...config.AvailabilityTarget) *Agent {
+	t.Helper()
+	server := httptest.NewServer(recorder)
+	t.Cleanup(server.Close)
+
+	logger := zerolog.New(io.Discard)
+	agent, err := New(Config{
+		PulseURL:            server.URL,
+		APIToken:            "test-token",
+		Interval:            time.Minute,
+		HostnameOverride:    "probe-agent",
+		StateDir:            t.TempDir(),
+		Logger:              &logger,
+		Collector:           &mockCollector{},
+		AvailabilityTargets: targets,
+		packageUpdates:      newPackageUpdateManager("windows", nil),
+		storageCleanup:      newStorageCleanupManager("windows", nil),
+	})
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	return agent
+}
+
+func TestAvailabilityResultsSurviveFailedDeliveryAndAreSentOnce(t *testing.T) {
+	recorder := &availabilityReportRecorder{}
+	agent := newAvailabilityDeliveryAgent(t, recorder)
+	ctx := context.Background()
+
+	for _, id := range []string{"target-a", "target-b"} {
+		agent.availability.enqueue(agentshost.AvailabilityProbeResult{
+			TargetID:  id,
+			Outcome:   string(availabilityprobe.OutcomeReachable),
+			CheckedAt: time.Now().UTC(),
+		})
+	}
+
+	recorder.setFail(true)
+	if err := agent.process(ctx); err != nil {
+		t.Fatalf("process() during outage error = %v", err)
+	}
+	if got := agent.availability.pending.Len(); got != 2 {
+		t.Fatalf("pending = %d after a failed delivery, want the results retained", got)
+	}
+	buffered, ok := agent.reportBuffer.Peek()
+	if !ok {
+		t.Fatal("failed report was not buffered")
+	}
+	if len(buffered.AvailabilityResults) != 0 {
+		t.Fatalf("buffered report kept %d results; a retry would double count them", len(buffered.AvailabilityResults))
+	}
+
+	// A result observed during the outage joins the same batch.
+	agent.availability.enqueue(agentshost.AvailabilityProbeResult{
+		TargetID:  "target-c",
+		Outcome:   string(availabilityprobe.OutcomeUnreachable),
+		CheckedAt: time.Now().UTC(),
+		Error:     "icmp probe timed out",
+	})
+
+	recorder.setFail(false)
+	if err := agent.process(ctx); err != nil {
+		t.Fatalf("process() after recovery error = %v", err)
+	}
+	if got := agent.availability.pending.Len(); got != 0 {
+		t.Fatalf("pending = %d after a successful delivery, want 0", got)
+	}
+
+	// A later report must not repeat what the server already counted.
+	if err := agent.process(ctx); err != nil {
+		t.Fatalf("process() error = %v", err)
+	}
+
+	delivered := map[string]int{}
+	for _, report := range recorder.received() {
+		for _, result := range report.AvailabilityResults {
+			delivered[result.TargetID]++
+		}
+	}
+	if len(delivered) != 3 {
+		t.Fatalf("delivered targets = %+v, want all three observations", delivered)
+	}
+	for id, count := range delivered {
+		if count != 1 {
+			t.Fatalf("target %q was delivered %d times, want exactly once", id, count)
+		}
+	}
+}
+
+func TestAvailabilityResultsAreNotRetiredWhenTheTokenIsRejected(t *testing.T) {
+	recorder := &availabilityReportRecorder{}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer server.Close()
+
+	agent := newAvailabilityDeliveryAgent(t, recorder)
+	agent.trimmedPulseURL = server.URL
+	agent.availability.enqueue(agentshost.AvailabilityProbeResult{
+		TargetID:  "target-a",
+		Outcome:   string(availabilityprobe.OutcomeReachable),
+		CheckedAt: time.Now().UTC(),
+	})
+
+	// A rejected token drops the report instead of buffering it, so the
+	// results have to stay queued for whenever the token is replaced.
+	if err := agent.process(context.Background()); err != nil {
+		t.Fatalf("process() error = %v", err)
+	}
+	if got := agent.availability.pending.Len(); got != 1 {
+		t.Fatalf("pending = %d after a rejected report, want the result retained", got)
+	}
+	if !agent.reportBuffer.IsEmpty() {
+		t.Fatal("a rejected report should not be buffered")
+	}
+}

internal/monitoring/availability_poller.go

@@ -412,6 +412,7 @@ func availabilityResourceFromTarget(target config.AvailabilityTarget, status Ava
 		Address:             target.Address,
 		Protocol:            string(target.Protocol),
 		ProbeOutcome:        status.Outcome,
+		ProbeAgentID:        status.ProbeAgentID,
 		UDPMode:             string(target.UDPMode),
 		Port:                target.Port,
 		Path:                target.Path,

internal/monitoring/availability_probe_agent.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe"
 	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
 	pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing"
 	"github.com/rs/zerolog/log"
 )
@@ -29,6 +30,33 @@ type ProbeAvailabilityResult struct {
 	Error         string
 }
 
+// probeAvailabilityResultsFromReport converts the wire results carried by a
+// host agent report. Outcome vocabulary is normalized here so the ingestion
+// path never has to trust an agent's spelling; probeResultOutcome then decides
+// what an unknown outcome means for failure accounting.
+func probeAvailabilityResultsFromReport(reported []agentshost.AvailabilityProbeResult) []ProbeAvailabilityResult {
+	if len(reported) == 0 {
+		return nil
+	}
+	results := make([]ProbeAvailabilityResult, 0, len(reported))
+	for _, entry := range reported {
+		outcome := availabilityprobe.Outcome(strings.ToLower(strings.TrimSpace(entry.Outcome)))
+		switch outcome {
+		case availabilityprobe.OutcomeReachable, availabilityprobe.OutcomeUnreachable, availabilityprobe.OutcomeIndeterminate:
+		default:
+			outcome = availabilityprobe.OutcomeIndeterminate
+		}
+		results = append(results, ProbeAvailabilityResult{
+			TargetID:      strings.TrimSpace(entry.TargetID),
+			Outcome:       outcome,
+			LatencyMillis: entry.LatencyMillis,
+			CheckedAt:     entry.CheckedAt,
+			Error:         strings.TrimSpace(entry.Error),
+		})
+	}
+	return results
+}
+
 // effectiveProbeAgentID returns the host agent that currently owns execution of
 // the target. It collapses to local execution ("") whenever the external probe
 // entitlement is absent, so a license lapse resumes local polling instead of

internal/monitoring/availability_probe_agent_test.go

@@ -4,8 +4,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
 	"github.com/rcourtman/pulse-go-rewrite/internal/config"
 	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
 	pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing"
 )
 
@@ -393,9 +395,94 @@ func TestAvailabilitySupplementalRecordsPresentStaleProbeAsIndeterminate(t *test
 	if data.LastError != availabilityProbeStaleError {
 		t.Fatalf("availability last error = %q, want %q", data.LastError, availabilityProbeStaleError)
 	}
+	if data.ProbeAgentID != "agent-1" {
+		t.Fatalf("availability probe agent = %q, want agent-1 so the UI can attribute the source", data.ProbeAgentID)
+	}
 
 	statuses := availabilityPollProvider{}.ConnectionStatuses(monitor)
 	if statuses["availability-remote"] {
 		t.Fatalf("connection statuses = %+v, want a stale probe to read as not connected", statuses)
 	}
 }
+
+func TestApplyHostReportIngestsAssignedAvailabilityResults(t *testing.T) {
+	monitor := newProbeAgentTestMonitor(t,
+		probeAgentTarget("remote", "probe-host"),
+		probeAgentTarget("foreign", "other-agent"),
+	)
+	monitor.alertManager = alerts.NewManager()
+	t.Cleanup(func() { monitor.alertManager.Stop() })
+	monitor.config = &config.Config{}
+	monitor.rateTracker = NewRateTracker()
+	monitor.hostTokenBindings = make(map[string]string)
+	monitor.SetLicenseChecker(licenseWithExternalProbe(true))
+
+	checkedAt := time.Now().UTC()
+	host, err := monitor.ApplyHostReport(agentshost.Report{
+		Agent: agentshost.AgentInfo{ID: "probe-host", Version: "6.0.0", IntervalSeconds: 30},
+		Host: agentshost.HostInfo{
+			ID:       "probe-host",
+			Hostname: "probe-host.local",
+			Platform: "linux",
+		},
+		AvailabilityResults: []agentshost.AvailabilityProbeResult{
+			{TargetID: "remote", Outcome: "reachable", LatencyMillis: 17, CheckedAt: checkedAt},
+			{TargetID: "foreign", Outcome: "reachable", LatencyMillis: 3, CheckedAt: checkedAt},
+		},
+		Timestamp: checkedAt,
+	}, nil)
+	if err != nil {
+		t.Fatalf("ApplyHostReport() error = %v", err)
+	}
+	// The ownership check compares the host ID that GetHostAgentConfig is
+	// keyed by, so the report must resolve to exactly that identity.
+	if host.ID != "probe-host" {
+		t.Fatalf("host ID = %q, want the assigned probe agent ID", host.ID)
+	}
+
+	statuses := monitor.AvailabilityStatusSnapshot()
+	status, ok := statuses["remote"]
+	if !ok {
+		t.Fatalf("statuses = %+v, want the assigned target applied", statuses)
+	}
+	if !status.Available || status.LatencyMillis != 17 {
+		t.Fatalf("status = %+v, want the reported observation", status)
+	}
+	if status.ProbeAgentID != "probe-host" {
+		t.Fatalf("probe agent attribution = %q, want probe-host", status.ProbeAgentID)
+	}
+	if !status.LastChecked.Equal(checkedAt) {
+		t.Fatalf("last checked = %v, want %v", status.LastChecked, checkedAt)
+	}
+	if _, ok := statuses["foreign"]; ok {
+		t.Fatal("a report claimed a target assigned to another agent")
+	}
+}
+
+func TestProbeAvailabilityResultsFromReportNormalizesOutcomes(t *testing.T) {
+	if got := probeAvailabilityResultsFromReport(nil); got != nil {
+		t.Fatalf("results = %+v, want nil for a report without probe work", got)
+	}
+
+	checkedAt := time.Now().UTC()
+	results := probeAvailabilityResultsFromReport([]agentshost.AvailabilityProbeResult{
+		{TargetID: " padded ", Outcome: " REACHABLE ", LatencyMillis: 4, CheckedAt: checkedAt},
+		{TargetID: "b", Outcome: "unreachable", Error: " icmp probe timed out "},
+		{TargetID: "c", Outcome: "who-knows"},
+		{TargetID: "d", Outcome: ""},
+	})
+	if len(results) != 4 {
+		t.Fatalf("results = %+v, want one per reported observation", results)
+	}
+	if results[0].TargetID != "padded" || results[0].Outcome != AvailabilityProbeReachable {
+		t.Fatalf("first result = %+v", results[0])
+	}
+	if results[1].Error != "icmp probe timed out" || results[1].Outcome != AvailabilityProbeUnreachable {
+		t.Fatalf("second result = %+v", results[1])
+	}
+	for _, result := range results[2:] {
+		if result.Outcome != AvailabilityProbeIndeterminate {
+			t.Fatalf("result %+v: unknown outcomes must read as indeterminate", result)
+		}
+	}
+}

internal/monitoring/monitor_agents.go

@@ -2851,6 +2851,9 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
 
 	// Store cluster peer sensor data if present and evict stale entries
 	m.applyClusterSensors(report.ClusterSensors, observedAt)
+	// Availability results are ingested only once the host identity is
+	// committed, because ownership is checked against that host ID.
+	m.ApplyProbeAvailabilityResults(host.ID, probeAvailabilityResultsFromReport(report.AvailabilityResults))
 	m.persistHostContinuity(host, report, reportOrder)
 	m.refreshUnifiedResourceStoreAfterAgentReport()
 

internal/unifiedresources/types.go

@@ -1686,6 +1686,7 @@ type AvailabilityData struct {
 	Address               string                             `json:"address,omitempty"`
 	Protocol              string                             `json:"protocol,omitempty"`
 	ProbeOutcome          string                             `json:"probeOutcome,omitempty"`
+	ProbeAgentID          string                             `json:"probeAgentId,omitempty"`
 	UDPMode               string                             `json:"udpMode,omitempty"`
 	Port                  int                                `json:"port,omitempty"`
 	Path                  string                             `json:"path,omitempty"`

pkg/agents/host/report.go

@@ -54,9 +54,25 @@ type Report struct {
 	Unraid         *UnraidStorage       `json:"unraid,omitempty"`
 	Ceph           *CephCluster         `json:"ceph,omitempty"`
 	ClusterSensors []ClusterNodeSensors `json:"clusterSensors,omitempty"`
-	Tags           []string             `json:"tags,omitempty"`
-	Timestamp      time.Time            `json:"timestamp"`
-	SequenceID     string               `json:"sequenceId,omitempty"`
+	// AvailabilityResults carries availability checks the agent executed on
+	// behalf of the server for targets assigned to it. The server owns failure
+	// accounting; the agent only reports what each check observed.
+	AvailabilityResults []AvailabilityProbeResult `json:"availabilityResults,omitempty"`
+	Tags                []string                  `json:"tags,omitempty"`
+	Timestamp           time.Time                 `json:"timestamp"`
+	SequenceID          string                    `json:"sequenceId,omitempty"`
+}
+
+// AvailabilityProbeResult is one completed availability check reported by a
+// remote probe agent. Outcome mirrors the shared probe vocabulary
+// ("reachable", "unreachable", "indeterminate"); anything else is treated as
+// indeterminate by the server.
+type AvailabilityProbeResult struct {
+	TargetID      string    `json:"targetId"`
+	Outcome       string    `json:"outcome"`
+	LatencyMillis int64     `json:"latencyMillis"`
+	CheckedAt     time.Time `json:"checkedAt"`
+	Error         string    `json:"error,omitempty"`
 }
 
 // ClusterNodeSensors contains temperature sensor data collected from a Proxmox

pkg/agents/host/report_test.go

@@ -510,3 +510,54 @@ func contains(s, substr string) bool {
 	}
 	return false
 }
+
+func TestReportAvailabilityResultsJSONRoundTrip(t *testing.T) {
+	checkedAt := time.Date(2026, 7, 27, 10, 30, 0, 0, time.UTC)
+	report := Report{
+		Agent: AgentInfo{ID: "probe-agent"},
+		Host:  HostInfo{Hostname: "probe-agent.local"},
+		AvailabilityResults: []AvailabilityProbeResult{
+			{TargetID: "target-1", Outcome: "reachable", LatencyMillis: 12, CheckedAt: checkedAt},
+			{TargetID: "target-2", Outcome: "unreachable", CheckedAt: checkedAt, Error: "icmp probe timed out"},
+		},
+		Timestamp: checkedAt,
+	}
+
+	data, err := json.Marshal(report)
+	if err != nil {
+		t.Fatalf("Failed to marshal: %v", err)
+	}
+	if !contains(string(data), `"availabilityResults":[`) {
+		t.Fatalf("availability results missing from payload: %s", data)
+	}
+	if contains(string(data), `"error":""`) {
+		t.Fatalf("empty probe error should be omitted: %s", data)
+	}
+
+	var decoded Report
+	if err := json.Unmarshal(data, &decoded); err != nil {
+		t.Fatalf("Failed to unmarshal: %v", err)
+	}
+	if len(decoded.AvailabilityResults) != 2 {
+		t.Fatalf("availability results = %+v, want 2", decoded.AvailabilityResults)
+	}
+	first := decoded.AvailabilityResults[0]
+	if first.TargetID != "target-1" || first.Outcome != "reachable" || first.LatencyMillis != 12 {
+		t.Fatalf("first result = %+v", first)
+	}
+	if !first.CheckedAt.Equal(checkedAt) {
+		t.Fatalf("checkedAt = %v, want %v", first.CheckedAt, checkedAt)
+	}
+	if decoded.AvailabilityResults[1].Error != "icmp probe timed out" {
+		t.Fatalf("second result = %+v", decoded.AvailabilityResults[1])
+	}
+
+	// A report without probe assignments must not carry the section at all.
+	bare, err := json.Marshal(Report{Host: HostInfo{Hostname: "plain"}})
+	if err != nil {
+		t.Fatalf("Failed to marshal bare report: %v", err)
+	}
+	if contains(string(bare), "availabilityResults") {
+		t.Fatalf("empty availability results should be omitted: %s", bare)
+	}
+}