diff --git a/control-plane/internal/config/config.go b/control-plane/internal/config/config.go index 71fbb053e..139b973ed 100644 --- a/control-plane/internal/config/config.go +++ b/control-plane/internal/config/config.go @@ -4,6 +4,7 @@ import ( "fmt" // Added for fmt.Errorf "os" // Added for os.Stat, os.ReadFile "path/filepath" // Added for filepath.Join + "strconv" "time" "gopkg.in/yaml.v3" // Added for yaml.Unmarshal @@ -32,10 +33,21 @@ type UIConfig struct { // AgentFieldConfig holds the core AgentField server configuration. type AgentFieldConfig struct { Port int `yaml:"port"` + NodeHealth NodeHealthConfig `yaml:"node_health" mapstructure:"node_health"` ExecutionCleanup ExecutionCleanupConfig `yaml:"execution_cleanup" mapstructure:"execution_cleanup"` ExecutionQueue ExecutionQueueConfig `yaml:"execution_queue" mapstructure:"execution_queue"` } +// NodeHealthConfig holds configuration for agent node health monitoring. +// Zero values are treated as "use default" — set explicitly to override. +type NodeHealthConfig struct { + CheckInterval time.Duration `yaml:"check_interval" mapstructure:"check_interval"` // How often to HTTP health check nodes (0 = default 10s) + CheckTimeout time.Duration `yaml:"check_timeout" mapstructure:"check_timeout"` // Timeout per HTTP health check (0 = default 5s) + ConsecutiveFailures int `yaml:"consecutive_failures" mapstructure:"consecutive_failures"` // Failures before marking inactive (0 = default 3; set 1 for instant) + RecoveryDebounce time.Duration `yaml:"recovery_debounce" mapstructure:"recovery_debounce"` // Wait before allowing inactive->active (0 = default 5s) + HeartbeatStaleThreshold time.Duration `yaml:"heartbeat_stale_threshold" mapstructure:"heartbeat_stale_threshold"` // Heartbeat age before marking stale (0 = default 60s) +} + // ExecutionCleanupConfig holds configuration for execution cleanup and garbage collection type ExecutionCleanupConfig struct { Enabled bool `yaml:"enabled" mapstructure:"enabled" default:"true"` @@ -170,4 +182,31 @@ func applyEnvOverrides(cfg *Config) { if apiKey := os.Getenv("AGENTFIELD_API_AUTH_API_KEY"); apiKey != "" { cfg.API.Auth.APIKey = apiKey } + + // Node health monitoring overrides + if val := os.Getenv("AGENTFIELD_HEALTH_CHECK_INTERVAL"); val != "" { + if d, err := time.ParseDuration(val); err == nil { + cfg.AgentField.NodeHealth.CheckInterval = d + } + } + if val := os.Getenv("AGENTFIELD_HEALTH_CHECK_TIMEOUT"); val != "" { + if d, err := time.ParseDuration(val); err == nil { + cfg.AgentField.NodeHealth.CheckTimeout = d + } + } + if val := os.Getenv("AGENTFIELD_HEALTH_CONSECUTIVE_FAILURES"); val != "" { + if i, err := strconv.Atoi(val); err == nil { + cfg.AgentField.NodeHealth.ConsecutiveFailures = i + } + } + if val := os.Getenv("AGENTFIELD_HEALTH_RECOVERY_DEBOUNCE"); val != "" { + if d, err := time.ParseDuration(val); err == nil { + cfg.AgentField.NodeHealth.RecoveryDebounce = d + } + } + if val := os.Getenv("AGENTFIELD_HEARTBEAT_STALE_THRESHOLD"); val != "" { + if d, err := time.ParseDuration(val); err == nil { + cfg.AgentField.NodeHealth.HeartbeatStaleThreshold = d + } + } } diff --git a/control-plane/internal/handlers/nodes.go b/control-plane/internal/handlers/nodes.go index cec63d038..06a80c2ef 100644 --- a/control-plane/internal/handlers/nodes.go +++ b/control-plane/internal/handlers/nodes.go @@ -299,9 +299,10 @@ var ( heartbeatCache = &HeartbeatCache{ nodes: make(map[string]*CachedNodeData), } - // Only write to DB if heartbeat is older than this threshold - // Must be less than health monitor grace period (12s) to prevent false inactivity - dbUpdateThreshold = 8 * time.Second + // Only write to DB if heartbeat is older than this threshold. + // Reduced from 8s to 2s to keep DB timestamps fresh and prevent + // other systems (reconciliation, health monitor) from seeing stale data. + dbUpdateThreshold = 2 * time.Second ) // shouldUpdateDatabase determines if a heartbeat should trigger a database update diff --git a/control-plane/internal/server/server.go b/control-plane/internal/server/server.go index 860128370..954bab880 100644 --- a/control-plane/internal/server/server.go +++ b/control-plane/internal/server/server.go @@ -116,9 +116,10 @@ func NewAgentFieldServer(cfg *config.Config) (*AgentFieldServer, error) { // Initialize StatusManager for unified status management statusManagerConfig := services.StatusManagerConfig{ - ReconcileInterval: 30 * time.Second, - StatusCacheTTL: 5 * time.Minute, - MaxTransitionTime: 2 * time.Minute, + ReconcileInterval: 30 * time.Second, + StatusCacheTTL: 5 * time.Minute, + MaxTransitionTime: 2 * time.Minute, + HeartbeatStaleThreshold: cfg.AgentField.NodeHealth.HeartbeatStaleThreshold, } // Create UIService first (without StatusManager) @@ -140,8 +141,13 @@ func NewAgentFieldServer(cfg *config.Config) (*AgentFieldServer, error) { executionsUIService := services.NewExecutionsUIService(storageProvider) // Initialize ExecutionsUIService - // Initialize health monitor with StatusManager integration - healthMonitorConfig := services.HealthMonitorConfig{} + // Initialize health monitor with configurable settings + healthMonitorConfig := services.HealthMonitorConfig{ + CheckInterval: cfg.AgentField.NodeHealth.CheckInterval, + CheckTimeout: cfg.AgentField.NodeHealth.CheckTimeout, + ConsecutiveFailures: cfg.AgentField.NodeHealth.ConsecutiveFailures, + RecoveryDebounce: cfg.AgentField.NodeHealth.RecoveryDebounce, + } healthMonitor := services.NewHealthMonitor(storageProvider, healthMonitorConfig, uiService, agentClient, statusManager, presenceManager) presenceManager.SetExpireCallback(healthMonitor.UnregisterAgent) diff --git a/control-plane/internal/services/health_monitor.go b/control-plane/internal/services/health_monitor.go index 40f58f8db..76d2e6dda 100644 --- a/control-plane/internal/services/health_monitor.go +++ b/control-plane/internal/services/health_monitor.go @@ -2,6 +2,7 @@ package services import ( "context" + "fmt" "sync" "time" @@ -13,17 +14,32 @@ import ( "github.com/Agent-Field/agentfield/control-plane/pkg/types" ) +// Health score constants for status updates. +const ( + // healthScoreActive is the score assigned when an HTTP health check passes. + // Below 100 to leave room for "excellent" states (e.g. agent + all MCP servers healthy). + healthScoreActive = 85 + + // healthScoreInactive is the score when an agent fails consecutive health checks. + healthScoreInactive = 0 +) + // HealthMonitorConfig holds configuration for the health monitor service type HealthMonitorConfig struct { - CheckInterval time.Duration // How often to check node health via HTTP + CheckInterval time.Duration // How often to check node health via HTTP + CheckTimeout time.Duration // Timeout for individual HTTP health checks + ConsecutiveFailures int // Number of consecutive failures before marking inactive + RecoveryDebounce time.Duration // Time to wait before allowing inactive->active recovery } // ActiveAgent represents an agent currently being monitored type ActiveAgent struct { - NodeID string - BaseURL string - LastStatus types.HealthStatus - LastChecked time.Time + NodeID string + BaseURL string + LastStatus types.HealthStatus + LastChecked time.Time + ConsecutiveFailures int // Track consecutive HTTP check failures + LastTransition time.Time // When the health status last changed } // HealthMonitor monitors the health of actively registered agent nodes @@ -37,6 +53,7 @@ type HealthMonitor struct { statusManager *StatusManager presence *PresenceManager stopCh chan struct{} + stopOnce sync.Once // Active agents registry - only agents currently running activeAgents map[string]*ActiveAgent @@ -49,9 +66,18 @@ type HealthMonitor struct { // NewHealthMonitor creates a new HTTP-first health monitor service func NewHealthMonitor(storage storage.StorageProvider, config HealthMonitorConfig, uiService *UIService, agentClient interfaces.AgentClient, statusManager *StatusManager, presence *PresenceManager) *HealthMonitor { - // Set default values - using efficient 10s intervals + // Set default values if config.CheckInterval == 0 { - config.CheckInterval = 10 * time.Second // HTTP health check every 10 seconds + config.CheckInterval = 10 * time.Second + } + if config.CheckTimeout == 0 { + config.CheckTimeout = 5 * time.Second + } + if config.ConsecutiveFailures == 0 { + config.ConsecutiveFailures = 3 // Require 3 failures before marking inactive + } + if config.RecoveryDebounce == 0 { + config.RecoveryDebounce = 5 * time.Second // Reduced from 30s for faster recovery } return &HealthMonitor{ @@ -77,10 +103,11 @@ func (hm *HealthMonitor) RegisterAgent(nodeID, baseURL string) { seenAt := time.Now() hm.activeAgents[nodeID] = &ActiveAgent{ - NodeID: nodeID, - BaseURL: baseURL, - LastStatus: types.HealthStatusUnknown, - LastChecked: seenAt, + NodeID: nodeID, + BaseURL: baseURL, + LastStatus: types.HealthStatusUnknown, + LastChecked: seenAt, + LastTransition: seenAt, // Initialize so debounce checks have a valid baseline } if hm.presence != nil { @@ -201,181 +228,225 @@ func (hm *HealthMonitor) Start() { } } -// Stop stops the health monitoring process +// Stop stops the health monitoring process. Safe to call multiple times. func (hm *HealthMonitor) Stop() { - close(hm.stopCh) + hm.stopOnce.Do(func() { + close(hm.stopCh) + }) } // checkActiveAgents performs HTTP health checks on all actively registered agents func (hm *HealthMonitor) checkActiveAgents() { hm.agentsMutex.RLock() - agents := make([]*ActiveAgent, 0, len(hm.activeAgents)) - for _, agent := range hm.activeAgents { - agents = append(agents, agent) + nodeIDs := make([]string, 0, len(hm.activeAgents)) + for id := range hm.activeAgents { + nodeIDs = append(nodeIDs, id) } hm.agentsMutex.RUnlock() - if len(agents) == 0 { + if len(nodeIDs) == 0 { logger.Logger.Debug().Msg("🏥 No active agents to monitor") return } - logger.Logger.Debug().Msgf("🏥 Checking health of %d active agents via HTTP", len(agents)) + logger.Logger.Debug().Msgf("🏥 Checking health of %d active agents via HTTP", len(nodeIDs)) - for _, agent := range agents { - hm.checkAgentHealth(agent) + for _, nodeID := range nodeIDs { + hm.checkAgentHealth(nodeID) } } -// checkAgentHealth performs HTTP health check for a single agent -func (hm *HealthMonitor) checkAgentHealth(agent *ActiveAgent) { +// checkAgentHealth performs HTTP health check for a single agent identified by nodeID. +// Uses consecutive failure tracking to prevent flapping from transient network issues. +// Accepts nodeID rather than *ActiveAgent to avoid holding stale pointers across the +// HTTP call boundary — the canonical state is always re-read from hm.activeAgents. +func (hm *HealthMonitor) checkAgentHealth(nodeID string) { // Early check: ensure agent is still in active registry before making HTTP call hm.agentsMutex.RLock() - _, exists := hm.activeAgents[agent.NodeID] + _, exists := hm.activeAgents[nodeID] hm.agentsMutex.RUnlock() if !exists { - logger.Logger.Debug().Msgf("🏥 Skipping health check for %s - agent no longer in active registry", agent.NodeID) + logger.Logger.Debug().Msgf("🏥 Skipping health check for %s - agent no longer in active registry", nodeID) return } - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), hm.config.CheckTimeout) defer cancel() - // Use HTTP /status endpoint as single source of truth - status, err := hm.agentClient.GetAgentStatus(ctx, agent.NodeID) + // Perform HTTP health check + status, err := hm.agentClient.GetAgentStatus(ctx, nodeID) - var newStatus types.HealthStatus + var checkPassed bool if err != nil { - // HTTP request failed - agent is offline - newStatus = types.HealthStatusInactive - logger.Logger.Debug().Msgf("🏥 Agent %s HTTP check failed: %v", agent.NodeID, err) + checkPassed = false + logger.Logger.Debug().Msgf("🏥 Agent %s HTTP check failed: %v", nodeID, err) } else if status.Status == "running" { - // FIXED: More conservative health detection to prevent false positives - // Only consider agent active if it's actually running and responsive - newStatus = types.HealthStatusActive - logger.Logger.Debug().Msgf("🏥 Agent %s HTTP check successful: %s", agent.NodeID, status.Status) + checkPassed = true + logger.Logger.Debug().Msgf("🏥 Agent %s HTTP check successful: %s", nodeID, status.Status) } else { - // Agent responded but not running - newStatus = types.HealthStatusInactive - logger.Logger.Debug().Msgf("🏥 Agent %s HTTP check shows not running: %s", agent.NodeID, status.Status) + checkPassed = false + logger.Logger.Debug().Msgf("🏥 Agent %s HTTP check shows not running: %s", nodeID, status.Status) } - // Update agent's last checked time + // Update agent state with consecutive failure tracking. + // Re-fetch from the map since the agent may have been unregistered/re-registered + // during the HTTP call above. hm.agentsMutex.Lock() - if activeAgent, exists := hm.activeAgents[agent.NodeID]; exists { - activeAgent.LastChecked = time.Now() - statusChanged := activeAgent.LastStatus != newStatus - activeAgent.LastStatus = newStatus + activeAgent, exists := hm.activeAgents[nodeID] + if !exists { hm.agentsMutex.Unlock() + return + } - // Only update database and broadcast events if status actually changed - if statusChanged { - logger.Logger.Debug().Msgf("🔄 Agent %s status changed to %s via HTTP check", agent.NodeID, newStatus) - - // FIXED: Add debouncing to prevent rapid oscillation - // If agent just changed to inactive, wait before allowing it to become active again - if newStatus == types.HealthStatusInactive { - // Mark agent as recently failed to prevent immediate reactivation - activeAgent.LastChecked = time.Now() - } else if newStatus == types.HealthStatusActive && activeAgent.LastStatus == types.HealthStatusInactive { - // If agent was recently inactive, require a longer period of stability - if time.Since(activeAgent.LastChecked) < 30*time.Second { - logger.Logger.Debug().Msgf("🏥 Agent %s status change to active too soon after inactive, skipping", agent.NodeID) - return - } + activeAgent.LastChecked = time.Now() + + if checkPassed { + // SUCCESS: Reset failure counter + previousFailures := activeAgent.ConsecutiveFailures + activeAgent.ConsecutiveFailures = 0 + + if previousFailures > 0 { + logger.Logger.Debug().Msgf("🏥 Agent %s check passed, reset failure counter from %d", nodeID, previousFailures) + } + + if activeAgent.LastStatus == types.HealthStatusInactive { + // Recovery from inactive: apply debounce + if time.Since(activeAgent.LastTransition) < hm.config.RecoveryDebounce { + hm.agentsMutex.Unlock() + logger.Logger.Debug().Msgf("🏥 Agent %s recovery debounce active, waiting", nodeID) + return } + activeAgent.LastStatus = types.HealthStatusActive + activeAgent.LastTransition = time.Now() + hm.agentsMutex.Unlock() + logger.Logger.Info().Msgf("✅ Agent %s recovered to active", nodeID) + hm.markAgentActive(nodeID) + return + } else if activeAgent.LastStatus != types.HealthStatusActive { + // First time becoming active (e.g. from unknown) + activeAgent.LastStatus = types.HealthStatusActive + activeAgent.LastTransition = time.Now() + hm.agentsMutex.Unlock() + hm.markAgentActive(nodeID) + return + } + // Already active, no status change needed — still refresh MCP health + hm.agentsMutex.Unlock() + hm.checkMCPHealthForNode(nodeID) + } else { + // FAILURE: Increment consecutive failure counter (capped to prevent unbounded growth) + if activeAgent.ConsecutiveFailures < hm.config.ConsecutiveFailures+1 { + activeAgent.ConsecutiveFailures++ + } - // Update through unified status system if available - if hm.statusManager != nil { - // Determine the new agent state based on health status - var newState types.AgentState - var healthScore int - - switch newStatus { - case types.HealthStatusActive: - newState = types.AgentStateActive - // FIXED: More conservative health score to prevent oscillation - // Only give high score if agent is consistently responsive - healthScore = 75 // Moderate health from HTTP check - case types.HealthStatusInactive: - newState = types.AgentStateInactive - healthScore = 0 // No health - default: - newState = types.AgentStateInactive - healthScore = 0 - } + logger.Logger.Debug().Msgf("🏥 Agent %s failure %d/%d", + nodeID, activeAgent.ConsecutiveFailures, hm.config.ConsecutiveFailures) + + // Only mark inactive after reaching the consecutive failure threshold + if activeAgent.ConsecutiveFailures >= hm.config.ConsecutiveFailures { + if activeAgent.LastStatus != types.HealthStatusInactive { + activeAgent.LastStatus = types.HealthStatusInactive + activeAgent.LastTransition = time.Now() + failCount := activeAgent.ConsecutiveFailures + hm.agentsMutex.Unlock() + logger.Logger.Warn().Msgf("Agent %s marked inactive after %d consecutive failures", nodeID, failCount) + hm.markAgentInactive(nodeID, failCount) + return + } + } + hm.agentsMutex.Unlock() + } +} - // Create status update - update := &types.AgentStatusUpdate{ - State: &newState, - HealthScore: &healthScore, - Source: types.StatusSourceHealthCheck, - Reason: "HTTP health check result", - } +// markAgentActive marks an agent as active through the unified status system +func (hm *HealthMonitor) markAgentActive(nodeID string) { + ctx := context.Background() + + if hm.statusManager != nil { + activeState := types.AgentStateActive + healthScore := healthScoreActive + update := &types.AgentStatusUpdate{ + State: &activeState, + HealthScore: &healthScore, + Source: types.StatusSourceHealthCheck, + Reason: "HTTP health check passed", + } - // Update through unified system - ctx := context.Background() - if err := hm.statusManager.UpdateAgentStatus(ctx, agent.NodeID, update); err != nil { - logger.Logger.Error().Err(err).Msgf("❌ Failed to update unified status for agent %s", agent.NodeID) - // Fallback to legacy update - if err := hm.storage.UpdateAgentHealth(ctx, agent.NodeID, newStatus); err != nil { - logger.Logger.Error().Err(err).Msgf("❌ Failed to update health status for agent %s", agent.NodeID) - } - } else if hm.presence != nil && newStatus == types.HealthStatusActive { - hm.presence.Touch(agent.NodeID, time.Now()) - } + if err := hm.statusManager.UpdateAgentStatus(ctx, nodeID, update); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to mark agent %s active via status manager", nodeID) + // Fallback to direct storage update + if err := hm.storage.UpdateAgentHealth(ctx, nodeID, types.HealthStatusActive); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to update health status for agent %s", nodeID) + } + return + } - // Check MCP health for active agents - if newStatus == types.HealthStatusActive { - hm.checkMCPHealthForNode(agent.NodeID) - } - } else { - // Fallback to legacy system for backward compatibility - ctx := context.Background() - if err := hm.storage.UpdateAgentHealth(ctx, agent.NodeID, newStatus); err != nil { - logger.Logger.Error().Err(err).Msgf("❌ Failed to update health status for agent %s", agent.NodeID) - return - } + if hm.presence != nil { + hm.presence.Touch(nodeID, time.Now()) + } - // Also update lifecycle status for consistency - var lifecycleStatus types.AgentLifecycleStatus - if newStatus == types.HealthStatusActive { - lifecycleStatus = types.AgentStatusReady - } else { - lifecycleStatus = types.AgentStatusOffline - } - if err := hm.storage.UpdateAgentLifecycleStatus(ctx, agent.NodeID, lifecycleStatus); err != nil { - logger.Logger.Error().Err(err).Msgf("❌ Failed to update lifecycle status for agent %s", agent.NodeID) - } + // Check MCP health for active agents + hm.checkMCPHealthForNode(nodeID) + } else { + // Legacy fallback + if err := hm.storage.UpdateAgentHealth(ctx, nodeID, types.HealthStatusActive); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to update health status for agent %s", nodeID) + return + } + if err := hm.storage.UpdateAgentLifecycleStatus(ctx, nodeID, types.AgentStatusReady); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to update lifecycle status for agent %s", nodeID) + } + if updatedAgent, err := hm.storage.GetAgent(ctx, nodeID); err == nil { + events.PublishNodeOnline(nodeID, updatedAgent) + if hm.presence != nil { + hm.presence.Touch(nodeID, time.Now()) + } + events.PublishNodeHealthChanged(nodeID, string(types.HealthStatusActive), updatedAgent) + if hm.uiService != nil { + hm.uiService.OnNodeStatusChanged(updatedAgent) + } + } + hm.checkMCPHealthForNode(nodeID) + } +} - // Broadcast status change events (legacy) - if updatedAgent, err := hm.storage.GetAgent(ctx, agent.NodeID); err == nil { - // Broadcast health-specific events - if newStatus == types.HealthStatusActive { - events.PublishNodeOnline(agent.NodeID, updatedAgent) - if hm.presence != nil { - hm.presence.Touch(agent.NodeID, time.Now()) - } - } else { - events.PublishNodeOffline(agent.NodeID, updatedAgent) - } - events.PublishNodeHealthChanged(agent.NodeID, string(newStatus), updatedAgent) - - // Send to UI service - if hm.uiService != nil { - hm.uiService.OnNodeStatusChanged(updatedAgent) - } - - // Check MCP health for active agents - if newStatus == types.HealthStatusActive { - hm.checkMCPHealthForNode(agent.NodeID) - } - } +// markAgentInactive marks an agent as inactive through the unified status system +func (hm *HealthMonitor) markAgentInactive(nodeID string, failCount int) { + ctx := context.Background() + + if hm.statusManager != nil { + inactiveState := types.AgentStateInactive + healthScore := healthScoreInactive + update := &types.AgentStatusUpdate{ + State: &inactiveState, + HealthScore: &healthScore, + Source: types.StatusSourceHealthCheck, + Reason: fmt.Sprintf("%d consecutive health check failures", failCount), + } + + if err := hm.statusManager.UpdateAgentStatus(ctx, nodeID, update); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to mark agent %s inactive via status manager", nodeID) + if err := hm.storage.UpdateAgentHealth(ctx, nodeID, types.HealthStatusInactive); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to update health status for agent %s", nodeID) } } } else { - hm.agentsMutex.Unlock() + // Legacy fallback + if err := hm.storage.UpdateAgentHealth(ctx, nodeID, types.HealthStatusInactive); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to update health status for agent %s", nodeID) + return + } + if err := hm.storage.UpdateAgentLifecycleStatus(ctx, nodeID, types.AgentStatusOffline); err != nil { + logger.Logger.Error().Err(err).Msgf("❌ Failed to update lifecycle status for agent %s", nodeID) + } + if updatedAgent, err := hm.storage.GetAgent(ctx, nodeID); err == nil { + events.PublishNodeOffline(nodeID, updatedAgent) + events.PublishNodeHealthChanged(nodeID, string(types.HealthStatusInactive), updatedAgent) + if hm.uiService != nil { + hm.uiService.OnNodeStatusChanged(updatedAgent) + } + } } } diff --git a/control-plane/internal/services/health_monitor_test.go b/control-plane/internal/services/health_monitor_test.go index db1351027..0ee9ef34d 100644 --- a/control-plane/internal/services/health_monitor_test.go +++ b/control-plane/internal/services/health_monitor_test.go @@ -305,11 +305,7 @@ func TestHealthMonitor_CheckAgentHealth_Healthy(t *testing.T) { mockClient.setStatusResponse(nodeID, "running") // Perform health check - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() - - hm.checkAgentHealth(activeAgent) + hm.checkAgentHealth(nodeID) // Wait a bit for async updates time.Sleep(100 * time.Millisecond) @@ -344,22 +340,20 @@ func TestHealthMonitor_CheckAgentHealth_Inactive(t *testing.T) { // Set mock to return error (simulating agent offline) mockClient.setStatusError(nodeID, errors.New("connection refused")) - // Perform health check - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() - - hm.checkAgentHealth(activeAgent) + // Consecutive failures required (default: 3) before marking inactive + for i := 0; i < hm.config.ConsecutiveFailures; i++ { + hm.checkAgentHealth(nodeID) + } // Wait a bit for async updates time.Sleep(100 * time.Millisecond) - // Verify status was updated to inactive + // Verify status was updated to inactive after consecutive failures hm.agentsMutex.RLock() updatedAgent := hm.activeAgents[nodeID] hm.agentsMutex.RUnlock() - assert.Equal(t, types.HealthStatusInactive, updatedAgent.LastStatus, "Agent should be marked as inactive") + assert.Equal(t, types.HealthStatusInactive, updatedAgent.LastStatus, "Agent should be marked as inactive after consecutive failures") } func TestHealthMonitor_CheckAgentHealth_NotRunning(t *testing.T) { @@ -383,17 +377,16 @@ func TestHealthMonitor_CheckAgentHealth_NotRunning(t *testing.T) { // Set mock to return non-running status mockClient.setStatusResponse(nodeID, "stopped") - // Perform health check - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() + // Consecutive failures required before marking inactive - hm.checkAgentHealth(activeAgent) + for i := 0; i < hm.config.ConsecutiveFailures; i++ { + hm.checkAgentHealth(nodeID) + } // Wait a bit for async updates time.Sleep(100 * time.Millisecond) - // Verify status was updated to inactive + // Verify status was updated to inactive after consecutive failures hm.agentsMutex.RLock() updatedAgent := hm.activeAgents[nodeID] hm.agentsMutex.RUnlock() @@ -402,53 +395,64 @@ func TestHealthMonitor_CheckAgentHealth_NotRunning(t *testing.T) { } func TestHealthMonitor_CheckAgentHealth_StatusTransitions(t *testing.T) { - hm, provider, mockClient, _, _ := setupHealthMonitorTest(t) - ctx := context.Background() + // Use short debounce for test speed + provider, ctx := setupTestStorage(t) + statusManager := NewStatusManager(provider, StatusManagerConfig{ReconcileInterval: 30 * time.Second}, nil, nil) + presenceConfig := PresenceManagerConfig{HeartbeatTTL: 5 * time.Second, SweepInterval: 1 * time.Second, HardEvictTTL: 10 * time.Second} + presenceManager := NewPresenceManager(statusManager, presenceConfig) + mockClient := newMockAgentClient() + + config := HealthMonitorConfig{ + CheckInterval: 100 * time.Millisecond, + ConsecutiveFailures: 2, // Use 2 for faster test + RecoveryDebounce: 200 * time.Millisecond, + } + hm := NewHealthMonitor(provider, config, nil, mockClient, statusManager, presenceManager) + + t.Cleanup(func() { + hm.Stop() + presenceManager.Stop() + _ = provider.Close(ctx) + }) nodeID := "test-agent-transitions" baseURL := "http://localhost:8001" - // Register agent in storage - agent := &types.AgentNode{ - ID: nodeID, - BaseURL: baseURL, - } + agent := &types.AgentNode{ID: nodeID, BaseURL: baseURL} err := provider.RegisterAgent(ctx, agent) require.NoError(t, err) - // Register in health monitor hm.RegisterAgent(nodeID, baseURL) - // Get active agent - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() // Test transition: Unknown -> Active mockClient.setStatusResponse(nodeID, "running") - hm.checkAgentHealth(activeAgent) + hm.checkAgentHealth(nodeID) time.Sleep(100 * time.Millisecond) hm.agentsMutex.RLock() assert.Equal(t, types.HealthStatusActive, hm.activeAgents[nodeID].LastStatus) hm.agentsMutex.RUnlock() - // Test transition: Active -> Inactive + // Test transition: Active -> Inactive (requires consecutive failures) mockClient.setStatusError(nodeID, errors.New("connection refused")) - hm.checkAgentHealth(activeAgent) + for i := 0; i < config.ConsecutiveFailures; i++ { + hm.checkAgentHealth(nodeID) + } time.Sleep(100 * time.Millisecond) hm.agentsMutex.RLock() assert.Equal(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus) hm.agentsMutex.RUnlock() - // Test transition: Inactive -> Active (should work after debouncing period) - // Wait for debounce period - time.Sleep(31 * time.Second) + // Test transition: Inactive -> Active (after debounce period) + time.Sleep(300 * time.Millisecond) // Wait past 200ms debounce mockClient.setStatusResponse(nodeID, "running") - mockClient.statusErrors = make(map[string]error) // Clear errors - hm.checkAgentHealth(activeAgent) + mockClient.mu.Lock() + delete(mockClient.statusErrors, nodeID) + mockClient.mu.Unlock() + hm.checkAgentHealth(nodeID) time.Sleep(100 * time.Millisecond) hm.agentsMutex.RLock() @@ -460,18 +464,11 @@ func TestHealthMonitor_CheckAgentHealth_UnregisteredAgent(t *testing.T) { hm, _, mockClient, _, _ := setupHealthMonitorTest(t) nodeID := "test-agent-unregistered" - baseURL := "http://localhost:8001" - - // Create agent but don't register it - activeAgent := &ActiveAgent{ - NodeID: nodeID, - BaseURL: baseURL, - } mockClient.setStatusResponse(nodeID, "running") - // Should skip check for unregistered agent - hm.checkAgentHealth(activeAgent) + // Should skip check for unregistered agent (not in active registry) + hm.checkAgentHealth(nodeID) // Verify no status calls were made (agent was not in registry) assert.Equal(t, 0, mockClient.getStatusCallCountFor(nodeID)) @@ -510,11 +507,8 @@ func TestHealthMonitor_MCP_CheckMCPHealth(t *testing.T) { mockClient.setStatusResponse(nodeID, "running") // Perform health check (should trigger MCP check for active agents) - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() - hm.checkAgentHealth(activeAgent) + hm.checkAgentHealth(nodeID) time.Sleep(200 * time.Millisecond) // Verify MCP health was checked @@ -562,11 +556,8 @@ func TestHealthMonitor_MCP_HealthChange(t *testing.T) { } mockClient.setMCPHealthResponse(nodeID, mcpResponse1) - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() - hm.checkAgentHealth(activeAgent) + hm.checkAgentHealth(nodeID) time.Sleep(200 * time.Millisecond) // Change MCP health @@ -581,7 +572,7 @@ func TestHealthMonitor_MCP_HealthChange(t *testing.T) { mockClient.setMCPHealthResponse(nodeID, mcpResponse2) // Second health check - hm.checkAgentHealth(activeAgent) + hm.checkAgentHealth(nodeID) time.Sleep(200 * time.Millisecond) // Verify MCP health was updated @@ -624,12 +615,9 @@ func TestHealthMonitor_MCP_NoChange(t *testing.T) { } mockClient.setMCPHealthResponse(nodeID, mcpResponse) - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() // First check - hm.checkAgentHealth(activeAgent) + hm.checkAgentHealth(nodeID) time.Sleep(200 * time.Millisecond) // Verify hasMCPHealthChanged returns false for same data @@ -664,11 +652,8 @@ func TestHealthMonitor_MCP_InactiveAgent(t *testing.T) { // Set agent as inactive mockClient.setStatusError(nodeID, errors.New("connection refused")) - hm.agentsMutex.RLock() - activeAgent := hm.activeAgents[nodeID] - hm.agentsMutex.RUnlock() - hm.checkAgentHealth(activeAgent) + hm.checkAgentHealth(nodeID) time.Sleep(200 * time.Millisecond) // MCP health should NOT be checked for inactive agents @@ -903,6 +888,9 @@ func TestHealthMonitor_RecoverFromDatabase_WithNodes(t *testing.T) { assert.True(t, agent2Exists, "agent-2 should be registered") assert.False(t, agent3Exists, "agent-3 should not be registered (no BaseURL)") + // Wait for async health checks to complete (RecoverFromDatabase runs them in a goroutine) + time.Sleep(200 * time.Millisecond) + // Verify health checks were performed assert.GreaterOrEqual(t, mockClient.getStatusCallCountFor("agent-1"), 1, "Should have checked agent-1 health") assert.GreaterOrEqual(t, mockClient.getStatusCallCountFor("agent-2"), 1, "Should have checked agent-2 health") @@ -929,13 +917,28 @@ func TestHealthMonitor_RecoverFromDatabase_MarksUnreachableNodesInactive(t *test err = hm.RecoverFromDatabase(ctx) require.NoError(t, err) - // Verify agent was registered + // Wait for async health checks to complete + time.Sleep(200 * time.Millisecond) + + // After recovery, agent is registered but a single check won't mark inactive + // (consecutive failures required). Run additional checks to reach the threshold. hm.agentsMutex.RLock() activeAgent, exists := hm.activeAgents["unreachable-agent"] hm.agentsMutex.RUnlock() assert.True(t, exists, "Agent should be registered") - assert.Equal(t, types.HealthStatusInactive, activeAgent.LastStatus, "Agent should be marked inactive after failed health check") + + // Run enough checks to trigger consecutive failure threshold + for i := 0; i < hm.config.ConsecutiveFailures; i++ { + hm.checkAgentHealth("unreachable-agent") + } + time.Sleep(100 * time.Millisecond) + + hm.agentsMutex.RLock() + activeAgent = hm.activeAgents["unreachable-agent"] + hm.agentsMutex.RUnlock() + + assert.Equal(t, types.HealthStatusInactive, activeAgent.LastStatus, "Agent should be marked inactive after consecutive failed health checks") } func TestHealthMonitor_RecoverFromDatabase_MarksReachableNodesActive(t *testing.T) { @@ -959,6 +962,9 @@ func TestHealthMonitor_RecoverFromDatabase_MarksReachableNodesActive(t *testing. err = hm.RecoverFromDatabase(ctx) require.NoError(t, err) + // Wait for async health checks to complete (RecoverFromDatabase runs them in a goroutine) + time.Sleep(200 * time.Millisecond) + // Verify agent was registered and marked active hm.agentsMutex.RLock() activeAgent, exists := hm.activeAgents["reachable-agent"] @@ -967,3 +973,652 @@ func TestHealthMonitor_RecoverFromDatabase_MarksReachableNodesActive(t *testing. assert.True(t, exists, "Agent should be registered") assert.Equal(t, types.HealthStatusActive, activeAgent.LastStatus, "Agent should be marked active after successful health check") } + +// --- Consecutive failure tests --- + +func TestHealthMonitor_ConsecutiveFailures_SingleFailureKeepsActive(t *testing.T) { + hm, provider, mockClient, _, _ := setupHealthMonitorTest(t) + ctx := context.Background() + + nodeID := "test-single-failure" + baseURL := "http://localhost:8001" + + agent := &types.AgentNode{ID: nodeID, BaseURL: baseURL} + err := provider.RegisterAgent(ctx, agent) + require.NoError(t, err) + + hm.RegisterAgent(nodeID, baseURL) + + // First make agent active + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + time.Sleep(100 * time.Millisecond) + + hm.agentsMutex.RLock() + assert.Equal(t, types.HealthStatusActive, hm.activeAgents[nodeID].LastStatus, "Agent should be active initially") + hm.agentsMutex.RUnlock() + + // Now simulate one failure — should NOT mark inactive + mockClient.setStatusError(nodeID, errors.New("connection refused")) + hm.checkAgentHealth(nodeID) + time.Sleep(100 * time.Millisecond) + + hm.agentsMutex.RLock() + assert.NotEqual(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus, + "Single failure should NOT mark agent inactive (consecutive failures required)") + assert.Equal(t, 1, hm.activeAgents[nodeID].ConsecutiveFailures, "Should have 1 consecutive failure") + hm.agentsMutex.RUnlock() +} + +func TestHealthMonitor_ConsecutiveFailures_ThreeFailuresMarksInactive(t *testing.T) { + hm, provider, mockClient, _, _ := setupHealthMonitorTest(t) + ctx := context.Background() + + nodeID := "test-three-failures" + baseURL := "http://localhost:8001" + + agent := &types.AgentNode{ID: nodeID, BaseURL: baseURL} + err := provider.RegisterAgent(ctx, agent) + require.NoError(t, err) + + hm.RegisterAgent(nodeID, baseURL) + + // Make agent active first + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + time.Sleep(100 * time.Millisecond) + + // Now fail 3 times (default ConsecutiveFailures threshold) + mockClient.setStatusError(nodeID, errors.New("connection refused")) + + for i := 0; i < hm.config.ConsecutiveFailures; i++ { + hm.checkAgentHealth(nodeID) + } + time.Sleep(100 * time.Millisecond) + + hm.agentsMutex.RLock() + assert.Equal(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus, + "Agent should be inactive after %d consecutive failures", hm.config.ConsecutiveFailures) + hm.agentsMutex.RUnlock() +} + +func TestHealthMonitor_ConsecutiveFailures_SuccessResetsCounter(t *testing.T) { + hm, provider, mockClient, _, _ := setupHealthMonitorTest(t) + ctx := context.Background() + + nodeID := "test-success-resets" + baseURL := "http://localhost:8001" + + agent := &types.AgentNode{ID: nodeID, BaseURL: baseURL} + err := provider.RegisterAgent(ctx, agent) + require.NoError(t, err) + + hm.RegisterAgent(nodeID, baseURL) + + // Make active first + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + time.Sleep(100 * time.Millisecond) + + // Fail twice (2 out of 3) + mockClient.setStatusError(nodeID, errors.New("connection refused")) + hm.checkAgentHealth(nodeID) + hm.checkAgentHealth(nodeID) + + hm.agentsMutex.RLock() + assert.Equal(t, 2, hm.activeAgents[nodeID].ConsecutiveFailures, "Should have 2 failures") + hm.agentsMutex.RUnlock() + + // Success resets the counter + mockClient.mu.Lock() + delete(mockClient.statusErrors, nodeID) + mockClient.mu.Unlock() + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + + hm.agentsMutex.RLock() + assert.Equal(t, 0, hm.activeAgents[nodeID].ConsecutiveFailures, "Success should reset failure counter") + assert.Equal(t, types.HealthStatusActive, hm.activeAgents[nodeID].LastStatus, "Agent should still be active") + hm.agentsMutex.RUnlock() + + // Fail twice more — should still NOT be inactive (counter was reset) + mockClient.setStatusError(nodeID, errors.New("connection refused")) + hm.checkAgentHealth(nodeID) + hm.checkAgentHealth(nodeID) + + hm.agentsMutex.RLock() + assert.NotEqual(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus, + "Agent should NOT be inactive — counter was reset by success") + hm.agentsMutex.RUnlock() +} + +func TestHealthMonitor_RecoveryDebounce_BlocksTooFastRecovery(t *testing.T) { + // Use a short but measurable debounce for testing + provider, ctx := setupTestStorage(t) + statusManager := NewStatusManager(provider, StatusManagerConfig{ReconcileInterval: 30 * time.Second}, nil, nil) + presenceConfig := PresenceManagerConfig{HeartbeatTTL: 5 * time.Second, SweepInterval: 1 * time.Second, HardEvictTTL: 10 * time.Second} + presenceManager := NewPresenceManager(statusManager, presenceConfig) + mockClient := newMockAgentClient() + + config := HealthMonitorConfig{ + CheckInterval: 100 * time.Millisecond, + ConsecutiveFailures: 2, + RecoveryDebounce: 500 * time.Millisecond, + } + hm := NewHealthMonitor(provider, config, nil, mockClient, statusManager, presenceManager) + + t.Cleanup(func() { + hm.Stop() + presenceManager.Stop() + _ = provider.Close(ctx) + }) + + nodeID := "test-debounce" + baseURL := "http://localhost:8001" + + agent := &types.AgentNode{ID: nodeID, BaseURL: baseURL} + err := provider.RegisterAgent(ctx, agent) + require.NoError(t, err) + hm.RegisterAgent(nodeID, baseURL) + + // Make active, then inactive + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + time.Sleep(50 * time.Millisecond) + + mockClient.setStatusError(nodeID, errors.New("connection refused")) + for i := 0; i < config.ConsecutiveFailures; i++ { + hm.checkAgentHealth(nodeID) + } + time.Sleep(50 * time.Millisecond) + + hm.agentsMutex.RLock() + require.Equal(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus) + hm.agentsMutex.RUnlock() + + // Immediately try to recover — should be blocked by debounce + mockClient.mu.Lock() + delete(mockClient.statusErrors, nodeID) + mockClient.mu.Unlock() + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + time.Sleep(50 * time.Millisecond) + + hm.agentsMutex.RLock() + assert.Equal(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus, + "Recovery should be blocked by debounce") + hm.agentsMutex.RUnlock() + + // Wait past debounce, then try again + time.Sleep(600 * time.Millisecond) + hm.checkAgentHealth(nodeID) + time.Sleep(50 * time.Millisecond) + + hm.agentsMutex.RLock() + assert.Equal(t, types.HealthStatusActive, hm.activeAgents[nodeID].LastStatus, + "Recovery should succeed after debounce period") + hm.agentsMutex.RUnlock() +} + +func TestHealthMonitor_Config_ConsecutiveFailuresConfigurable(t *testing.T) { + provider, ctx := setupTestStorage(t) + statusManager := NewStatusManager(provider, StatusManagerConfig{ReconcileInterval: 30 * time.Second}, nil, nil) + presenceConfig := PresenceManagerConfig{HeartbeatTTL: 5 * time.Second, SweepInterval: 1 * time.Second, HardEvictTTL: 10 * time.Second} + presenceManager := NewPresenceManager(statusManager, presenceConfig) + mockClient := newMockAgentClient() + + // Configure to require 5 consecutive failures + config := HealthMonitorConfig{ + CheckInterval: 100 * time.Millisecond, + ConsecutiveFailures: 5, + RecoveryDebounce: 100 * time.Millisecond, + } + hm := NewHealthMonitor(provider, config, nil, mockClient, statusManager, presenceManager) + + t.Cleanup(func() { + hm.Stop() + presenceManager.Stop() + _ = provider.Close(ctx) + }) + + nodeID := "test-config-failures" + baseURL := "http://localhost:8001" + + agent := &types.AgentNode{ID: nodeID, BaseURL: baseURL} + err := provider.RegisterAgent(ctx, agent) + require.NoError(t, err) + hm.RegisterAgent(nodeID, baseURL) + + // Make active first + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + time.Sleep(50 * time.Millisecond) + + // Fail 4 times — should NOT be inactive (threshold is 5) + mockClient.setStatusError(nodeID, errors.New("connection refused")) + for i := 0; i < 4; i++ { + hm.checkAgentHealth(nodeID) + } + + hm.agentsMutex.RLock() + assert.NotEqual(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus, + "4 failures should not mark inactive when threshold is 5") + assert.Equal(t, 4, hm.activeAgents[nodeID].ConsecutiveFailures) + hm.agentsMutex.RUnlock() + + // 5th failure — NOW it should be inactive + hm.checkAgentHealth(nodeID) + time.Sleep(50 * time.Millisecond) + + hm.agentsMutex.RLock() + assert.Equal(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus, + "5th failure should mark inactive when threshold is 5") + hm.agentsMutex.RUnlock() +} + +// ============================================================================= +// INTEGRATION TESTS — All 3 services running concurrently with competing signals +// ============================================================================= +// +// These tests start HealthMonitor, StatusManager, and PresenceManager running +// concurrently (as they do in production) and simulate the exact scenario that +// causes flapping: health check failures happening while heartbeats are flowing. + +// TestIntegration_NoFlapping_HeartbeatsDuringTransientFailures is the critical +// integration test. It runs all 3 services concurrently and verifies that an +// agent NEVER flaps to inactive while heartbeats are being sent, even when +// HTTP health checks are intermittently failing. +func TestIntegration_NoFlapping_HeartbeatsDuringTransientFailures(t *testing.T) { + provider, ctx := setupTestStorage(t) + + // --- Wire services exactly as production (server.go) --- + mockClient := newMockAgentClient() + + // StatusManager with short reconciliation and the same mock client + // (so UpdateFromHeartbeat -> GetAgentStatus uses mock HTTP too) + statusConfig := StatusManagerConfig{ + ReconcileInterval: 200 * time.Millisecond, + StatusCacheTTL: 50 * time.Millisecond, // Short cache so state changes propagate fast + HeartbeatStaleThreshold: 2 * time.Second, + MaxTransitionTime: 1 * time.Second, + } + statusManager := NewStatusManager(provider, statusConfig, nil, mockClient) + + // PresenceManager with short TTLs + presenceConfig := PresenceManagerConfig{ + HeartbeatTTL: 2 * time.Second, + SweepInterval: 200 * time.Millisecond, + HardEvictTTL: 5 * time.Second, + } + presenceManager := NewPresenceManager(statusManager, presenceConfig) + + // HealthMonitor with fast checks but requiring 3 consecutive failures + hmConfig := HealthMonitorConfig{ + CheckInterval: 100 * time.Millisecond, + CheckTimeout: 2 * time.Second, + ConsecutiveFailures: 3, + RecoveryDebounce: 200 * time.Millisecond, + } + hm := NewHealthMonitor(provider, hmConfig, nil, mockClient, statusManager, presenceManager) + presenceManager.SetExpireCallback(hm.UnregisterAgent) + + // Cleanup + t.Cleanup(func() { + hm.Stop() + presenceManager.Stop() + statusManager.Stop() + _ = provider.Close(ctx) + }) + + // --- Register agent in storage --- + nodeID := "integration-agent-1" + node := &types.AgentNode{ + ID: nodeID, + TeamID: "team", + BaseURL: "http://localhost:9999", + Version: "1.0.0", + HealthStatus: types.HealthStatusActive, + LifecycleStatus: types.AgentStatusReady, + LastHeartbeat: time.Now(), + Reasoners: []types.ReasonerDefinition{}, + Skills: []types.SkillDefinition{}, + } + require.NoError(t, provider.RegisterAgent(ctx, node)) + + // Agent starts healthy + mockClient.setStatusResponse(nodeID, "running") + + // Register with health monitor and presence + hm.RegisterAgent(nodeID, "http://localhost:9999") + presenceManager.Touch(nodeID, time.Now()) + + // --- Start all 3 services concurrently (like production) --- + go hm.Start() + statusManager.Start() + presenceManager.Start() + + // Let services stabilize — agent should be active + time.Sleep(300 * time.Millisecond) + + // Verify agent is active before the test + snapshot, err := statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + require.NoError(t, err) + require.Equal(t, types.AgentStateActive, snapshot.State, + "Agent should be active before flapping test begins") + + // --- THE FLAPPING SCENARIO --- + // Health checks start failing (simulating network blip), but agent keeps + // sending heartbeats (proving it's alive). The agent should NEVER go inactive. + + // Track all status transitions to detect flapping + var statusHistory []types.AgentState + var historyMu sync.Mutex + + // Start heartbeat sender goroutine (simulates agent sending heartbeats every 100ms) + heartbeatDone := make(chan struct{}) + go func() { + defer close(heartbeatDone) + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + for i := 0; i < 30; i++ { // 30 heartbeats over ~3 seconds + <-ticker.C + readyStatus := types.AgentStatusReady + _ = statusManager.UpdateFromHeartbeat(ctx, nodeID, &readyStatus, nil) + presenceManager.Touch(nodeID, time.Now()) + + // Record current state + snap, err := statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + if err == nil { + historyMu.Lock() + statusHistory = append(statusHistory, snap.State) + historyMu.Unlock() + } + } + }() + + // Introduce intermittent health check failures after 200ms + time.Sleep(200 * time.Millisecond) + mockClient.setStatusError(nodeID, errors.New("connection timeout")) + + // Let it run with failing health checks + flowing heartbeats for 2 seconds + time.Sleep(2 * time.Second) + + // Restore health checks + mockClient.mu.Lock() + delete(mockClient.statusErrors, nodeID) + mockClient.mu.Unlock() + mockClient.setStatusResponse(nodeID, "running") + + // Wait for heartbeat goroutine to finish + <-heartbeatDone + + // Let services settle + time.Sleep(300 * time.Millisecond) + + // --- ASSERTIONS --- + + // 1. Agent should be active now + finalSnapshot, err := statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + require.NoError(t, err) + assert.Equal(t, types.AgentStateActive, finalSnapshot.State, + "Agent must be active — heartbeats were flowing the entire time") + + // 2. Agent should NEVER have been inactive during the test + historyMu.Lock() + defer historyMu.Unlock() + + inactiveCount := 0 + for _, state := range statusHistory { + if state == types.AgentStateInactive { + inactiveCount++ + } + } + assert.Zero(t, inactiveCount, + "Agent should NEVER have been inactive while heartbeats were flowing. "+ + "Got %d inactive readings out of %d samples. This is the flapping bug.", + inactiveCount, len(statusHistory)) + + // 3. Log the full history for debugging if test fails + if t.Failed() { + t.Logf("Status history (%d samples):", len(statusHistory)) + for i, state := range statusHistory { + t.Logf(" [%d] %s", i, state) + } + } +} + +// TestIntegration_ProperInactiveWhenHeartbeatsStop verifies that when an agent +// genuinely goes down (both heartbeats AND health checks stop), the system +// correctly marks it inactive after the consecutive failure threshold. +func TestIntegration_ProperInactiveWhenHeartbeatsStop(t *testing.T) { + provider, ctx := setupTestStorage(t) + + mockClient := newMockAgentClient() + + statusConfig := StatusManagerConfig{ + ReconcileInterval: 200 * time.Millisecond, + StatusCacheTTL: 50 * time.Millisecond, + HeartbeatStaleThreshold: 1 * time.Second, + MaxTransitionTime: 500 * time.Millisecond, + } + statusManager := NewStatusManager(provider, statusConfig, nil, mockClient) + + presenceConfig := PresenceManagerConfig{ + HeartbeatTTL: 1 * time.Second, + SweepInterval: 200 * time.Millisecond, + HardEvictTTL: 3 * time.Second, + } + presenceManager := NewPresenceManager(statusManager, presenceConfig) + + hmConfig := HealthMonitorConfig{ + CheckInterval: 100 * time.Millisecond, + CheckTimeout: 2 * time.Second, + ConsecutiveFailures: 3, + RecoveryDebounce: 200 * time.Millisecond, + } + hm := NewHealthMonitor(provider, hmConfig, nil, mockClient, statusManager, presenceManager) + presenceManager.SetExpireCallback(hm.UnregisterAgent) + + t.Cleanup(func() { + hm.Stop() + presenceManager.Stop() + statusManager.Stop() + _ = provider.Close(ctx) + }) + + // Register healthy agent + nodeID := "integration-agent-down" + node := &types.AgentNode{ + ID: nodeID, + TeamID: "team", + BaseURL: "http://localhost:9998", + Version: "1.0.0", + HealthStatus: types.HealthStatusActive, + LifecycleStatus: types.AgentStatusReady, + LastHeartbeat: time.Now(), + Reasoners: []types.ReasonerDefinition{}, + Skills: []types.SkillDefinition{}, + } + require.NoError(t, provider.RegisterAgent(ctx, node)) + + mockClient.setStatusResponse(nodeID, "running") + hm.RegisterAgent(nodeID, "http://localhost:9998") + presenceManager.Touch(nodeID, time.Now()) + + // Start all services + go hm.Start() + statusManager.Start() + presenceManager.Start() + + // Let agent stabilize as active + time.Sleep(300 * time.Millisecond) + + snapshot, err := statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + require.NoError(t, err) + require.Equal(t, types.AgentStateActive, snapshot.State) + + // --- Agent goes down: no more heartbeats, health checks fail --- + mockClient.setStatusError(nodeID, errors.New("connection refused")) + + // Wait for consecutive failures to trigger inactive marking + // 3 failures at 100ms interval = ~300ms + some processing time + time.Sleep(800 * time.Millisecond) + + // Agent should now be inactive + finalSnapshot, err := statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + require.NoError(t, err) + assert.Equal(t, types.AgentStateInactive, finalSnapshot.State, + "Agent should be inactive when both heartbeats and health checks have stopped") +} + +// TestIntegration_RecoveryAfterGenuineOutage verifies the full lifecycle: +// healthy -> outage (goes inactive) -> comes back (sends heartbeat) -> recovers to active. +func TestIntegration_RecoveryAfterGenuineOutage(t *testing.T) { + provider, ctx := setupTestStorage(t) + + mockClient := newMockAgentClient() + + statusConfig := StatusManagerConfig{ + ReconcileInterval: 200 * time.Millisecond, + StatusCacheTTL: 50 * time.Millisecond, + HeartbeatStaleThreshold: 1 * time.Second, + MaxTransitionTime: 500 * time.Millisecond, + } + statusManager := NewStatusManager(provider, statusConfig, nil, mockClient) + + presenceConfig := PresenceManagerConfig{ + HeartbeatTTL: 1 * time.Second, + SweepInterval: 200 * time.Millisecond, + HardEvictTTL: 3 * time.Second, + } + presenceManager := NewPresenceManager(statusManager, presenceConfig) + + hmConfig := HealthMonitorConfig{ + CheckInterval: 100 * time.Millisecond, + CheckTimeout: 2 * time.Second, + ConsecutiveFailures: 3, + RecoveryDebounce: 200 * time.Millisecond, + } + hm := NewHealthMonitor(provider, hmConfig, nil, mockClient, statusManager, presenceManager) + presenceManager.SetExpireCallback(hm.UnregisterAgent) + + t.Cleanup(func() { + hm.Stop() + presenceManager.Stop() + statusManager.Stop() + _ = provider.Close(ctx) + }) + + // Register healthy agent + nodeID := "integration-agent-recovery" + node := &types.AgentNode{ + ID: nodeID, + TeamID: "team", + BaseURL: "http://localhost:9997", + Version: "1.0.0", + HealthStatus: types.HealthStatusActive, + LifecycleStatus: types.AgentStatusReady, + LastHeartbeat: time.Now(), + Reasoners: []types.ReasonerDefinition{}, + Skills: []types.SkillDefinition{}, + } + require.NoError(t, provider.RegisterAgent(ctx, node)) + + mockClient.setStatusResponse(nodeID, "running") + hm.RegisterAgent(nodeID, "http://localhost:9997") + presenceManager.Touch(nodeID, time.Now()) + + // Start all services + go hm.Start() + statusManager.Start() + presenceManager.Start() + + // Phase 1: Agent is healthy + time.Sleep(300 * time.Millisecond) + snapshot, err := statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + require.NoError(t, err) + require.Equal(t, types.AgentStateActive, snapshot.State, "Phase 1: should be active") + + // Phase 2: Agent goes down (genuine outage) + mockClient.setStatusError(nodeID, errors.New("connection refused")) + time.Sleep(800 * time.Millisecond) + + snapshot, err = statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + require.NoError(t, err) + require.Equal(t, types.AgentStateInactive, snapshot.State, "Phase 2: should be inactive after outage") + + // Phase 3: Agent comes back — health checks pass and heartbeat sent + mockClient.mu.Lock() + delete(mockClient.statusErrors, nodeID) + mockClient.mu.Unlock() + mockClient.setStatusResponse(nodeID, "running") + + // Re-register with health monitor (agent would re-register on reconnect) + hm.RegisterAgent(nodeID, "http://localhost:9997") + presenceManager.Touch(nodeID, time.Now()) + + // Send a heartbeat to signal recovery + readyStatus := types.AgentStatusReady + err = statusManager.UpdateFromHeartbeat(ctx, nodeID, &readyStatus, nil) + require.NoError(t, err) + + // Wait for health check cycle + debounce + time.Sleep(600 * time.Millisecond) + + snapshot, err = statusManager.GetAgentStatusSnapshot(ctx, nodeID, nil) + require.NoError(t, err) + assert.Equal(t, types.AgentStateActive, snapshot.State, + "Phase 3: agent should recover to active after coming back") +} + +// TestHealthMonitor_Config_ConsecutiveFailuresOne verifies that setting +// ConsecutiveFailures=1 restores the old "single failure = instant inactive" +// behavior for operators who want aggressive failure detection. +func TestHealthMonitor_Config_ConsecutiveFailuresOne(t *testing.T) { + provider, ctx := setupTestStorage(t) + statusManager := NewStatusManager(provider, StatusManagerConfig{ReconcileInterval: 30 * time.Second}, nil, nil) + presenceConfig := PresenceManagerConfig{HeartbeatTTL: 5 * time.Second, SweepInterval: 1 * time.Second, HardEvictTTL: 10 * time.Second} + presenceManager := NewPresenceManager(statusManager, presenceConfig) + mockClient := newMockAgentClient() + + config := HealthMonitorConfig{ + CheckInterval: 100 * time.Millisecond, + ConsecutiveFailures: 1, // Single failure = instant inactive + RecoveryDebounce: 100 * time.Millisecond, + } + hm := NewHealthMonitor(provider, config, nil, mockClient, statusManager, presenceManager) + + t.Cleanup(func() { + hm.Stop() + presenceManager.Stop() + _ = provider.Close(ctx) + }) + + nodeID := "test-instant-fail" + baseURL := "http://localhost:8001" + + agent := &types.AgentNode{ID: nodeID, BaseURL: baseURL} + err := provider.RegisterAgent(ctx, agent) + require.NoError(t, err) + hm.RegisterAgent(nodeID, baseURL) + + // Make active first + mockClient.setStatusResponse(nodeID, "running") + hm.checkAgentHealth(nodeID) + time.Sleep(50 * time.Millisecond) + + hm.agentsMutex.RLock() + require.Equal(t, types.HealthStatusActive, hm.activeAgents[nodeID].LastStatus) + hm.agentsMutex.RUnlock() + + // Single failure should immediately mark inactive + mockClient.setStatusError(nodeID, errors.New("connection refused")) + hm.checkAgentHealth(nodeID) + time.Sleep(50 * time.Millisecond) + + hm.agentsMutex.RLock() + assert.Equal(t, types.HealthStatusInactive, hm.activeAgents[nodeID].LastStatus, + "With ConsecutiveFailures=1, a single failure should mark agent inactive immediately") + assert.Equal(t, 1, hm.activeAgents[nodeID].ConsecutiveFailures) + hm.agentsMutex.RUnlock() +} diff --git a/control-plane/internal/services/status_manager.go b/control-plane/internal/services/status_manager.go index afeba0018..ec99d6bce 100644 --- a/control-plane/internal/services/status_manager.go +++ b/control-plane/internal/services/status_manager.go @@ -15,9 +15,10 @@ import ( // StatusManagerConfig holds configuration for the status manager type StatusManagerConfig struct { - ReconcileInterval time.Duration // How often to reconcile status - StatusCacheTTL time.Duration // How long to cache status - MaxTransitionTime time.Duration // Max time for state transitions + ReconcileInterval time.Duration // How often to reconcile status + StatusCacheTTL time.Duration // How long to cache status + MaxTransitionTime time.Duration // Max time for state transitions + HeartbeatStaleThreshold time.Duration // How old a heartbeat must be before marking inactive (default 60s) } // StatusManager provides a single source of truth for agent status @@ -91,6 +92,9 @@ func NewStatusManager(storage storage.StorageProvider, config StatusManagerConfi if config.MaxTransitionTime == 0 { config.MaxTransitionTime = 2 * time.Minute } + if config.HeartbeatStaleThreshold == 0 { + config.HeartbeatStaleThreshold = 60 * time.Second + } return &StatusManager{ storage: storage, @@ -384,23 +388,12 @@ func (sm *StatusManager) UpdateFromHeartbeat(ctx context.Context, nodeID string, currentStatus = types.NewAgentStatus(types.AgentStateStarting, types.StatusSourceHeartbeat) } - // INTELLIGENT HEARTBEAT PROCESSING: - // If we recently performed a live health check that determined the agent is offline, - // don't override that with heartbeat data (which could be stale/delayed) - if currentStatus.Source == types.StatusSourceHealthCheck && - currentStatus.State == types.AgentStateInactive && - currentStatus.LastVerified != nil && - time.Since(*currentStatus.LastVerified) < 10*time.Second { - - logger.Logger.Debug(). - Str("node_id", nodeID). - Str("current_source", string(currentStatus.Source)). - Str("current_state", string(currentStatus.State)). - Msg("🚫 Ignoring heartbeat update - recent health check determined agent is offline") - - // Don't process this heartbeat as it conflicts with recent live health check - return nil - } + // HEARTBEAT PRIORITY: Heartbeats are the primary signal of agent liveness. + // If an agent is sending heartbeats, it is alive regardless of what HTTP + // health checks report. Health checks may fail due to transient network + // issues, but a heartbeat is direct proof of life from the agent itself. + // The health monitor requires consecutive failures before marking inactive, + // so there is no need to suppress heartbeats here. // Update from heartbeat currentStatus.UpdateFromHeartbeat(lifecycleStatus, mcpStatus) @@ -657,9 +650,9 @@ func (sm *StatusManager) performReconciliation() { // needsReconciliation checks if an agent needs status reconciliation func (sm *StatusManager) needsReconciliation(agent *types.AgentNode) bool { - // Check if last heartbeat is too old + // Check if last heartbeat is too old (uses configurable threshold, default 60s) timeSinceHeartbeat := time.Since(agent.LastHeartbeat) - if timeSinceHeartbeat > 30*time.Second && agent.HealthStatus == types.HealthStatusActive { + if timeSinceHeartbeat > sm.config.HeartbeatStaleThreshold && agent.HealthStatus == types.HealthStatusActive { return true } @@ -679,7 +672,7 @@ func (sm *StatusManager) reconcileAgentStatus(ctx context.Context, agent *types. var newHealthStatus types.HealthStatus var newLifecycleStatus types.AgentLifecycleStatus - if timeSinceHeartbeat > 30*time.Second { + if timeSinceHeartbeat > sm.config.HeartbeatStaleThreshold { newHealthStatus = types.HealthStatusInactive newLifecycleStatus = types.AgentStatusOffline } else { diff --git a/control-plane/internal/services/status_manager_test.go b/control-plane/internal/services/status_manager_test.go index 7d756efca..ed8ffc4e8 100644 --- a/control-plane/internal/services/status_manager_test.go +++ b/control-plane/internal/services/status_manager_test.go @@ -14,6 +14,7 @@ import ( "github.com/Agent-Field/agentfield/control-plane/internal/storage" "github.com/Agent-Field/agentfield/control-plane/pkg/types" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -487,3 +488,85 @@ func (h *testStatusEventHandler) OnStatusChanged(nodeID string, oldStatus, newSt h.onStatusChanged(nodeID, oldStatus, newStatus) } } + +// --- Heartbeat priority and reconciliation threshold tests --- + +func TestStatusManager_UpdateFromHeartbeat_NeverDropped(t *testing.T) { + provider, ctx := setupStatusManagerStorage(t) + registerTestAgent(t, provider, ctx, "node-heartbeat-priority") + + sm := NewStatusManager(provider, StatusManagerConfig{ + ReconcileInterval: 30 * time.Second, + StatusCacheTTL: 5 * time.Minute, + }, nil, nil) + + // First, mark the agent as inactive via a health check source + // (simulating what HealthMonitor would do) + inactiveState := types.AgentStateInactive + healthScore := 0 + update := &types.AgentStatusUpdate{ + State: &inactiveState, + HealthScore: &healthScore, + Source: types.StatusSourceHealthCheck, + Reason: "HTTP health check failed", + } + err := sm.UpdateAgentStatus(ctx, "node-heartbeat-priority", update) + require.NoError(t, err) + + // Verify agent is inactive + status, err := sm.GetAgentStatusSnapshot(ctx, "node-heartbeat-priority", nil) + require.NoError(t, err) + require.Equal(t, types.AgentStateInactive, status.State) + require.Equal(t, types.StatusSourceHealthCheck, status.Source) + + // Now send a heartbeat IMMEDIATELY (within what used to be the 10s drop window). + // Previously this heartbeat would be silently ignored. Now it MUST be processed. + readyStatus := types.AgentStatusReady + err = sm.UpdateFromHeartbeat(ctx, "node-heartbeat-priority", &readyStatus, nil) + require.NoError(t, err, "Heartbeat should never be dropped") + + // Verify the heartbeat was processed — agent should no longer be inactive + status, err = sm.GetAgentStatusSnapshot(ctx, "node-heartbeat-priority", nil) + require.NoError(t, err) + require.Equal(t, types.StatusSourceHeartbeat, status.Source, + "Source should be heartbeat, proving it was processed") + require.NotEqual(t, types.AgentStateInactive, status.State, + "Agent should not be inactive after receiving a heartbeat") +} + +func TestStatusManager_Reconciliation_UsesConfiguredThreshold(t *testing.T) { + provider, _ := setupStatusManagerStorage(t) + + // Create StatusManager with default 60s threshold + sm := NewStatusManager(provider, StatusManagerConfig{ + ReconcileInterval: 30 * time.Second, + HeartbeatStaleThreshold: 60 * time.Second, + }, nil, nil) + + // Agent with heartbeat 45 seconds ago — should NOT need reconciliation + recentAgent := &types.AgentNode{ + ID: "node-recent", + HealthStatus: types.HealthStatusActive, + LastHeartbeat: time.Now().Add(-45 * time.Second), + } + assert.False(t, sm.needsReconciliation(recentAgent), + "Agent with 45s-old heartbeat should NOT need reconciliation (threshold is 60s)") + + // Agent with heartbeat 65 seconds ago — SHOULD need reconciliation + staleAgent := &types.AgentNode{ + ID: "node-stale", + HealthStatus: types.HealthStatusActive, + LastHeartbeat: time.Now().Add(-65 * time.Second), + } + assert.True(t, sm.needsReconciliation(staleAgent), + "Agent with 65s-old heartbeat should need reconciliation (threshold is 60s)") + + // Agent already inactive — should NOT need reconciliation even if stale + inactiveAgent := &types.AgentNode{ + ID: "node-inactive", + HealthStatus: types.HealthStatusInactive, + LastHeartbeat: time.Now().Add(-120 * time.Second), + } + assert.False(t, sm.needsReconciliation(inactiveAgent), + "Already inactive agent should not need reconciliation") +} diff --git a/control-plane/web/client/src/pages/NodesPage.tsx b/control-plane/web/client/src/pages/NodesPage.tsx index d846e289a..9687b0625 100644 --- a/control-plane/web/client/src/pages/NodesPage.tsx +++ b/control-plane/web/client/src/pages/NodesPage.tsx @@ -456,6 +456,27 @@ export function NodesPage() { // eslint-disable-next-line react-hooks/exhaustive-deps }, []); // Only run once on mount + // Periodic light refresh to keep timestamps (last_heartbeat) fresh + // SSE events don't carry heartbeat timestamps, so we poll every 30s + useEffect(() => { + let active = true; + const interval = setInterval(async () => { + try { + const data = await getNodesSummary(); + if (active) { + setNodes(data.nodes); + setLastRefresh(new Date()); + } + } catch { + // Silent fail on background refresh — don't disrupt the UI + } + }, 30000); + return () => { + active = false; + clearInterval(interval); + }; + }, []); + // Handle bulk status refresh const handleBulkRefresh = ( status: AgentStatus | Record