feat(control-plane): instance identity on execution reads and an orphan-reap kill switch for multi-replica agents

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

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":

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{}

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"`

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

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),

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)
 

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) != "" &&