diff --git a/control-plane/.env.example b/control-plane/.env.example index 80cc31704..1eaee66d7 100644 --- a/control-plane/.env.example +++ b/control-plane/.env.example @@ -8,6 +8,8 @@ AGENTFIELD_CONFIG_FILE=./config/agentfield.yaml # Agent restarts and orderly drains (YAML: agentfield.node_health.agent_*_grace). # AGENTFIELD_AGENT_RESTART_GRACE=15s # AGENTFIELD_AGENT_DRAIN_GRACE=60s +# Disable when replicas > 1 share one node id; the stale sweep remains the backstop. +# AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED=true # Database Configuration (for local mode) AGENTFIELD_STORAGE_LOCAL_DATABASE_PATH=./agentfield_local.db diff --git a/control-plane/internal/config/config.go b/control-plane/internal/config/config.go index 9debc284d..e864722b8 100644 --- a/control-plane/internal/config/config.go +++ b/control-plane/internal/config/config.go @@ -217,6 +217,20 @@ type NodeHealthConfig struct { // registers, allowing the departing process to finish accepted work. // 0 = default 60s. Set to a negative duration to disable deferred cleanup. AgentDrainGrace time.Duration `yaml:"agent_drain_grace" mapstructure:"agent_drain_grace"` + // AgentOrphanReapEnabled controls whether a replacement registration marks + // the departing instance's in-flight executions orphaned. Default true. + // Set false for deployments running replicas>1 behind one node id, where a + // sibling replica registering is indistinguishable from a replacement. A + // pointer preserves the distinction between omitted (default true) and an + // explicit false across YAML, Viper, database overlays, and programmatic use. + AgentOrphanReapEnabled *bool `yaml:"agent_orphan_reap_enabled" mapstructure:"agent_orphan_reap_enabled"` +} + +// EffectiveAgentOrphanReapEnabled applies the documented zero-value default +// for callers that construct NodeHealthConfig directly instead of using a +// config loader. +func (c NodeHealthConfig) EffectiveAgentOrphanReapEnabled() bool { + return c.AgentOrphanReapEnabled == nil || *c.AgentOrphanReapEnabled } // ExecutionCleanupConfig holds configuration for execution cleanup and garbage collection @@ -581,6 +595,11 @@ func LoadConfig(configPath string) (*Config, error) { // ApplyDefaults fills values that should be stable across config loaders. func ApplyDefaults(cfg *Config) { + nodeHealth := &cfg.AgentField.NodeHealth + if nodeHealth.AgentOrphanReapEnabled == nil { + enabled := true + nodeHealth.AgentOrphanReapEnabled = &enabled + } cleanup := &cfg.AgentField.ExecutionCleanup // Cleanup is enabled by default so stale executions are still swept even // when retention is disabled. A zero retention period intentionally means @@ -782,6 +801,7 @@ func ApplyEnvOverrides(cfg *Config) { cfg.AgentField.NodeHealth.AgentDrainGrace = d } } + applyOptionalBoolEnv("AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED", &cfg.AgentField.NodeHealth.AgentOrphanReapEnabled) // LLM health monitoring overrides if val := os.Getenv("AGENTFIELD_LLM_HEALTH_ENABLED"); val != "" { @@ -1055,6 +1075,24 @@ func applyBoolEnv(name string, target *bool) bool { return false } +func applyOptionalBoolEnv(name string, target **bool) bool { + value := os.Getenv(name) + if value == "" { + return false + } + parsed, err := strconv.ParseBool(value) + if err != nil { + current := true + if *target != nil { + current = **target + } + log.Printf("warning: invalid %s=%q: %v; keeping %t", name, value, err, current) + return true + } + *target = &parsed + return true +} + func parseEnvBool(value string) bool { switch strings.ToLower(strings.TrimSpace(value)) { case "1", "true", "yes", "y", "on", "enabled": diff --git a/control-plane/internal/config/config_additional_test.go b/control-plane/internal/config/config_additional_test.go index f2212c558..d5167a85d 100644 --- a/control-plane/internal/config/config_additional_test.go +++ b/control-plane/internal/config/config_additional_test.go @@ -1,11 +1,15 @@ package config import ( + "bytes" + "log" "os" "path/filepath" "strings" "testing" "time" + + "github.com/spf13/viper" ) func TestEffectiveNodeLogProxy(t *testing.T) { @@ -662,6 +666,91 @@ func TestAgentDrainGraceFromEnvironment(t *testing.T) { } } +func TestOrphanReapEnabledParsing(t *testing.T) { + tests := []struct { + name string + value string + want bool + warn bool + }{ + {name: "unset", want: true}, + {name: "true", value: "true", want: true}, + {name: "false", value: "false", want: false}, + {name: "one", value: "1", want: true}, + {name: "zero", value: "0", want: false}, + {name: "invalid", value: "maybe", want: true, warn: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED", tt.value) + var logs bytes.Buffer + previousWriter := log.Writer() + log.SetOutput(&logs) + t.Cleanup(func() { log.SetOutput(previousWriter) }) + cfg := Config{} + ApplyDefaults(&cfg) + ApplyEnvOverrides(&cfg) + got := cfg.AgentField.NodeHealth.EffectiveAgentOrphanReapEnabled() + if got != tt.want { + t.Fatalf("expected %t for %q, got %t", tt.want, tt.value, got) + } + if got := strings.Contains(logs.String(), "invalid AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED=\"maybe\""); got != tt.warn { + t.Fatalf("warning presence = %t, want %t; logs: %s", got, tt.warn, logs.String()) + } + }) + } +} + +func TestOrphanReapEnabledYAMLFalseIsPreserved(t *testing.T) { + path := filepath.Join(t.TempDir(), "agentfield.yaml") + contents := []byte("agentfield:\n node_health:\n agent_orphan_reap_enabled: false\n") + if err := os.WriteFile(path, contents, 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + cfg, err := LoadConfig(path) + if err != nil { + t.Fatalf("load config: %v", err) + } + if cfg.AgentField.NodeHealth.EffectiveAgentOrphanReapEnabled() { + t.Fatal("expected explicit YAML false to be preserved") + } + if cfg.AgentField.NodeHealth.AgentOrphanReapEnabled == nil { + t.Fatal("expected explicit YAML false to retain presence") + } +} + +// TestOrphanReapEnabledViperFalseIsPreserved covers the loader the shipped +// binaries actually use. The pointer retains key presence through mapstructure +// so ApplyDefaults can distinguish an explicit false from an omitted value. +func TestOrphanReapEnabledViperFalseIsPreserved(t *testing.T) { + v := viper.New() + v.SetConfigType("yaml") + if err := v.ReadConfig(strings.NewReader("agentfield:\n node_health:\n agent_orphan_reap_enabled: false\n")); err != nil { + t.Fatalf("read viper config: %v", err) + } + var cfg Config + if err := v.Unmarshal(&cfg); err != nil { + t.Fatalf("unmarshal: %v", err) + } + ApplyDefaults(&cfg) + if cfg.AgentField.NodeHealth.EffectiveAgentOrphanReapEnabled() { + t.Fatal("expected explicit viper false to survive ApplyDefaults") + } + if cfg.AgentField.NodeHealth.AgentOrphanReapEnabled == nil { + t.Fatal("expected Viper to retain explicit false presence") + } +} + +func TestNodeHealthYAMLRejectsNonMapping(t *testing.T) { + path := filepath.Join(t.TempDir(), "agentfield.yaml") + if err := os.WriteFile(path, []byte("agentfield:\n node_health: 5\n"), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + if _, err := LoadConfig(path); err == nil { + t.Fatal("expected a parse error for a scalar node_health block") + } +} + func TestShutdownTimeoutEnvAcceptsBareSecondsLikeTheSDKs(t *testing.T) { t.Setenv("AGENTFIELD_SHUTDOWN_TIMEOUT", "20") cfg := &Config{} diff --git a/control-plane/internal/handlers/execute.go b/control-plane/internal/handlers/execute.go index 119b74949..de8792fce 100644 --- a/control-plane/internal/handlers/execute.go +++ b/control-plane/internal/handlers/execute.go @@ -86,6 +86,8 @@ type AsyncExecuteResponse struct { type ExecutionStatusResponse struct { ExecutionID string `json:"execution_id"` RunID string `json:"run_id"` + AgentNodeID string `json:"agent_node_id,omitempty"` + InstanceID string `json:"instance_id,omitempty"` Status string `json:"status"` StatusReason *string `json:"status_reason,omitempty"` Result interface{} `json:"result,omitempty"` diff --git a/control-plane/internal/handlers/execute_agent_restart.go b/control-plane/internal/handlers/execute_agent_restart.go index c48c35ba3..c9782e2bd 100644 --- a/control-plane/internal/handlers/execute_agent_restart.go +++ b/control-plane/internal/handlers/execute_agent_restart.go @@ -65,6 +65,7 @@ const ( // change it without racing the async worker pool. var agentRestartGraceNanos atomic.Int64 var agentDrainGraceNanos atomic.Int64 +var agentOrphanReapEnabled atomic.Bool var updatingAgentNodes = struct { sync.RWMutex names map[string]int @@ -73,6 +74,7 @@ var updatingAgentNodes = struct { func init() { agentRestartGraceNanos.Store(int64(defaultAgentRestartGrace)) agentDrainGraceNanos.Store(int64(defaultAgentDrainGrace)) + agentOrphanReapEnabled.Store(true) } // SetAgentRestartGrace configures how long a dispatch waits for a restarting @@ -87,6 +89,7 @@ func agentRestartGrace() time.Duration { } func SetAgentDrainGrace(d time.Duration) { agentDrainGraceNanos.Store(int64(d)) } +func SetAgentOrphanReapEnabled(v bool) { agentOrphanReapEnabled.Store(v) } // agentIsDraining reports whether an offline node should be treated as // draining: it went quiet recently enough that a replacement instance is @@ -113,6 +116,7 @@ func agentIsDraining(agent *types.AgentNode) bool { } func AgentDrainGrace() time.Duration { return time.Duration(agentDrainGraceNanos.Load()) } func AgentRestartGrace() time.Duration { return agentRestartGrace() } +func AgentOrphanReapEnabled() bool { return agentOrphanReapEnabled.Load() } // waitForDrainingAgent holds admission before any execution row exists while // an orderly shutdown is in progress. A replacement registration is detected diff --git a/control-plane/internal/handlers/execute_helpers.go b/control-plane/internal/handlers/execute_helpers.go index 955788821..48b26236c 100644 --- a/control-plane/internal/handlers/execute_helpers.go +++ b/control-plane/internal/handlers/execute_helpers.go @@ -322,6 +322,8 @@ func renderStatus(exec *types.Execution) ExecutionStatusResponse { resp := ExecutionStatusResponse{ ExecutionID: exec.ExecutionID, RunID: exec.RunID, + AgentNodeID: exec.AgentNodeID, + InstanceID: exec.InstanceID, Status: exec.Status, StatusReason: exec.StatusReason, Result: decodeJSON(exec.ResultPayload), diff --git a/control-plane/internal/handlers/execute_status_update_test.go b/control-plane/internal/handlers/execute_status_update_test.go index 53b59eafb..e8ecc3aeb 100644 --- a/control-plane/internal/handlers/execute_status_update_test.go +++ b/control-plane/internal/handlers/execute_status_update_test.go @@ -3,6 +3,7 @@ package handlers import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -34,6 +35,7 @@ func TestUpdateExecutionStatusHandler_Success(t *testing.T) { ExecutionID: "exec-1", RunID: "run-1", AgentNodeID: "node-1", + InstanceID: "instance-1", ReasonerID: "reasoner-a", Status: types.ExecutionStatusRunning, StartedAt: time.Now().UTC(), @@ -61,6 +63,8 @@ func TestUpdateExecutionStatusHandler_Success(t *testing.T) { var payload ExecutionStatusResponse require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &payload)) require.Equal(t, "exec-1", payload.ExecutionID) + require.Equal(t, "node-1", payload.AgentNodeID) + require.Equal(t, "instance-1", payload.InstanceID) require.Equal(t, types.ExecutionStatusSucceeded, payload.Status) require.NotNil(t, payload.CompletedAt) @@ -75,6 +79,108 @@ func TestUpdateExecutionStatusHandler_Success(t *testing.T) { require.Equal(t, int64(1000), *updated.DurationMS) } +func TestGetExecutionStatusHandler_ReturnsAgentAndInstanceIdentifiers(t *testing.T) { + gin.SetMode(gin.TestMode) + store := newTestExecutionStorage(nil) + now := time.Now().UTC() + require.NoError(t, store.CreateExecutionRecord(context.Background(), &types.Execution{ + ExecutionID: "exec-with-instance", + RunID: "run-1", + AgentNodeID: "node-1", + InstanceID: "instance-1", + Status: types.ExecutionStatusRunning, + StartedAt: now, + CreatedAt: now, + UpdatedAt: now, + })) + + router := gin.New() + router.GET("/api/v1/executions/:execution_id", GetExecutionStatusHandler(store)) + resp := httptest.NewRecorder() + router.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/api/v1/executions/exec-with-instance", nil)) + + require.Equal(t, http.StatusOK, resp.Code) + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &payload)) + require.Equal(t, "node-1", payload["agent_node_id"]) + require.Equal(t, "instance-1", payload["instance_id"]) +} + +func TestGetExecutionStatusHandler_OmitsEmptyInstanceID(t *testing.T) { + gin.SetMode(gin.TestMode) + store := newTestExecutionStorage(nil) + now := time.Now().UTC() + require.NoError(t, store.CreateExecutionRecord(context.Background(), &types.Execution{ + ExecutionID: "exec-without-instance", + RunID: "run-1", + AgentNodeID: "go-node", + Status: types.ExecutionStatusRunning, + StartedAt: now, + CreatedAt: now, + UpdatedAt: now, + })) + + router := gin.New() + router.GET("/api/v1/executions/:execution_id", GetExecutionStatusHandler(store)) + resp := httptest.NewRecorder() + router.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/api/v1/executions/exec-without-instance", nil)) + + require.Equal(t, http.StatusOK, resp.Code) + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &payload)) + require.Equal(t, "go-node", payload["agent_node_id"]) + _, present := payload["instance_id"] + require.False(t, present) +} + +type batchStatusFallbackErrorStore struct{ *testExecutionStorage } + +func (s *batchStatusFallbackErrorStore) GetExecutionRecordsBatch(context.Context, []string) (map[string]*types.Execution, error) { + return nil, fmt.Errorf("batch unavailable") +} + +func (s *batchStatusFallbackErrorStore) GetExecutionRecord(ctx context.Context, executionID string) (*types.Execution, error) { + if executionID == "exec-error" { + return nil, fmt.Errorf("read failed") + } + return s.testExecutionStorage.GetExecutionRecord(ctx, executionID) +} + +func TestBatchExecutionStatusHandler_IdentifierFieldsOnlyForFoundExecutions(t *testing.T) { + gin.SetMode(gin.TestMode) + base := newTestExecutionStorage(nil) + now := time.Now().UTC() + require.NoError(t, base.CreateExecutionRecord(context.Background(), &types.Execution{ + ExecutionID: "exec-found", + RunID: "run-1", + AgentNodeID: "node-1", + InstanceID: "instance-1", + Status: types.ExecutionStatusRunning, + StartedAt: now, + CreatedAt: now, + UpdatedAt: now, + })) + store := &batchStatusFallbackErrorStore{testExecutionStorage: base} + router := gin.New() + router.POST("/api/v1/executions/batch-status", BatchExecutionStatusHandler(store)) + req := httptest.NewRequest(http.MethodPost, "/api/v1/executions/batch-status", strings.NewReader(`{"execution_ids":["exec-found","exec-missing","exec-error"]}`)) + req.Header.Set("Content-Type", "application/json") + resp := httptest.NewRecorder() + router.ServeHTTP(resp, req) + + require.Equal(t, http.StatusOK, resp.Code) + var payload map[string]map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &payload)) + require.Equal(t, "node-1", payload["exec-found"]["agent_node_id"]) + require.Equal(t, "instance-1", payload["exec-found"]["instance_id"]) + for _, id := range []string{"exec-missing", "exec-error"} { + _, hasAgentNodeID := payload[id]["agent_node_id"] + _, hasInstanceID := payload[id]["instance_id"] + require.False(t, hasAgentNodeID, id) + require.False(t, hasInstanceID, id) + } +} + func TestUpdateExecutionStatusHandler_Failed(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/control-plane/internal/handlers/nodes_register.go b/control-plane/internal/handlers/nodes_register.go index e398fe5e6..949ebc930 100644 --- a/control-plane/internal/handlers/nodes_register.go +++ b/control-plane/internal/handlers/nodes_register.go @@ -641,7 +641,9 @@ func RegisterNodeHandler(storageProvider storage.StorageProvider, uiService *ser // Strict guard: BOTH must be non-empty. An empty stored value means // the prior process was on an older SDK that didn't report instance_id; // we can't safely conclude its work is dead, so we don't reap. - shouldReapOrphans := isReRegistration && + // The feature gate prevents the deferred reap goroutine from being armed. + shouldReapOrphans := AgentOrphanReapEnabled() && + isReRegistration && existingNode != nil && strings.TrimSpace(existingNode.InstanceID) != "" && strings.TrimSpace(newNode.InstanceID) != "" && diff --git a/control-plane/internal/handlers/nodes_register_orphan_reap_test.go b/control-plane/internal/handlers/nodes_register_orphan_reap_test.go index 3a4f6964b..3ec7d4445 100644 --- a/control-plane/internal/handlers/nodes_register_orphan_reap_test.go +++ b/control-plane/internal/handlers/nodes_register_orphan_reap_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "github.com/Agent-Field/agentfield/control-plane/internal/services" "github.com/Agent-Field/agentfield/control-plane/pkg/types" "github.com/gin-gonic/gin" @@ -353,3 +354,71 @@ var assertErrFakeReapFailure = &fakeReapError{} type fakeReapError struct{} func (*fakeReapError) Error() string { return "fake reap failure (expected in test)" } + +func TestReRegistrationSkipsOrphanReapWhenDisabled(t *testing.T) { + previousGrace := AgentDrainGrace() + previousEnabled := AgentOrphanReapEnabled() + SetAgentDrainGrace(20 * time.Millisecond) + SetAgentOrphanReapEnabled(false) + t.Cleanup(func() { + SetAgentDrainGrace(previousGrace) + SetAgentOrphanReapEnabled(previousEnabled) + }) + + gin.SetMode(gin.TestMode) + registrationStore := &orphanReapStorageStub{ + nodeRESTStorageStub: nodeRESTStorageStub{ + agent: &types.AgentNode{ + ID: "github-buddy", + BaseURL: "http://10.0.0.5:8080", + LifecycleStatus: types.AgentStatusReady, + HealthStatus: types.HealthStatusActive, + InstanceID: "alpha", + }, + }, + } + registerRouter := gin.New() + registerRouter.POST("/nodes/register", RegisterNodeHandler(registrationStore, nil, nil, nil, nil, nil)) + rec := registerNodeWithBody(t, registerRouter, `{ + "id":"github-buddy", + "base_url":"http://10.0.0.5:8080", + "instance_id":"beta", + "callback_discovery":{"mode":"manual","preferred":"http://10.0.0.5:8080"} + }`) + require.Equal(t, http.StatusCreated, rec.Code) + time.Sleep(3 * AgentDrainGrace()) + require.Empty(t, registrationStore.orphanCalls) + + executionStore := newTestExecutionStorage(&types.AgentNode{ + ID: "github-buddy", + BaseURL: "http://10.0.0.5:8080", + Reasoners: []types.ReasonerDefinition{{ID: "reasoner-a"}}, + }) + now := time.Now().UTC() + require.NoError(t, executionStore.CreateExecutionRecord(context.Background(), &types.Execution{ + ExecutionID: "exec-old-instance", + RunID: "run-1", + AgentNodeID: "github-buddy", + InstanceID: "alpha", + ReasonerID: "reasoner-a", + Status: types.ExecutionStatusRunning, + StartedAt: now, + CreatedAt: now, + UpdatedAt: now, + })) + beforeCallback, err := executionStore.GetExecutionRecord(context.Background(), "exec-old-instance") + require.NoError(t, err) + require.Equal(t, types.ExecutionStatusRunning, beforeCallback.Status) + + statusRouter := gin.New() + statusRouter.PUT("/api/v1/executions/:execution_id/status", UpdateExecutionStatusHandler(executionStore, services.NewFilePayloadStore(t.TempDir()), nil, 90*time.Second)) + statusReq := httptest.NewRequest(http.MethodPut, "/api/v1/executions/exec-old-instance/status", bytes.NewBufferString(`{"status":"succeeded","result":{"ok":true}}`)) + statusReq.Header.Set("Content-Type", "application/json") + statusResp := httptest.NewRecorder() + statusRouter.ServeHTTP(statusResp, statusReq) + require.Equal(t, http.StatusOK, statusResp.Code) + + completed, err := executionStore.GetExecutionRecord(context.Background(), "exec-old-instance") + require.NoError(t, err) + require.Equal(t, types.ExecutionStatusSucceeded, completed.Status) +} diff --git a/control-plane/internal/handlers/ui/executions.go b/control-plane/internal/handlers/ui/executions.go index 0905f172b..c7e55003f 100644 --- a/control-plane/internal/handlers/ui/executions.go +++ b/control-plane/internal/handlers/ui/executions.go @@ -161,6 +161,7 @@ type ExecutionDetailsResponse struct { SessionID *string `json:"session_id,omitempty"` ActorID *string `json:"actor_id,omitempty"` AgentNodeID string `json:"agent_node_id"` + InstanceID string `json:"instance_id,omitempty"` ParentWorkflowID *string `json:"parent_workflow_id,omitempty"` RootWorkflowID *string `json:"root_workflow_id,omitempty"` WorkflowDepth *int `json:"workflow_depth,omitempty"` @@ -766,6 +767,7 @@ func (h *ExecutionHandler) toExecutionDetails(ctx context.Context, exec *types.E SessionID: exec.SessionID, ActorID: exec.ActorID, AgentNodeID: exec.AgentNodeID, + InstanceID: exec.InstanceID, ParentWorkflowID: exec.ParentExecutionID, RootWorkflowID: nil, WorkflowDepth: nil, diff --git a/control-plane/internal/handlers/ui/executions_instance_id_test.go b/control-plane/internal/handlers/ui/executions_instance_id_test.go new file mode 100644 index 000000000..368e52587 --- /dev/null +++ b/control-plane/internal/handlers/ui/executions_instance_id_test.go @@ -0,0 +1,42 @@ +package ui + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Agent-Field/agentfield/control-plane/pkg/types" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestGetExecutionDetailsGlobalHandlerReturnsInstanceID(t *testing.T) { + store, _ := setupUIHandlerStorage(t) + now := time.Now().UTC() + require.NoError(t, store.CreateExecutionRecord(context.Background(), &types.Execution{ + ExecutionID: "exec-1", + RunID: "run-1", + AgentNodeID: "node-1", + InstanceID: "instance-1", + ReasonerID: "reasoner-a", + Status: types.ExecutionStatusRunning, + StartedAt: now, + CreatedAt: now, + UpdatedAt: now, + })) + + handler := NewExecutionHandler(store, nil, nil) + router := gin.New() + router.GET("/api/ui/v1/executions/:execution_id/details", handler.GetExecutionDetailsGlobalHandler) + resp := httptest.NewRecorder() + router.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/api/ui/v1/executions/exec-1/details", nil)) + + require.Equal(t, http.StatusOK, resp.Code) + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &payload)) + require.Equal(t, "node-1", payload["agent_node_id"]) + require.Equal(t, "instance-1", payload["instance_id"]) +} diff --git a/control-plane/internal/server/config_db.go b/control-plane/internal/server/config_db.go index 6fbadd247..77cd2b318 100644 --- a/control-plane/internal/server/config_db.go +++ b/control-plane/internal/server/config_db.go @@ -44,6 +44,12 @@ func overlayDBConfig(cfg *config.Config, store storage.StorageProvider) error { // Restore storage config (never overridden from DB) cfg.Storage = savedStorage + // The command loaders apply environment variables before server construction, + // while the database overlay necessarily happens during construction. Reapply + // them here so the documented env > DB precedence remains true, including on + // runtime config reloads. + config.ApplyEnvOverrides(cfg) + fmt.Printf("[config] Loaded config from database (key: %s, version: %d, updated: %s)\n", entry.Key, entry.Version, entry.UpdatedAt.Format(time.RFC3339)) return nil @@ -56,9 +62,7 @@ func mergeDBConfig(target, dbCfg *config.Config) { if dbCfg.AgentField.Port != 0 { target.AgentField.Port = dbCfg.AgentField.Port } - if dbCfg.AgentField.NodeHealth.CheckInterval != 0 { - target.AgentField.NodeHealth = dbCfg.AgentField.NodeHealth - } + mergeDBNodeHealthConfig(&target.AgentField.NodeHealth, dbCfg.AgentField.NodeHealth) // ARD exposure is intentionally not merged from DB config. File/env config // defines the deployment guardrails; runtime opt-in state lives in ard.state. // Merge execution cleanup field-by-field to avoid zeroing out unset fields @@ -135,3 +139,31 @@ func mergeDBConfig(target, dbCfg *config.Config) { target.UI = dbCfg.UI } } + +func mergeDBNodeHealthConfig(target *config.NodeHealthConfig, dbCfg config.NodeHealthConfig) { + if dbCfg.CheckInterval != 0 { + target.CheckInterval = dbCfg.CheckInterval + } + if dbCfg.CheckTimeout != 0 { + target.CheckTimeout = dbCfg.CheckTimeout + } + if dbCfg.ConsecutiveFailures != 0 { + target.ConsecutiveFailures = dbCfg.ConsecutiveFailures + } + if dbCfg.RecoveryDebounce != 0 { + target.RecoveryDebounce = dbCfg.RecoveryDebounce + } + if dbCfg.HeartbeatStaleThreshold != 0 { + target.HeartbeatStaleThreshold = dbCfg.HeartbeatStaleThreshold + } + if dbCfg.AgentRestartGrace != 0 { + target.AgentRestartGrace = dbCfg.AgentRestartGrace + } + if dbCfg.AgentDrainGrace != 0 { + target.AgentDrainGrace = dbCfg.AgentDrainGrace + } + if dbCfg.AgentOrphanReapEnabled != nil { + enabled := *dbCfg.AgentOrphanReapEnabled + target.AgentOrphanReapEnabled = &enabled + } +} diff --git a/control-plane/internal/server/config_db_test.go b/control-plane/internal/server/config_db_test.go index b3ad1ac37..995568492 100644 --- a/control-plane/internal/server/config_db_test.go +++ b/control-plane/internal/server/config_db_test.go @@ -28,12 +28,14 @@ func (s *configStoreStub) GetConfig(_ context.Context, key string) (*storage.Con } func baseConfigForDBTests() config.Config { + orphanReapEnabled := true return config.Config{ AgentField: config.AgentFieldConfig{ Port: 8080, NodeHealth: config.NodeHealthConfig{ - CheckInterval: 10 * time.Second, - CheckTimeout: 5 * time.Second, + CheckInterval: 10 * time.Second, + CheckTimeout: 5 * time.Second, + AgentOrphanReapEnabled: &orphanReapEnabled, }, ExecutionCleanup: config.ExecutionCleanupConfig{ Enabled: true, @@ -120,13 +122,20 @@ func TestMergeDBConfigPreservesStorageSection(t *testing.T) { func TestMergeDBConfigAppliesNonZeroDBValues(t *testing.T) { cfg := baseConfigForDBTests() + orphanReapEnabled := false dbCfg := &config.Config{ AgentField: config.AgentFieldConfig{ Port: 9090, NodeHealth: config.NodeHealthConfig{ - CheckInterval: 15 * time.Second, - CheckTimeout: 7 * time.Second, + CheckInterval: 15 * time.Second, + CheckTimeout: 7 * time.Second, + ConsecutiveFailures: 4, + RecoveryDebounce: 8 * time.Second, + HeartbeatStaleThreshold: 90 * time.Second, + AgentRestartGrace: 20 * time.Second, + AgentDrainGrace: 2 * time.Minute, + AgentOrphanReapEnabled: &orphanReapEnabled, }, ExecutionCleanup: config.ExecutionCleanupConfig{ Enabled: false, @@ -274,15 +283,93 @@ func TestOverlayDBConfigInvalidYAMLDoesNotMutateLoadedConfig(t *testing.T) { require.Equal(t, original, cfg) } +func TestOverlayDBConfigPartialNodeHealthPreservesOmittedValues(t *testing.T) { + cfg := baseConfigForDBTests() + cfg.AgentField.NodeHealth.ConsecutiveFailures = 3 + cfg.AgentField.NodeHealth.RecoveryDebounce = 5 * time.Second + cfg.AgentField.NodeHealth.HeartbeatStaleThreshold = time.Minute + cfg.AgentField.NodeHealth.AgentRestartGrace = 15 * time.Second + cfg.AgentField.NodeHealth.AgentDrainGrace = time.Minute + + err := overlayDBConfig(&cfg, &configStoreStub{entry: &storage.ConfigEntry{ + Key: dbConfigKey, + Value: `agentfield: + node_health: + check_interval: 20s + check_timeout: 7s + agent_drain_grace: 90s +`, + Version: 1, + UpdatedAt: time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC), + }}) + require.NoError(t, err) + + require.Equal(t, 20*time.Second, cfg.AgentField.NodeHealth.CheckInterval) + require.Equal(t, 7*time.Second, cfg.AgentField.NodeHealth.CheckTimeout) + require.Equal(t, 3, cfg.AgentField.NodeHealth.ConsecutiveFailures) + require.Equal(t, 5*time.Second, cfg.AgentField.NodeHealth.RecoveryDebounce) + require.Equal(t, time.Minute, cfg.AgentField.NodeHealth.HeartbeatStaleThreshold) + require.Equal(t, 15*time.Second, cfg.AgentField.NodeHealth.AgentRestartGrace) + require.Equal(t, 90*time.Second, cfg.AgentField.NodeHealth.AgentDrainGrace) + require.True(t, cfg.AgentField.NodeHealth.EffectiveAgentOrphanReapEnabled()) +} + +func TestOverlayDBConfigExplicitOrphanReapFalseWithoutCheckInterval(t *testing.T) { + cfg := baseConfigForDBTests() + + err := overlayDBConfig(&cfg, &configStoreStub{entry: &storage.ConfigEntry{ + Key: dbConfigKey, + Value: `agentfield: + node_health: + agent_orphan_reap_enabled: false +`, + Version: 1, + UpdatedAt: time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC), + }}) + require.NoError(t, err) + require.False(t, cfg.AgentField.NodeHealth.EffectiveAgentOrphanReapEnabled()) + require.NotNil(t, cfg.AgentField.NodeHealth.AgentOrphanReapEnabled) +} + +func TestOverlayDBConfigEnvironmentWinsOrphanReapSetting(t *testing.T) { + tests := []struct { + name string + envValue string + dbValue string + want bool + }{ + {name: "environment false beats database true", envValue: "false", dbValue: "true", want: false}, + {name: "environment true beats database false", envValue: "true", dbValue: "false", want: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED", tt.envValue) + cfg := baseConfigForDBTests() + config.ApplyEnvOverrides(&cfg) + + err := overlayDBConfig(&cfg, &configStoreStub{entry: &storage.ConfigEntry{ + Key: dbConfigKey, + Value: "agentfield:\n node_health:\n agent_orphan_reap_enabled: " + tt.dbValue + "\n", + Version: 1, + UpdatedAt: time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC), + }}) + require.NoError(t, err) + require.Equal(t, tt.want, cfg.AgentField.NodeHealth.EffectiveAgentOrphanReapEnabled()) + }) + } +} + func TestOverlayDBConfigRoundTripPreservesStorageAndMergesExpected(t *testing.T) { cfg := baseConfigForDBTests() + orphanReapEnabled := true dbCfg := config.Config{ AgentField: config.AgentFieldConfig{ Port: 7070, NodeHealth: config.NodeHealthConfig{ - CheckInterval: 20 * time.Second, - CheckTimeout: 9 * time.Second, + CheckInterval: 20 * time.Second, + CheckTimeout: 9 * time.Second, + AgentOrphanReapEnabled: &orphanReapEnabled, }, ExecutionCleanup: config.ExecutionCleanupConfig{ Enabled: false, diff --git a/control-plane/internal/server/server.go b/control-plane/internal/server/server.go index 4ee3e463c..e4e2c06fd 100644 --- a/control-plane/internal/server/server.go +++ b/control-plane/internal/server/server.go @@ -141,6 +141,24 @@ func newRouter() *gin.Engine { return router } +func configureAgentRestartSettings(nodeHealth config.NodeHealthConfig) { + if grace := nodeHealth.AgentRestartGrace; grace != 0 { + handlers.SetAgentRestartGrace(grace) + } + if grace := nodeHealth.AgentDrainGrace; grace != 0 { + handlers.SetAgentDrainGrace(grace) + } + orphanReapEnabled := nodeHealth.EffectiveAgentOrphanReapEnabled() + handlers.SetAgentOrphanReapEnabled(orphanReapEnabled) + if !orphanReapEnabled { + logger.Logger.Warn().Msg("agent orphan reap on re-registration is disabled (AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED=false); in-flight executions of a departing instance are left to the stale-execution sweep") + } + logger.Logger.Info(). + Dur("agent_restart_grace", handlers.AgentRestartGrace()). + Dur("agent_drain_grace", handlers.AgentDrainGrace()). + Msg("configured agent restart and drain grace windows") +} + // NewAgentFieldServer creates a new instance of the AgentFieldServer. func NewAgentFieldServer(cfg *config.Config) (*AgentFieldServer, error) { // Define agentfieldHome at the very top @@ -173,16 +191,7 @@ func NewAgentFieldServer(cfg *config.Config) (*AgentFieldServer, error) { // Configure execution event payload redaction from logging config. handlers.SetRedactPayloads(cfg.Logging.ShouldRedactPayloads()) - if grace := cfg.AgentField.NodeHealth.AgentRestartGrace; grace != 0 { - handlers.SetAgentRestartGrace(grace) - } - if grace := cfg.AgentField.NodeHealth.AgentDrainGrace; grace != 0 { - handlers.SetAgentDrainGrace(grace) - } - logger.Logger.Info(). - Dur("agent_restart_grace", handlers.AgentRestartGrace()). - Dur("agent_drain_grace", handlers.AgentDrainGrace()). - Msg("configured agent restart and drain grace windows") + configureAgentRestartSettings(cfg.AgentField.NodeHealth) Router := newRouter() diff --git a/control-plane/internal/server/server_orphan_reap_test.go b/control-plane/internal/server/server_orphan_reap_test.go new file mode 100644 index 000000000..7f7ea3db1 --- /dev/null +++ b/control-plane/internal/server/server_orphan_reap_test.go @@ -0,0 +1,64 @@ +package server + +import ( + "strings" + "testing" + "time" + + "github.com/Agent-Field/agentfield/control-plane/internal/config" + "github.com/Agent-Field/agentfield/control-plane/internal/handlers" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +func TestConfigureAgentRestartSettingsLogsWhenOrphanReapDisabled(t *testing.T) { + previous := handlers.AgentOrphanReapEnabled() + t.Cleanup(func() { handlers.SetAgentOrphanReapEnabled(previous) }) + logs := captureServerLogger(t, zerolog.DebugLevel) + + disabled := false + configureAgentRestartSettings(config.NodeHealthConfig{AgentOrphanReapEnabled: &disabled}) + + require.False(t, handlers.AgentOrphanReapEnabled()) + require.Contains(t, logs.String(), "agent orphan reap on re-registration is disabled") + require.Equal(t, 1, strings.Count(logs.String(), "agent orphan reap on re-registration is disabled")) +} + +func TestConfigureAgentRestartSettingsZeroValueDefaultsOrphanReapEnabled(t *testing.T) { + previous := handlers.AgentOrphanReapEnabled() + t.Cleanup(func() { handlers.SetAgentOrphanReapEnabled(previous) }) + logs := captureServerLogger(t, zerolog.DebugLevel) + + configureAgentRestartSettings(config.NodeHealthConfig{}) + + require.True(t, handlers.AgentOrphanReapEnabled()) + require.NotContains(t, logs.String(), "agent orphan reap on re-registration is disabled") +} + +// TestConfigureAgentRestartSettingsAppliesGraceWindows pins that moving the +// grace wiring into configureAgentRestartSettings kept its behaviour: a +// non-zero configured window reaches the handlers package, and a zero value +// leaves the existing default untouched (the "0 = use default" contract +// documented on NodeHealthConfig). +func TestConfigureAgentRestartSettingsAppliesGraceWindows(t *testing.T) { + previousRestart := handlers.AgentRestartGrace() + previousDrain := handlers.AgentDrainGrace() + previousEnabled := handlers.AgentOrphanReapEnabled() + t.Cleanup(func() { + handlers.SetAgentRestartGrace(previousRestart) + handlers.SetAgentDrainGrace(previousDrain) + handlers.SetAgentOrphanReapEnabled(previousEnabled) + }) + + configureAgentRestartSettings(config.NodeHealthConfig{ + AgentRestartGrace: 7 * time.Second, + AgentDrainGrace: 11 * time.Second, + }) + require.Equal(t, 7*time.Second, handlers.AgentRestartGrace()) + require.Equal(t, 11*time.Second, handlers.AgentDrainGrace()) + + // Zero means "leave the configured value alone", not "reset to zero". + configureAgentRestartSettings(config.NodeHealthConfig{}) + require.Equal(t, 7*time.Second, handlers.AgentRestartGrace()) + require.Equal(t, 11*time.Second, handlers.AgentDrainGrace()) +} diff --git a/docs/ENVIRONMENT_VARIABLES.md b/docs/ENVIRONMENT_VARIABLES.md index 127e803e6..2d1700454 100644 --- a/docs/ENVIRONMENT_VARIABLES.md +++ b/docs/ENVIRONMENT_VARIABLES.md @@ -120,6 +120,7 @@ The telemetry payload does not include prompts, inputs, outputs, logs, secrets, - `AGENTFIELD_SHUTDOWN_MIN_DELAY` (default: `0`): Control-plane-only delay between SIGTERM/SIGINT and closing the listener; zero preserves existing timing. Accepts bare seconds or a Go duration; invalid or negative values warn and keep the current value. This name is reserved for the control plane so future SDKs do not acquire another dual-meaning shutdown variable. Equivalent YAML: `agentfield.shutdown_min_delay`. It also affects `af server`, the code path the desktop app launches, so a value in `~/.agentfield/agentfield.yaml` slows local Ctrl+C too. See the [Kubernetes shutdown and drain recipe](deploying-on-kubernetes.md). - `AGENTFIELD_AGENT_RESTART_GRACE` (default: `15s`): How long an execution waits for an agent process to return during a coordinated restart; a negative duration disables the wait. - `AGENTFIELD_AGENT_DRAIN_GRACE` (default: `60s`): How long instance-scoped non-terminal work may keep completing after a replacement agent instance registers, before it is marked `agent_restart_orphaned`. The deferred in-memory timer is lost on a control-plane restart; the stale-execution sweep configured by `AGENTFIELD_EXECUTION_STALE_TIMEOUT` is the backstop. Equivalent YAML: `agentfield.node_health.agent_drain_grace`. See the [Kubernetes shutdown and drain recipe](deploying-on-kubernetes.md). +- `AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED` (default: `true`): Whether re-registration starts the deferred reap of the departing instance's in-flight executions. Set this to `false` when `replicas > 1` share one node ID, because a sibling registering is indistinguishable from a replacement and can otherwise reap still-running work. The stale-execution sweep remains the backstop. Invalid values keep the default `true` and log a warning. Equivalent YAML: `agentfield.node_health.agent_orphan_reap_enabled`. For Kubernetes, set `terminationGracePeriodSeconds` above the SDK drain window so the departing pod can return accepted work. Keep the agent `version` stable across rolling updates: changing it creates a separate versioned registration, so its work is recovered only by the stale sweep rather than this re-registration drain timer. diff --git a/docs/api/EXECUTE.md b/docs/api/EXECUTE.md index 18e2babd2..218e27500 100644 --- a/docs/api/EXECUTE.md +++ b/docs/api/EXECUTE.md @@ -69,7 +69,7 @@ An accepted asynchronous request returns HTTP `202`: If webhook registration failed, the response may include `webhook_error`. -Poll `GET /api/v1/executions/{execution_id}`. Its response contains `execution_id`, `run_id`, `status`, `started_at`, and `webhook_registered`, plus applicable `status_reason`, `result`, `error`, `error_details`, `completed_at`, `duration_ms`, `webhook_events`, and approval fields. +Poll `GET /api/v1/executions/{execution_id}`. Its response contains `execution_id`, `run_id`, `agent_node_id`, `status`, `started_at`, and `webhook_registered`, plus applicable `instance_id`, `status_reason`, `result`, `error`, `error_details`, `completed_at`, `duration_ms`, `webhook_events`, and approval fields. `instance_id` is present only when the agent reported one; today only the Python SDK does, so Go and TypeScript nodes omit it. It identifies the instance the execution was created against and is not re-stamped when dispatch is replayed across an agent restart. A restart-absorbed execution therefore names the departed process even though the replacement process ran the work; re-stamping is deliberately out of scope because this column is the reap scope key. Found entries returned by `POST /api/v1/executions/batch-status` expose the same identifiers, while synthetic `not_found` and `error` entries omit both `agent_node_id` and `instance_id`. This polling route is a thin status view. To retrieve the full stored execution, including input, result, status, notes, and timestamps, use `POST /api/v1/agentic/query`: diff --git a/docs/api/EXECUTION_RESTART.md b/docs/api/EXECUTION_RESTART.md index 062a6f456..0464fc2f7 100644 --- a/docs/api/EXECUTION_RESTART.md +++ b/docs/api/EXECUTION_RESTART.md @@ -47,4 +47,6 @@ Operators polling execution state should branch on the stable category before an | `agent_client_error:` | The agent reported a client-facing HTTP 4xx failure. | | `llm_unavailable`, `concurrency_limit`, `agent_timeout`, `agent_error`, `agent_unreachable`, `bad_response`, `internal_error`, `validation`, `permission_denied`, `node_unavailable`, `target_not_found` | Canonical failure categories used for operator routing and HTTP mapping. | +The instance-scoped orphan reap also sweeps legacy rows whose `instance_id` is empty. Such a row can therefore have an `agent_restart_orphaned` `status_reason` naming an instance that never owned it. The explicit `instance_id` on execution reads distinguishes that legacy case from an execution created against the departed instance. + Do not emulate restart by re-submitting `/execute`: execute has no idempotency key, creates unrelated executions, and cannot establish restart lineage or replay boundaries. diff --git a/docs/deploying-on-kubernetes.md b/docs/deploying-on-kubernetes.md index 21ea75999..1b6b5758b 100644 --- a/docs/deploying-on-kubernetes.md +++ b/docs/deploying-on-kubernetes.md @@ -63,10 +63,14 @@ Set an agent pod's `terminationGracePeriodSeconds` at least 15 seconds above its Agent registration is keyed by agent node ID and agent `version`. Keep `version` stable during an ordinary rollout: changing it creates a distinct registered version and bypasses the replacement-instance drain behavior described below. -Agent Deployments must run `replicas: 1` today. Executions are stamped with the node row's current `InstanceID` — the last registrant — and the callback URL is a single field per node ID and version. The replacement path is triggered by *any* re-registration carrying a non-empty `instance_id` that differs from the stored one, not by rollouts specifically, so horizontal replicas of one node ID are indistinguishable from a replacement. The reap additionally sweeps non-terminal rows whose `instance_id` is empty (rows written by SDKs that predate instance stamping). Every shipped agent manifest already sets `replicas: 1`. +Agent Deployments should normally run `replicas: 1` today. Executions are stamped with the node row's current `InstanceID` — the last registrant — and the callback URL is a single field per node ID and version. The replacement path is triggered by *any* re-registration carrying a non-empty `instance_id` that differs from the stored one, not by rollouts specifically, so horizontal replicas of one node ID are indistinguishable from a replacement. The reap additionally sweeps non-terminal rows whose `instance_id` is empty (rows written by SDKs that predate instance stamping). Every shipped agent manifest already sets `replicas: 1`. + +If an existing deployment runs `replicas > 1` behind one node ID, set `AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED=false`. This prevents one sibling's registration from reaping another sibling's live work; the stale sweep controlled by `AGENTFIELD_EXECUTION_STALE_TIMEOUT` remains the backstop. It does not add per-replica routing or attribution, so a single replica remains the recommended topology. When a replacement instance registers with the same node ID and version, the control plane stops routing new work to the departing instance and gives its in-flight executions `AGENTFIELD_AGENT_DRAIN_GRACE` (default `60s`) to finish. This grace is implemented by an in-memory timer, so a control-plane restart loses the timer; the stale sweep controlled by `AGENTFIELD_EXECUTION_STALE_TIMEOUT` (default `30m`) remains the backstop. Executions are scoped by `instance_id`, so only work belonging to the departing instance is reaped. +Do not treat `instance_id` as guaranteed pod attribution: only the Python SDK reports one today, so Go and TypeScript nodes leave it empty. Where it is reported it is a bare `uuid4().hex` that the SDK never logs. The only current path from that value to a pod is the control plane's re-registration reap log, which records `old_instance_id` and `new_instance_id` from the deferred reap goroutine. + Dispatch to a node whose last heartbeat falls within the drain window is held for `AGENTFIELD_AGENT_RESTART_GRACE` (default `15s`) while a replacement can register. If none does, the request returns HTTP `503` with `Retry-After: 1`. ### Sizing the drain grace @@ -92,7 +96,7 @@ For an agent whose longest reasoner runs about 10 minutes: - control-plane env `AGENTFIELD_AGENT_DRAIN_GRACE: "12m"` — `11m` drain + 5s settlement + callback latency + headroom; - control-plane env `AGENTFIELD_AGENT_RESTART_GRACE: "15s"` — leave it at the default; see the warning below. -Write these with unit suffixes. `AGENTFIELD_AGENT_DRAIN_GRACE` is parsed with plain Go duration syntax and has no bare-seconds fallback and no warning on failure, so a bare `660` is silently ignored and the `60s` default survives — unlike `AGENTFIELD_SHUTDOWN_TIMEOUT`, which does accept bare seconds and at least warns. `AGENTFIELD_AGENT_DRAIN_GRACE=0s` does **not** disable the reap: a zero value keeps the `60s` default. A negative duration makes the reap fire immediately. There is no opt-out. +Write these with unit suffixes. `AGENTFIELD_AGENT_DRAIN_GRACE` is parsed with plain Go duration syntax and has no bare-seconds fallback and no warning on failure, so a bare `660` is silently ignored and the `60s` default survives — unlike `AGENTFIELD_SHUTDOWN_TIMEOUT`, which does accept bare seconds and at least warns. `AGENTFIELD_AGENT_DRAIN_GRACE=0s` does **not** disable the reap: a zero value keeps the `60s` default. A negative duration makes the reap fire immediately. To disable the deferred reap, set `AGENTFIELD_AGENT_ORPHAN_REAP_ENABLED=false` instead. ### What raising the drain grace costs