fix(lifecycle): guard agent execution metadata against concurrent map access (#2472)
Co-authored-by: Kandev Agent <agent@kandev.dev>
This commit is contained in:
@@ -168,7 +168,7 @@ func executionFromAgentExecution(exec *lifecycle.AgentExecution) *Execution {
|
||||
ExitCode: exec.ExitCode,
|
||||
ErrorMessage: exec.ErrorMessage,
|
||||
ACPSessionID: exec.ACPSessionID,
|
||||
Metadata: exec.Metadata,
|
||||
Metadata: exec.MetadataSnapshot(),
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
package lifecycle
|
||||
|
||||
// AgentExecution.metadata is written and read from unrelated goroutines: the
|
||||
// launch path caches resolved profile env and MCP artifacts on it, the prompt
|
||||
// path stores model overrides, and the stop path (RemoveExecution ->
|
||||
// cleanupPassthroughMCPConfig) deletes the passthrough keys. Those overlap in
|
||||
// practice — a scheduled stop can land while a relaunch is still writing — and
|
||||
// an unsynchronized Go map crashes the process with "concurrent map writes"
|
||||
// rather than corrupting quietly.
|
||||
//
|
||||
// The field is therefore unexported and every access goes through the helpers
|
||||
// below, all of which hold metadataMu. Never hand the live map to a caller or
|
||||
// store it on another struct: use MetadataSnapshot, which returns a copy.
|
||||
|
||||
func (e *AgentExecution) setMetadataValue(key string, value interface{}) {
|
||||
if e == nil {
|
||||
return
|
||||
}
|
||||
e.metadataMu.Lock()
|
||||
defer e.metadataMu.Unlock()
|
||||
if e.metadata == nil {
|
||||
e.metadata = make(map[string]interface{})
|
||||
}
|
||||
e.metadata[key] = value
|
||||
}
|
||||
|
||||
// deleteMetadataValues removes keys from the execution's metadata. Deleting an
|
||||
// absent key is a no-op, so callers need not check first.
|
||||
func (e *AgentExecution) deleteMetadataValues(keys ...string) {
|
||||
if e == nil {
|
||||
return
|
||||
}
|
||||
e.metadataMu.Lock()
|
||||
defer e.metadataMu.Unlock()
|
||||
if e.metadata == nil {
|
||||
return
|
||||
}
|
||||
for _, key := range keys {
|
||||
delete(e.metadata, key)
|
||||
}
|
||||
}
|
||||
|
||||
func (e *AgentExecution) metadataValue(key string) (interface{}, bool) {
|
||||
if e == nil {
|
||||
return nil, false
|
||||
}
|
||||
e.metadataMu.RLock()
|
||||
defer e.metadataMu.RUnlock()
|
||||
value, ok := e.metadata[key]
|
||||
return value, ok
|
||||
}
|
||||
|
||||
// metadataString returns the value at key when it is a string, and "" otherwise.
|
||||
func (e *AgentExecution) metadataString(key string) string {
|
||||
value, _ := e.metadataValue(key)
|
||||
s, _ := value.(string)
|
||||
return s
|
||||
}
|
||||
|
||||
// metadataBool returns the value at key when it is a bool, and false otherwise.
|
||||
func (e *AgentExecution) metadataBool(key string) bool {
|
||||
value, _ := e.metadataValue(key)
|
||||
b, _ := value.(bool)
|
||||
return b
|
||||
}
|
||||
|
||||
// metadataInt returns the value at key coerced to an int, and 0 otherwise. It
|
||||
// accepts the numeric shapes metadata arrives in after a JSON round-trip.
|
||||
func (e *AgentExecution) metadataInt(key string) int {
|
||||
value, ok := e.metadataValue(key)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
return metadataInt(map[string]interface{}{key: value}, key)
|
||||
}
|
||||
|
||||
// MetadataSnapshot returns a shallow copy of the execution's metadata. This is
|
||||
// what every caller that needs the whole map must use — passing the live map to
|
||||
// an executor backend, an MCP resolver, or a persistence shape leaves it being
|
||||
// read after this function returns, outside the lock.
|
||||
func (e *AgentExecution) MetadataSnapshot() map[string]interface{} {
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
e.metadataMu.RLock()
|
||||
defer e.metadataMu.RUnlock()
|
||||
if len(e.metadata) == 0 {
|
||||
return nil
|
||||
}
|
||||
snapshot := make(map[string]interface{}, len(e.metadata))
|
||||
for key, value := range e.metadata {
|
||||
snapshot[key] = value
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestExecutionMetadataConcurrentAccessIsSafe pins the invariant behind the
|
||||
// "fatal error: concurrent map writes" crash in RemoveExecution: the passthrough
|
||||
// cleanup deletes metadata keys while launch/prompt-path writers are still
|
||||
// storing on the same execution. Run with -race, this fails on any access that
|
||||
// bypasses metadataMu.
|
||||
func TestExecutionMetadataConcurrentAccessIsSafe(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
execution := &AgentExecution{ID: "exec-1", SessionID: "session-1"}
|
||||
setPassthroughMCPFiles(execution, []string{filepath.Join(t.TempDir(), "mcp.json")})
|
||||
setPassthroughMCPEnv(execution, map[string]string{"KANDEV_MCP": "1"})
|
||||
|
||||
workers := []func(int){
|
||||
func(i int) { execution.setMetadataValue("task_description", fmt.Sprintf("desc-%d", i)) },
|
||||
func(i int) { setPassthroughMCPEnv(execution, map[string]string{"KANDEV_MCP": fmt.Sprint(i)}) },
|
||||
func(i int) { setPassthroughMCPFiles(execution, []string{fmt.Sprintf("file-%d.json", i)}) },
|
||||
func(int) { mgr.cleanupPassthroughMCPConfig(execution) },
|
||||
func(int) { _ = execution.MetadataSnapshot() },
|
||||
func(int) { _ = execution.metadataString(MetadataKeyModelOverride) },
|
||||
func(int) { _ = execution.metadataBool(MetadataKeyIsRemote) },
|
||||
func(int) { _ = execution.metadataInt(MetadataKeyLocalPort) },
|
||||
func(int) { _ = getPassthroughMCPFiles(execution) },
|
||||
func(int) { _ = getPassthroughMCPEnv(execution) },
|
||||
}
|
||||
|
||||
const iterations = 200
|
||||
start := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
for _, worker := range workers {
|
||||
wg.Add(1)
|
||||
go func(work func(int)) {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
for i := range iterations {
|
||||
work(i)
|
||||
}
|
||||
}(worker)
|
||||
}
|
||||
close(start)
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// TestRemoveExecutionRacesMetadataWrites reproduces the reported stack directly:
|
||||
// a scheduled stop calls RemoveExecution -> cleanupPassthroughMCPConfig while
|
||||
// another goroutine still writes metadata on the same execution.
|
||||
func TestRemoveExecutionRacesMetadataWrites(t *testing.T) {
|
||||
mgr, execution, profile := newClaudePassthroughMCPTestManager(t)
|
||||
if err := mgr.executionStore.Add(execution); err != nil {
|
||||
t.Fatalf("add execution: %v", err)
|
||||
}
|
||||
if _, _, _, _, err := mgr.passthroughAgentCommand(t.Context(), execution, profile); err != nil {
|
||||
t.Fatalf("passthroughAgentCommand returned error: %v", err)
|
||||
}
|
||||
if len(getPassthroughMCPFiles(execution)) == 0 {
|
||||
t.Fatal("passthrough MCP config file was not stored")
|
||||
}
|
||||
|
||||
start := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
mgr.RemoveExecution(execution.ID)
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
for i := range 500 {
|
||||
execution.setMetadataValue("task_description", fmt.Sprintf("desc-%d", i))
|
||||
}
|
||||
}()
|
||||
close(start)
|
||||
wg.Wait()
|
||||
|
||||
if files := getPassthroughMCPFiles(execution); len(files) != 0 {
|
||||
t.Fatalf("passthrough MCP config files metadata = %v, want empty", files)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExecutionMetadataReadersReturnCopies covers the helpers that hand a
|
||||
// container out of the lock. Each must return a copy, or a caller mutating the
|
||||
// result reaches the stored map without holding metadataMu.
|
||||
func TestExecutionMetadataReadersReturnCopies(t *testing.T) {
|
||||
execution := &AgentExecution{ID: "exec-1"}
|
||||
setPassthroughMCPEnv(execution, map[string]string{"KANDEV_MCP": "1"})
|
||||
setPassthroughMCPFiles(execution, []string{"a.json"})
|
||||
|
||||
env := getPassthroughMCPEnv(execution)
|
||||
env["KANDEV_MCP"] = "tampered"
|
||||
env["EXTRA"] = "leaked"
|
||||
if got := getPassthroughMCPEnv(execution); got["KANDEV_MCP"] != "1" || len(got) != 1 {
|
||||
t.Fatalf("mutating the returned env leaked into the execution: %v", got)
|
||||
}
|
||||
|
||||
files := getPassthroughMCPFiles(execution)
|
||||
files[0] = "tampered.json"
|
||||
if got := getPassthroughMCPFiles(execution); got[0] != "a.json" {
|
||||
t.Fatalf("mutating the returned file list leaked into the execution: %v", got)
|
||||
}
|
||||
|
||||
// The same rule applies to the test seeding helper: it copies rather than
|
||||
// adopting the caller's map.
|
||||
seed := map[string]interface{}{"name": "kandev"}
|
||||
seeded := &AgentExecution{ID: "exec-2"}
|
||||
seeded.SetMetadataForTesting(seed)
|
||||
seed["name"] = "tampered"
|
||||
if got := seeded.metadataString("name"); got != "kandev" {
|
||||
t.Fatalf("SetMetadataForTesting adopted the caller's map: name = %q", got)
|
||||
}
|
||||
|
||||
seeded.SetMetadataForTesting(nil)
|
||||
if got := seeded.MetadataSnapshot(); got != nil {
|
||||
t.Fatalf("SetMetadataForTesting(nil) = %v, want nil metadata", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecutionMetadataAccessors(t *testing.T) {
|
||||
execution := &AgentExecution{ID: "exec-1"}
|
||||
|
||||
// Reads on empty metadata return zero values rather than panicking.
|
||||
if got := execution.metadataString("missing"); got != "" {
|
||||
t.Fatalf("metadataString on empty metadata = %q, want empty", got)
|
||||
}
|
||||
if execution.metadataBool("missing") {
|
||||
t.Fatal("metadataBool on empty metadata = true, want false")
|
||||
}
|
||||
if got := execution.metadataInt("missing"); got != 0 {
|
||||
t.Fatalf("metadataInt on empty metadata = %d, want 0", got)
|
||||
}
|
||||
if got := execution.MetadataSnapshot(); got != nil {
|
||||
t.Fatalf("MetadataSnapshot on empty metadata = %v, want nil", got)
|
||||
}
|
||||
|
||||
// setMetadataValue lazily allocates.
|
||||
execution.setMetadataValue("name", "kandev")
|
||||
execution.setMetadataValue("remote", true)
|
||||
execution.setMetadataValue("port", float64(45678)) // the shape metadata has after a JSON round-trip
|
||||
|
||||
if got := execution.metadataString("name"); got != "kandev" {
|
||||
t.Fatalf("metadataString = %q, want kandev", got)
|
||||
}
|
||||
if !execution.metadataBool("remote") {
|
||||
t.Fatal("metadataBool = false, want true")
|
||||
}
|
||||
if got := execution.metadataInt("port"); got != 45678 {
|
||||
t.Fatalf("metadataInt = %d, want 45678", got)
|
||||
}
|
||||
if got := execution.metadataString("remote"); got != "" {
|
||||
t.Fatalf("metadataString on a bool = %q, want empty", got)
|
||||
}
|
||||
|
||||
// The snapshot is a copy: mutating it must not reach the execution.
|
||||
snapshot := execution.MetadataSnapshot()
|
||||
snapshot["name"] = "tampered"
|
||||
delete(snapshot, "remote")
|
||||
if got := execution.metadataString("name"); got != "kandev" {
|
||||
t.Fatalf("snapshot mutation leaked into execution: name = %q", got)
|
||||
}
|
||||
if !execution.metadataBool("remote") {
|
||||
t.Fatal("snapshot deletion leaked into execution")
|
||||
}
|
||||
|
||||
execution.deleteMetadataValues("name", "absent")
|
||||
if _, ok := execution.metadataValue("name"); ok {
|
||||
t.Fatal("deleteMetadataValues did not remove the key")
|
||||
}
|
||||
if !execution.metadataBool("remote") {
|
||||
t.Fatal("deleteMetadataValues removed an unrelated key")
|
||||
}
|
||||
|
||||
// Every helper tolerates a nil receiver, which several call sites rely on.
|
||||
var nilExecution *AgentExecution
|
||||
nilExecution.setMetadataValue("name", "kandev")
|
||||
nilExecution.deleteMetadataValues("name")
|
||||
if _, ok := nilExecution.metadataValue("name"); ok {
|
||||
t.Fatal("metadataValue on nil execution reported a value")
|
||||
}
|
||||
if got := nilExecution.MetadataSnapshot(); got != nil {
|
||||
t.Fatalf("MetadataSnapshot on nil execution = %v, want nil", got)
|
||||
}
|
||||
}
|
||||
@@ -454,7 +454,7 @@ func (ri *ExecutorInstance) ToAgentExecution(req *ExecutorCreateRequest) *AgentE
|
||||
RuntimeName: ri.RuntimeName,
|
||||
Status: v1.AgentStatusRunning,
|
||||
StartedAt: time.Now(),
|
||||
Metadata: metadata,
|
||||
metadata: metadata,
|
||||
agentctl: ri.Client,
|
||||
standaloneInstanceID: ri.StandaloneInstanceID,
|
||||
standalonePort: ri.StandalonePort,
|
||||
|
||||
@@ -127,7 +127,7 @@ func TestPersistAuthToken(t *testing.T) {
|
||||
AuthToken: "handshake-token-abc",
|
||||
}
|
||||
execution := &AgentExecution{
|
||||
Metadata: make(map[string]interface{}),
|
||||
metadata: make(map[string]interface{}),
|
||||
}
|
||||
|
||||
m.persistAuthToken(context.Background(), instance, execution)
|
||||
@@ -138,9 +138,10 @@ func TestPersistAuthToken(t *testing.T) {
|
||||
}
|
||||
|
||||
// Verify metadata has the secret ID
|
||||
secretID, ok := execution.Metadata[MetadataKeyAuthTokenSecret].(string)
|
||||
raw, _ := execution.metadataValue(MetadataKeyAuthTokenSecret)
|
||||
secretID, ok := raw.(string)
|
||||
if !ok || secretID == "" {
|
||||
t.Fatalf("expected secret ID in metadata, got %v", execution.Metadata[MetadataKeyAuthTokenSecret])
|
||||
t.Fatalf("expected secret ID in metadata, got %v", raw)
|
||||
}
|
||||
|
||||
// Verify we can retrieve the token
|
||||
@@ -158,14 +159,14 @@ func TestPersistAuthToken(t *testing.T) {
|
||||
m := &Manager{logger: log, secretStore: store}
|
||||
|
||||
instance := &ExecutorInstance{InstanceID: "exec-123456789012"}
|
||||
execution := &AgentExecution{Metadata: make(map[string]interface{})}
|
||||
execution := &AgentExecution{metadata: make(map[string]interface{})}
|
||||
|
||||
m.persistAuthToken(context.Background(), instance, execution)
|
||||
|
||||
if len(store.store) != 0 {
|
||||
t.Fatal("expected no secrets stored")
|
||||
}
|
||||
if _, ok := execution.Metadata[MetadataKeyAuthTokenSecret]; ok {
|
||||
if _, ok := execution.metadataValue(MetadataKeyAuthTokenSecret); ok {
|
||||
t.Fatal("expected no metadata key")
|
||||
}
|
||||
})
|
||||
@@ -177,7 +178,7 @@ func TestPersistAuthToken(t *testing.T) {
|
||||
InstanceID: "exec-123456789012",
|
||||
AuthToken: "some-token",
|
||||
}
|
||||
execution := &AgentExecution{Metadata: make(map[string]interface{})}
|
||||
execution := &AgentExecution{metadata: make(map[string]interface{})}
|
||||
|
||||
// Should not panic
|
||||
m.persistAuthToken(context.Background(), instance, execution)
|
||||
@@ -195,10 +196,7 @@ func TestPersistAuthToken(t *testing.T) {
|
||||
|
||||
m.persistAuthToken(context.Background(), instance, execution)
|
||||
|
||||
if execution.Metadata == nil {
|
||||
t.Fatal("expected metadata to be initialized")
|
||||
}
|
||||
if _, ok := execution.Metadata[MetadataKeyAuthTokenSecret]; !ok {
|
||||
if _, ok := execution.metadataValue(MetadataKeyAuthTokenSecret); !ok {
|
||||
t.Fatal("expected secret ID in metadata")
|
||||
}
|
||||
})
|
||||
@@ -214,23 +212,25 @@ func TestPersistRuntimeSecrets(t *testing.T) {
|
||||
AuthToken: "agentctl-token",
|
||||
BootstrapNonce: "bootstrap-nonce",
|
||||
}
|
||||
execution := &AgentExecution{Metadata: make(map[string]interface{})}
|
||||
execution := &AgentExecution{metadata: make(map[string]interface{})}
|
||||
|
||||
m.persistRuntimeSecrets(context.Background(), instance, execution)
|
||||
|
||||
authSecretID, ok := execution.Metadata[MetadataKeyAuthTokenSecret].(string)
|
||||
rawAuth, _ := execution.metadataValue(MetadataKeyAuthTokenSecret)
|
||||
authSecretID, ok := rawAuth.(string)
|
||||
if !ok || authSecretID == "" {
|
||||
t.Fatalf("expected auth token secret ID, got %v", execution.Metadata[MetadataKeyAuthTokenSecret])
|
||||
t.Fatalf("expected auth token secret ID, got %v", rawAuth)
|
||||
}
|
||||
nonceSecretID, ok := execution.Metadata[MetadataKeyBootstrapNonceSecret].(string)
|
||||
rawNonce, _ := execution.metadataValue(MetadataKeyBootstrapNonceSecret)
|
||||
nonceSecretID, ok := rawNonce.(string)
|
||||
if !ok || nonceSecretID == "" {
|
||||
t.Fatalf("expected bootstrap nonce secret ID, got %v", execution.Metadata[MetadataKeyBootstrapNonceSecret])
|
||||
t.Fatalf("expected bootstrap nonce secret ID, got %v", rawNonce)
|
||||
}
|
||||
|
||||
if got := m.revealRuntimeSecret(context.Background(), execution.Metadata, MetadataKeyAuthTokenSecret); got != "agentctl-token" {
|
||||
if got := m.revealRuntimeSecret(context.Background(), execution.MetadataSnapshot(), MetadataKeyAuthTokenSecret); got != "agentctl-token" {
|
||||
t.Fatalf("revealed auth token = %q, want agentctl-token", got)
|
||||
}
|
||||
if got := m.revealRuntimeSecret(context.Background(), execution.Metadata, MetadataKeyBootstrapNonceSecret); got != "bootstrap-nonce" {
|
||||
if got := m.revealRuntimeSecret(context.Background(), execution.MetadataSnapshot(), MetadataKeyBootstrapNonceSecret); got != "bootstrap-nonce" {
|
||||
t.Fatalf("revealed bootstrap nonce = %q, want bootstrap-nonce", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -879,10 +879,7 @@ func (m *Manager) persistRuntimeSecret(
|
||||
return
|
||||
}
|
||||
|
||||
if execution.Metadata == nil {
|
||||
execution.Metadata = make(map[string]interface{})
|
||||
}
|
||||
execution.Metadata[metadataKey] = secret.ID
|
||||
execution.setMetadataValue(metadataKey, secret.ID)
|
||||
|
||||
m.logger.Debug("persisted runtime secret in secret store",
|
||||
zap.String("instance_id", instance.InstanceID),
|
||||
|
||||
@@ -338,10 +338,7 @@ func (m *Manager) SetSessionModel(ctx context.Context, executionID, modelID stri
|
||||
|
||||
if execution.PassthroughProcessID != "" {
|
||||
if err := m.executionStore.WithLock(executionID, func(exec *AgentExecution) {
|
||||
if exec.Metadata == nil {
|
||||
exec.Metadata = make(map[string]interface{})
|
||||
}
|
||||
exec.Metadata[MetadataKeyModelOverride] = modelID
|
||||
exec.setMetadataValue(MetadataKeyModelOverride, modelID)
|
||||
}); err != nil {
|
||||
return fmt.Errorf("failed to persist model override for execution %q: %w", executionID, err)
|
||||
}
|
||||
@@ -967,12 +964,7 @@ func (m *Manager) IsRemoteSession(ctx context.Context, sessionID string) bool {
|
||||
if execution.RuntimeName == executor.NameSprites {
|
||||
return true
|
||||
}
|
||||
if execution.Metadata != nil {
|
||||
if isRemote, ok := execution.Metadata[MetadataKeyIsRemote].(bool); ok && isRemote {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
return execution.metadataBool(MetadataKeyIsRemote)
|
||||
}
|
||||
|
||||
// Fall back to database records (post-restart, execution not yet recreated).
|
||||
@@ -1001,12 +993,7 @@ func (m *Manager) ShouldUseContainerShell(ctx context.Context, sessionID string)
|
||||
execution.RuntimeName == executor.NameSprites {
|
||||
return true
|
||||
}
|
||||
if execution.Metadata != nil {
|
||||
if isRemote, ok := execution.Metadata[MetadataKeyIsRemote].(bool); ok && isRemote {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
return execution.metadataBool(MetadataKeyIsRemote)
|
||||
}
|
||||
|
||||
// Fall back to database records (post-restart, execution not yet recreated).
|
||||
@@ -1605,7 +1592,7 @@ func (m *Manager) stopAgentViaBackend(ctx context.Context, executionID string, e
|
||||
ContainerID: execution.ContainerID,
|
||||
StandaloneInstanceID: execution.standaloneInstanceID,
|
||||
StandalonePort: execution.standalonePort,
|
||||
Metadata: execution.Metadata,
|
||||
Metadata: execution.MetadataSnapshot(),
|
||||
StopReason: reason,
|
||||
AgentStopFailed: agentStopFailed,
|
||||
}
|
||||
@@ -1675,7 +1662,7 @@ func (m *Manager) buildFreshAgentCommand(ctx context.Context, execution *AgentEx
|
||||
permissionValues["allow_indexing"] = profileInfo.AllowIndexing
|
||||
permissionValues["dangerously_skip_permissions"] = profileInfo.DangerouslySkipPermissions
|
||||
}
|
||||
if override, ok := execution.Metadata[MetadataKeyModelOverride].(string); ok && override != "" {
|
||||
if override := execution.metadataString(MetadataKeyModelOverride); override != "" {
|
||||
model = override
|
||||
}
|
||||
|
||||
|
||||
@@ -245,7 +245,7 @@ func TestManager_RestartAgentProcess_Success(t *testing.T) {
|
||||
AgentCommand: "auggie --model test",
|
||||
Status: v1.AgentStatusRunning,
|
||||
WorkspacePath: "/workspace",
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
"task_description": "review the changes",
|
||||
},
|
||||
agentctl: client,
|
||||
@@ -841,7 +841,7 @@ func TestManager_SetSessionModel_Passthrough_PersistsOverride(t *testing.T) {
|
||||
SessionID: "session-pt",
|
||||
PassthroughProcessID: "pt-process-1",
|
||||
Status: v1.AgentStatusRunning,
|
||||
Metadata: map[string]interface{}{},
|
||||
metadata: map[string]interface{}{},
|
||||
agentctl: nil, // passthrough sessions have no agentctl client
|
||||
}
|
||||
require.NoError(t, mgr.executionStore.Add(exec))
|
||||
@@ -851,21 +851,21 @@ func TestManager_SetSessionModel_Passthrough_PersistsOverride(t *testing.T) {
|
||||
// must already be persisted by the time the restart fires.
|
||||
_ = mgr.SetSessionModel(context.Background(), exec.ID, "claude-opus-4-7")
|
||||
|
||||
require.Equal(t, "claude-opus-4-7", exec.Metadata[MetadataKeyModelOverride],
|
||||
require.Equal(t, "claude-opus-4-7", exec.metadataString(MetadataKeyModelOverride),
|
||||
"SetSessionModel must persist model override on passthrough executions")
|
||||
}
|
||||
|
||||
func TestEffectivePassthroughModel(t *testing.T) {
|
||||
t.Run("returns override when set", func(t *testing.T) {
|
||||
execution := &AgentExecution{
|
||||
Metadata: map[string]interface{}{MetadataKeyModelOverride: "claude-opus-4-7"},
|
||||
metadata: map[string]interface{}{MetadataKeyModelOverride: "claude-opus-4-7"},
|
||||
}
|
||||
profile := &AgentProfileInfo{Model: "claude-sonnet-4-6"}
|
||||
require.Equal(t, "claude-opus-4-7", effectivePassthroughModel(execution, profile))
|
||||
})
|
||||
|
||||
t.Run("falls back to profile model when override empty", func(t *testing.T) {
|
||||
execution := &AgentExecution{Metadata: map[string]interface{}{}}
|
||||
execution := &AgentExecution{metadata: map[string]interface{}{}}
|
||||
profile := &AgentProfileInfo{Model: "claude-sonnet-4-6"}
|
||||
require.Equal(t, "claude-sonnet-4-6", effectivePassthroughModel(execution, profile))
|
||||
})
|
||||
@@ -1439,7 +1439,7 @@ func TestIsRemoteSession(t *testing.T) {
|
||||
SessionID: "session-2",
|
||||
RuntimeName: executor.NameStandalone,
|
||||
Status: v1.AgentStatusRunning,
|
||||
Metadata: map[string]interface{}{MetadataKeyIsRemote: true},
|
||||
metadata: map[string]interface{}{MetadataKeyIsRemote: true},
|
||||
})
|
||||
mgr := &Manager{executionStore: store}
|
||||
require.True(t, mgr.IsRemoteSession(context.Background(), "session-2"))
|
||||
@@ -1547,7 +1547,7 @@ func TestShouldUseContainerShell(t *testing.T) {
|
||||
SessionID: "session-3",
|
||||
RuntimeName: executor.NameStandalone,
|
||||
Status: v1.AgentStatusRunning,
|
||||
Metadata: map[string]interface{}{MetadataKeyIsRemote: true},
|
||||
metadata: map[string]interface{}{MetadataKeyIsRemote: true},
|
||||
})
|
||||
mgr := &Manager{executionStore: store}
|
||||
require.True(t, mgr.ShouldUseContainerShell(context.Background(), "session-3"))
|
||||
|
||||
@@ -1048,7 +1048,7 @@ func (m *Manager) promoteWorkspaceExecution(ctx context.Context, execution *Agen
|
||||
if !agentConfig.Enabled() {
|
||||
return nil, fmt.Errorf("agent type %q is disabled", agentTypeName)
|
||||
}
|
||||
preferNative := m.preferNativeBinary(agentConfig, execution.RuntimeName, execution.Metadata)
|
||||
preferNative := m.preferNativeBinary(agentConfig, execution.RuntimeName, execution.MetadataSnapshot())
|
||||
cmds, err := m.buildAgentCommand(req, profileInfo, agentConfig, preferNative)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -1554,10 +1554,7 @@ func (m *Manager) SetExecutionDescription(_ context.Context, executionID string,
|
||||
if !exists {
|
||||
return fmt.Errorf("execution %q not found", executionID)
|
||||
}
|
||||
if execution.Metadata == nil {
|
||||
execution.Metadata = make(map[string]interface{})
|
||||
}
|
||||
execution.Metadata["task_description"] = description
|
||||
execution.setMetadataValue("task_description", description)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1567,10 +1564,7 @@ func (m *Manager) SetExecutionEnv(_ context.Context, executionID string, env map
|
||||
if !exists {
|
||||
return fmt.Errorf("execution %q not found", executionID)
|
||||
}
|
||||
if execution.Metadata == nil {
|
||||
execution.Metadata = make(map[string]interface{})
|
||||
}
|
||||
execution.Metadata["runtime_env"] = cloneStringMap(env)
|
||||
execution.setMetadataValue("runtime_env", cloneStringMap(env))
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1683,31 +1677,20 @@ func (m *Manager) createBootMessage(ctx context.Context, execution *AgentExecuti
|
||||
|
||||
// getTaskDescriptionFromMetadata extracts the task description string from execution metadata.
|
||||
func getTaskDescriptionFromMetadata(execution *AgentExecution) string {
|
||||
if execution.Metadata == nil {
|
||||
return ""
|
||||
}
|
||||
if desc, ok := execution.Metadata["task_description"].(string); ok {
|
||||
return desc
|
||||
}
|
||||
return ""
|
||||
return execution.metadataString("task_description")
|
||||
}
|
||||
|
||||
// getAttachmentsFromMetadata extracts attachments from execution metadata.
|
||||
func getAttachmentsFromMetadata(execution *AgentExecution) []MessageAttachment {
|
||||
if execution.Metadata == nil {
|
||||
return nil
|
||||
}
|
||||
attachments, ok := execution.Metadata["attachments"].([]MessageAttachment)
|
||||
if ok {
|
||||
return attachments
|
||||
}
|
||||
return nil
|
||||
value, _ := execution.metadataValue("attachments")
|
||||
attachments, _ := value.([]MessageAttachment)
|
||||
return attachments
|
||||
}
|
||||
|
||||
// configureAndStartAgent configures the agent command and starts the agent subprocess.
|
||||
// Returns the effective boot command (full command with adapter args, or base command).
|
||||
func (m *Manager) configureAndStartAgent(ctx context.Context, execution *AgentExecution, approvalPolicy string) (string, error) {
|
||||
env := runtimeEnvFromMetadata(execution.Metadata)
|
||||
env := runtimeEnvFromMetadata(execution.MetadataSnapshot())
|
||||
m.mergeAgentProfileEnvForExecution(ctx, execution, env)
|
||||
if err := spillLargeWakePayloadEnv(env, execution.WorkspacePath, m.logger.Zap()); err != nil {
|
||||
m.updateExecutionError(execution.ID, "failed to prepare agent env: "+err.Error())
|
||||
|
||||
@@ -768,7 +768,7 @@ func TestConfigureAndStartAgent_DoesNotSendTaskDescriptionEnv(t *testing.T) {
|
||||
AgentProfileID: "profile-1",
|
||||
AgentCommand: "npx -y @agentclientprotocol/codex-acp",
|
||||
WorkspacePath: t.TempDir(),
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
"runtime_env": map[string]string{"KEEP_ME": "yes"},
|
||||
"task_description": strings.Repeat("large prompt\n", 1000),
|
||||
},
|
||||
@@ -827,7 +827,7 @@ func TestConfigureAndStartAgent_SpillsLargeWakePayloadEnv(t *testing.T) {
|
||||
AgentProfileID: "profile-1",
|
||||
AgentCommand: "npx -y @agentclientprotocol/codex-acp",
|
||||
WorkspacePath: workspace,
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
"runtime_env": map[string]string{
|
||||
"KANDEV_RUN_ID": "run-2",
|
||||
"KANDEV_WAKE_PAYLOAD_JSON": payload,
|
||||
@@ -864,7 +864,7 @@ func TestSetExecutionEnv_DoesNotSnapshotProfileEnvVars(t *testing.T) {
|
||||
ID: "exec-1",
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
Metadata: map[string]interface{}{},
|
||||
metadata: map[string]interface{}{},
|
||||
}
|
||||
if err := mgr.executionStore.Add(execution); err != nil {
|
||||
t.Fatalf("seed execution: %v", err)
|
||||
@@ -874,9 +874,10 @@ func TestSetExecutionEnv_DoesNotSnapshotProfileEnvVars(t *testing.T) {
|
||||
t.Fatalf("SetExecutionEnv: %v", err)
|
||||
}
|
||||
|
||||
runtimeEnv, ok := execution.Metadata["runtime_env"].(map[string]string)
|
||||
rawEnv, _ := execution.metadataValue("runtime_env")
|
||||
runtimeEnv, ok := rawEnv.(map[string]string)
|
||||
if !ok {
|
||||
t.Fatalf("runtime_env missing or wrong type: %#v", execution.Metadata["runtime_env"])
|
||||
t.Fatalf("runtime_env missing or wrong type: %#v", rawEnv)
|
||||
}
|
||||
if runtimeEnv["EXECUTOR_ONLY"] != "executor" {
|
||||
t.Fatalf("executor env missing: %+v", runtimeEnv)
|
||||
@@ -2235,10 +2236,10 @@ func TestLaunch_PersistsDockerRuntimeSecrets(t *testing.T) {
|
||||
t.Fatalf("Launch returned error: %v", err)
|
||||
}
|
||||
|
||||
if got := mgr.revealRuntimeSecret(context.Background(), execution.Metadata, MetadataKeyAuthTokenSecret); got != "agentctl-token" {
|
||||
if got := mgr.revealRuntimeSecret(context.Background(), execution.MetadataSnapshot(), MetadataKeyAuthTokenSecret); got != "agentctl-token" {
|
||||
t.Fatalf("revealed auth token = %q, want agentctl-token", got)
|
||||
}
|
||||
if got := mgr.revealRuntimeSecret(context.Background(), execution.Metadata, MetadataKeyBootstrapNonceSecret); got != "bootstrap-nonce" {
|
||||
if got := mgr.revealRuntimeSecret(context.Background(), execution.MetadataSnapshot(), MetadataKeyBootstrapNonceSecret); got != "bootstrap-nonce" {
|
||||
t.Fatalf("revealed bootstrap nonce = %q, want bootstrap-nonce", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -59,7 +59,7 @@ func (m *Manager) Start(ctx context.Context) error {
|
||||
RuntimeName: ri.RuntimeName,
|
||||
Status: v1.AgentStatusRunning,
|
||||
StartedAt: time.Now(),
|
||||
Metadata: ri.Metadata,
|
||||
metadata: ri.Metadata,
|
||||
agentctl: ri.Client,
|
||||
standaloneInstanceID: ri.StandaloneInstanceID,
|
||||
standalonePort: ri.StandalonePort,
|
||||
|
||||
@@ -3,6 +3,7 @@ package lifecycle
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"maps"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -287,18 +288,7 @@ func passthroughMCPConfigPort(execution *AgentExecution) int {
|
||||
if execution.standalonePort > 0 {
|
||||
return execution.standalonePort
|
||||
}
|
||||
if execution.Metadata == nil {
|
||||
return 0
|
||||
}
|
||||
switch value := execution.Metadata["standalone_port"].(type) {
|
||||
case int:
|
||||
return value
|
||||
case int64:
|
||||
return int(value)
|
||||
case float64:
|
||||
return int(value)
|
||||
}
|
||||
return 0
|
||||
return execution.metadataInt("standalone_port")
|
||||
}
|
||||
|
||||
func safePassthroughMCPConfigName(value string) string {
|
||||
@@ -364,7 +354,7 @@ func (m *Manager) passthroughMCPServers(ctx context.Context, execution *AgentExe
|
||||
Type: string(mcpconfig.ServerTypeHTTP),
|
||||
URL: fmt.Sprintf("http://localhost:%d/mcp", port),
|
||||
}}
|
||||
profileServers, err := m.resolveMcpServersWithParams(ctx, execution.AgentProfileID, execution.Metadata, agentConfig)
|
||||
profileServers, err := m.resolveMcpServersWithParams(ctx, execution.AgentProfileID, execution.MetadataSnapshot(), agentConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -559,10 +549,7 @@ func (m *Manager) cleanupPassthroughMCPConfig(execution *AgentExecution) {
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
if execution.Metadata != nil {
|
||||
delete(execution.Metadata, metadataKeyPassthroughMCPFiles)
|
||||
delete(execution.Metadata, metadataKeyPassthroughMCPEnv)
|
||||
}
|
||||
execution.deleteMetadataValues(metadataKeyPassthroughMCPFiles, metadataKeyPassthroughMCPEnv)
|
||||
}
|
||||
|
||||
func appendUnique(list []string, value string) []string {
|
||||
@@ -577,10 +564,11 @@ func appendUnique(list []string, value string) []string {
|
||||
// getPassthroughMCPFiles reads the recorded config-file list, tolerating both
|
||||
// []string (in-memory) and []interface{} (JSON-decoded after a restart).
|
||||
func getPassthroughMCPFiles(execution *AgentExecution) []string {
|
||||
if execution == nil || execution.Metadata == nil {
|
||||
if execution == nil {
|
||||
return nil
|
||||
}
|
||||
switch v := execution.Metadata[metadataKeyPassthroughMCPFiles].(type) {
|
||||
value, _ := execution.metadataValue(metadataKeyPassthroughMCPFiles)
|
||||
switch v := value.(type) {
|
||||
case []string:
|
||||
return append([]string(nil), v...)
|
||||
case []interface{}:
|
||||
@@ -597,21 +585,23 @@ func getPassthroughMCPFiles(execution *AgentExecution) []string {
|
||||
}
|
||||
|
||||
func setPassthroughMCPFiles(execution *AgentExecution, files []string) {
|
||||
if execution.Metadata == nil {
|
||||
execution.Metadata = map[string]interface{}{}
|
||||
}
|
||||
execution.Metadata[metadataKeyPassthroughMCPFiles] = files
|
||||
execution.setMetadataValue(metadataKeyPassthroughMCPFiles, files)
|
||||
}
|
||||
|
||||
// getPassthroughMCPEnv reads the recorded MCP env map, tolerating both
|
||||
// map[string]string and map[string]interface{} (JSON-decoded after a restart).
|
||||
// Like getPassthroughMCPFiles it returns a copy, so a caller that mutates the
|
||||
// result cannot reach the stored map behind metadataMu's back.
|
||||
func getPassthroughMCPEnv(execution *AgentExecution) map[string]string {
|
||||
if execution == nil || execution.Metadata == nil {
|
||||
if execution == nil {
|
||||
return nil
|
||||
}
|
||||
switch v := execution.Metadata[metadataKeyPassthroughMCPEnv].(type) {
|
||||
value, _ := execution.metadataValue(metadataKeyPassthroughMCPEnv)
|
||||
switch v := value.(type) {
|
||||
case map[string]string:
|
||||
return v
|
||||
out := make(map[string]string, len(v))
|
||||
maps.Copy(out, v)
|
||||
return out
|
||||
case map[string]interface{}:
|
||||
out := make(map[string]string, len(v))
|
||||
for key, item := range v {
|
||||
@@ -629,10 +619,7 @@ func setPassthroughMCPEnv(execution *AgentExecution, env map[string]string) {
|
||||
if len(env) == 0 {
|
||||
return
|
||||
}
|
||||
if execution.Metadata == nil {
|
||||
execution.Metadata = map[string]interface{}{}
|
||||
}
|
||||
execution.Metadata[metadataKeyPassthroughMCPEnv] = env
|
||||
execution.setMetadataValue(metadataKeyPassthroughMCPEnv, env)
|
||||
}
|
||||
|
||||
// passthroughAgentCommand validates passthrough support and builds the command for a passthrough session.
|
||||
@@ -799,10 +786,8 @@ func profileModel(p *AgentProfileInfo) string {
|
||||
// model is baked into the CLI command at launch time — there is no live channel
|
||||
// to swap it, so the PTY must be relaunched with a new --model.
|
||||
func effectivePassthroughModel(execution *AgentExecution, profile *AgentProfileInfo) string {
|
||||
if execution != nil && execution.Metadata != nil {
|
||||
if override, ok := execution.Metadata[MetadataKeyModelOverride].(string); ok && override != "" {
|
||||
return override
|
||||
}
|
||||
if override := execution.metadataString(MetadataKeyModelOverride); override != "" {
|
||||
return override
|
||||
}
|
||||
return profileModel(profile)
|
||||
}
|
||||
|
||||
+1
-3
@@ -63,9 +63,7 @@ func newAutoInjectExecution(description string) *AgentExecution {
|
||||
PassthroughProcessID: "proc-abc",
|
||||
}
|
||||
if description != "" {
|
||||
exec.Metadata = map[string]interface{}{
|
||||
"task_description": description,
|
||||
}
|
||||
exec.setMetadataValue("task_description", description)
|
||||
}
|
||||
return exec
|
||||
}
|
||||
|
||||
@@ -92,7 +92,7 @@ func newPassthroughMCPTestManager(t *testing.T, agentName string) (*Manager, *Ag
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
WorkspacePath: t.TempDir(),
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
"standalone_port": 45678,
|
||||
},
|
||||
standalonePort: 45678,
|
||||
@@ -607,7 +607,7 @@ func TestGetPassthroughMCPFilesDecodesRestartShapes(t *testing.T) {
|
||||
// After a backend restart, Metadata is rehydrated from JSON, so a []string
|
||||
// becomes []interface{} of strings. The reader must tolerate both shapes.
|
||||
t.Run("in-memory []string", func(t *testing.T) {
|
||||
exec := &AgentExecution{Metadata: map[string]interface{}{
|
||||
exec := &AgentExecution{metadata: map[string]interface{}{
|
||||
metadataKeyPassthroughMCPFiles: []string{"/a.json", "/b.json"},
|
||||
}}
|
||||
got := getPassthroughMCPFiles(exec)
|
||||
@@ -616,7 +616,7 @@ func TestGetPassthroughMCPFilesDecodesRestartShapes(t *testing.T) {
|
||||
}
|
||||
})
|
||||
t.Run("JSON-decoded []interface{}", func(t *testing.T) {
|
||||
exec := &AgentExecution{Metadata: map[string]interface{}{
|
||||
exec := &AgentExecution{metadata: map[string]interface{}{
|
||||
metadataKeyPassthroughMCPFiles: []interface{}{"/a.json", 42, "/b.json"},
|
||||
}}
|
||||
got := getPassthroughMCPFiles(exec)
|
||||
@@ -634,7 +634,7 @@ func TestGetPassthroughMCPFilesDecodesRestartShapes(t *testing.T) {
|
||||
|
||||
func TestGetPassthroughMCPEnvDecodesRestartShapes(t *testing.T) {
|
||||
t.Run("in-memory map[string]string", func(t *testing.T) {
|
||||
exec := &AgentExecution{Metadata: map[string]interface{}{
|
||||
exec := &AgentExecution{metadata: map[string]interface{}{
|
||||
metadataKeyPassthroughMCPEnv: map[string]string{"OPENCODE_CONFIG": "/oc.json"},
|
||||
}}
|
||||
if got := getPassthroughMCPEnv(exec); got["OPENCODE_CONFIG"] != "/oc.json" {
|
||||
@@ -642,7 +642,7 @@ func TestGetPassthroughMCPEnvDecodesRestartShapes(t *testing.T) {
|
||||
}
|
||||
})
|
||||
t.Run("JSON-decoded map[string]interface{}", func(t *testing.T) {
|
||||
exec := &AgentExecution{Metadata: map[string]interface{}{
|
||||
exec := &AgentExecution{metadata: map[string]interface{}{
|
||||
metadataKeyPassthroughMCPEnv: map[string]interface{}{"OPENCODE_CONFIG": "/oc.json", "BAD": 1},
|
||||
}}
|
||||
got := getPassthroughMCPEnv(exec)
|
||||
@@ -657,7 +657,7 @@ func TestGetPassthroughMCPEnvDecodesRestartShapes(t *testing.T) {
|
||||
|
||||
func TestWritePassthroughMCPFilesUnionTrackingOnRelaunch(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
exec := &AgentExecution{Metadata: map[string]interface{}{}}
|
||||
exec := &AgentExecution{metadata: map[string]interface{}{}}
|
||||
path := filepath.Join(t.TempDir(), "cfg.json")
|
||||
file := mcpconfig.PassthroughConfigFile{Path: path, Content: []byte("{}\n")}
|
||||
|
||||
@@ -696,7 +696,7 @@ func TestPassthroughMCPServersMergesProfileAndDropsKandevCollision(t *testing.T)
|
||||
}}
|
||||
// The default policy for an unknown runtime denies all transports; allow
|
||||
// stdio so the profile servers survive resolution.
|
||||
execution.Metadata["executor_mcp_policy"] = `{"allow_stdio":true}`
|
||||
execution.setMetadataValue("executor_mcp_policy", `{"allow_stdio":true}`)
|
||||
|
||||
if _, _, _, _, err := mgr.passthroughAgentCommand(context.Background(), execution, profile); err != nil {
|
||||
t.Fatalf("passthroughAgentCommand returned error: %v", err)
|
||||
@@ -796,7 +796,7 @@ func TestWritePassthroughMCPFilesSkipsDanglingLeafSymlink(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
execution := &AgentExecution{WorkspacePath: ws, Metadata: map[string]interface{}{}}
|
||||
execution := &AgentExecution{WorkspacePath: ws, metadata: map[string]interface{}{}}
|
||||
if err := mgr.writePassthroughMCPFiles(execution, []mcpconfig.PassthroughConfigFile{
|
||||
{Path: leaf, Content: []byte(`{"mcpServers":{}}`), MergeKey: "mcpServers"},
|
||||
}); err != nil {
|
||||
@@ -833,7 +833,7 @@ func TestResumePassthroughSessionWithoutRunnerDoesNotWriteMCPConfig(t *testing.T
|
||||
func TestPassthroughAgentCommandErrorsWhenMCPPortMissing(t *testing.T) {
|
||||
mgr, execution, profile := newClaudePassthroughMCPTestManager(t)
|
||||
execution.standalonePort = 0
|
||||
delete(execution.Metadata, "standalone_port")
|
||||
execution.deleteMetadataValues("standalone_port")
|
||||
|
||||
_, _, _, _, err := mgr.passthroughAgentCommand(context.Background(), execution, profile)
|
||||
if err == nil {
|
||||
@@ -847,7 +847,7 @@ func TestPassthroughAgentCommandErrorsWhenMCPPortMissing(t *testing.T) {
|
||||
func TestFreshPassthroughCommandErrorsWhenMCPPortMissing(t *testing.T) {
|
||||
mgr, execution, _ := newClaudePassthroughMCPTestManager(t)
|
||||
execution.standalonePort = 0
|
||||
delete(execution.Metadata, "standalone_port")
|
||||
execution.deleteMetadataValues("standalone_port")
|
||||
|
||||
if _, _, _, err := mgr.freshPassthroughCommand(context.Background(), execution); err == nil {
|
||||
t.Fatal("freshPassthroughCommand returned nil, want missing MCP port error")
|
||||
@@ -859,7 +859,7 @@ func TestFreshPassthroughCommandErrorsWhenMCPPortMissing(t *testing.T) {
|
||||
func TestResumePassthroughCommandErrorsWhenMCPPortMissing(t *testing.T) {
|
||||
mgr, execution, _ := newClaudePassthroughMCPTestManager(t)
|
||||
execution.standalonePort = 0
|
||||
delete(execution.Metadata, "standalone_port")
|
||||
execution.deleteMetadataValues("standalone_port")
|
||||
|
||||
resolved, err := mgr.resolvePassthroughAgent(context.Background(), execution)
|
||||
if err != nil {
|
||||
@@ -874,7 +874,7 @@ func TestResumePassthroughCommandErrorsWhenMCPPortMissing(t *testing.T) {
|
||||
|
||||
func TestRemoveExecutionCleansPassthroughMCPConfig(t *testing.T) {
|
||||
mgr, execution, profile := newClaudePassthroughMCPTestManager(t)
|
||||
execution.Metadata = nil
|
||||
execution.metadata = nil
|
||||
if err := mgr.executionStore.Add(execution); err != nil {
|
||||
t.Fatalf("add execution: %v", err)
|
||||
}
|
||||
|
||||
@@ -58,7 +58,7 @@ func (m *Manager) resolveMcpServers(ctx context.Context, execution *AgentExecuti
|
||||
if execution == nil {
|
||||
return nil, nil
|
||||
}
|
||||
return m.resolveMcpServersWithParams(ctx, execution.AgentProfileID, execution.Metadata, agentConfig)
|
||||
return m.resolveMcpServersWithParams(ctx, execution.AgentProfileID, execution.MetadataSnapshot(), agentConfig)
|
||||
}
|
||||
|
||||
// applyExecutorMcpPolicy reads executor metadata and applies any executor-scoped MCP policy
|
||||
|
||||
@@ -16,7 +16,7 @@ func TestMaterializeRuntimeProjectMCPForPiWritesProjectFile(t *testing.T) {
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
WorkspacePath: t.TempDir(),
|
||||
Metadata: map[string]interface{}{},
|
||||
metadata: map[string]interface{}{},
|
||||
standalonePort: 45678,
|
||||
}
|
||||
agentConfig, ok := mgr.registry.Get("pi-acp")
|
||||
@@ -58,7 +58,7 @@ func TestMaterializeRuntimeProjectMCPForCursorWritesProjectFile(t *testing.T) {
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
WorkspacePath: t.TempDir(),
|
||||
Metadata: map[string]interface{}{},
|
||||
metadata: map[string]interface{}{},
|
||||
standalonePort: 45678,
|
||||
}
|
||||
agentConfig, ok := mgr.registry.Get("cursor-acp")
|
||||
@@ -96,7 +96,7 @@ func TestMaterializeRuntimeProjectMCPSkipsWhenPortUnavailable(t *testing.T) {
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
WorkspacePath: t.TempDir(),
|
||||
Metadata: map[string]interface{}{},
|
||||
metadata: map[string]interface{}{},
|
||||
}
|
||||
agentConfig, ok := mgr.registry.Get("pi-acp")
|
||||
if !ok {
|
||||
@@ -125,7 +125,7 @@ func TestPromoteWorkspaceExecutionResetsCommandWhenProjectMCPFails(t *testing.T)
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
WorkspacePath: workspace,
|
||||
Metadata: map[string]interface{}{},
|
||||
metadata: map[string]interface{}{},
|
||||
standalonePort: 45678,
|
||||
}
|
||||
req := &LaunchRequest{
|
||||
|
||||
@@ -54,7 +54,7 @@ func (m *Manager) pollOneRemoteStatus(ctx context.Context, execution *AgentExecu
|
||||
WorkspacePath: execution.WorkspacePath,
|
||||
StandaloneInstanceID: execution.standaloneInstanceID,
|
||||
StandalonePort: execution.standalonePort,
|
||||
Metadata: execution.Metadata,
|
||||
Metadata: execution.MetadataSnapshot(),
|
||||
}
|
||||
status, statusErr := provider.GetRemoteStatus(ctx, instance)
|
||||
if statusErr != nil {
|
||||
|
||||
@@ -131,7 +131,7 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) (re
|
||||
// Check if we're reconnecting to an existing running agent process.
|
||||
// When the existing process is still alive inside a remote executor (e.g., Sprites),
|
||||
// we skip subprocess launch and go directly to ACP session initialization.
|
||||
reuseExisting, _ := execution.Metadata["reuse_existing_process"].(bool)
|
||||
reuseExisting := execution.metadataBool("reuse_existing_process")
|
||||
|
||||
if !reuseExisting && execution.AgentCommand == "" {
|
||||
return fmt.Errorf("execution %q has no agent command configured", executionID)
|
||||
@@ -188,7 +188,7 @@ func (m *Manager) preflightRemoteContributionPushes(ctx context.Context, executi
|
||||
if execution == nil || execution.agentctl == nil {
|
||||
return nil
|
||||
}
|
||||
bindings, err := remoteContributionsFromMetadata(execution.Metadata)
|
||||
bindings, err := remoteContributionsFromMetadata(execution.MetadataSnapshot())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -159,7 +159,7 @@ func TestStartAgentProcess_RunsContributionPreflightBeforeAgentStart(t *testing.
|
||||
SessionID: "session-preflight",
|
||||
AgentCommand: "agent",
|
||||
AgentProfileID: "profile-preflight",
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
MetadataKeyRemoteContributions: map[string]models.RemoteContribution{"": binding},
|
||||
},
|
||||
agentctl: agentctl.NewClient(host, port, newTestLogger()),
|
||||
@@ -222,7 +222,7 @@ func TestStartAgentProcessBoundsContributionPreflight(t *testing.T) {
|
||||
execution := &AgentExecution{
|
||||
ID: "exec-preflight-timeout", SessionID: "session-preflight-timeout",
|
||||
AgentCommand: "agent", AgentProfileID: "profile-preflight-timeout",
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
MetadataKeyRemoteContributions: map[string]models.RemoteContribution{"": binding},
|
||||
},
|
||||
agentctl: agentctl.NewClient(host, port, newTestLogger()),
|
||||
|
||||
@@ -119,41 +119,3 @@ func (m *Manager) renameBranchForSession(
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (e *AgentExecution) setMetadataValue(key string, value interface{}) {
|
||||
if e == nil {
|
||||
return
|
||||
}
|
||||
e.metadataMu.Lock()
|
||||
defer e.metadataMu.Unlock()
|
||||
if e.Metadata == nil {
|
||||
e.Metadata = make(map[string]interface{})
|
||||
}
|
||||
e.Metadata[key] = value
|
||||
}
|
||||
|
||||
func (e *AgentExecution) metadataValue(key string) (interface{}, bool) {
|
||||
if e == nil {
|
||||
return nil, false
|
||||
}
|
||||
e.metadataMu.RLock()
|
||||
defer e.metadataMu.RUnlock()
|
||||
value, ok := e.Metadata[key]
|
||||
return value, ok
|
||||
}
|
||||
|
||||
func (e *AgentExecution) metadataSnapshot() map[string]interface{} {
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
e.metadataMu.RLock()
|
||||
defer e.metadataMu.RUnlock()
|
||||
if len(e.Metadata) == 0 {
|
||||
return nil
|
||||
}
|
||||
snapshot := make(map[string]interface{}, len(e.Metadata))
|
||||
for key, value := range e.Metadata {
|
||||
snapshot[key] = value
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
|
||||
@@ -68,7 +68,7 @@ func TestRenameBranchForSessionUsesRepositoryScopedAgentctlOperation(t *testing.
|
||||
execution := &AgentExecution{
|
||||
ID: "execution-1",
|
||||
SessionID: "session-1",
|
||||
Metadata: map[string]interface{}{MetadataKeyWorktreeBranch: "feature/provisional-abc"},
|
||||
metadata: map[string]interface{}{MetadataKeyWorktreeBranch: "feature/provisional-abc"},
|
||||
agentctl: agentctlclient.NewClient(parsed.Hostname(), port, newTestLogger()),
|
||||
}
|
||||
store := NewExecutionStore()
|
||||
@@ -87,7 +87,7 @@ func TestRenameBranchForSessionUsesRepositoryScopedAgentctlOperation(t *testing.
|
||||
if got.NewName != "feature/final-title-abc" || got.Repo != "backend" {
|
||||
t.Fatalf("request = %#v, want final branch and backend repo", got)
|
||||
}
|
||||
if gotBranch, _ := execution.Metadata[MetadataKeyWorktreeBranch].(string); gotBranch != "feature/provisional-abc" {
|
||||
if gotBranch := execution.metadataString(MetadataKeyWorktreeBranch); gotBranch != "feature/provisional-abc" {
|
||||
t.Fatalf("non-primary metadata branch = %q, want unchanged", gotBranch)
|
||||
}
|
||||
}
|
||||
@@ -108,7 +108,7 @@ func TestRenameBranchForSessionUpdatesPrimaryExecutionMetadata(t *testing.T) {
|
||||
execution := &AgentExecution{
|
||||
ID: "execution-1",
|
||||
SessionID: "session-1",
|
||||
Metadata: map[string]interface{}{MetadataKeyWorktreeBranch: "feature/provisional-abc"},
|
||||
metadata: map[string]interface{}{MetadataKeyWorktreeBranch: "feature/provisional-abc"},
|
||||
agentctl: agentctlclient.NewClient(parsed.Hostname(), port, newTestLogger()),
|
||||
}
|
||||
store := NewExecutionStore()
|
||||
@@ -123,7 +123,7 @@ func TestRenameBranchForSessionUpdatesPrimaryExecutionMetadata(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("RenameBranchForSession returned error: %v", err)
|
||||
}
|
||||
if got, _ := execution.Metadata[MetadataKeyWorktreeBranch].(string); got != "feature/final-title-abc" {
|
||||
if got := execution.metadataString(MetadataKeyWorktreeBranch); got != "feature/final-title-abc" {
|
||||
t.Fatalf("primary metadata branch = %q, want final branch", got)
|
||||
}
|
||||
if writer.branch != "feature/final-title-abc" {
|
||||
@@ -150,7 +150,7 @@ func TestRenameBranchForSessionReportsPrimarySnapshotFailure(t *testing.T) {
|
||||
execution := &AgentExecution{
|
||||
ID: "execution-rotated",
|
||||
SessionID: "session-rotated",
|
||||
Metadata: map[string]interface{}{MetadataKeyWorktreeBranch: "feature/provisional-abc"},
|
||||
metadata: map[string]interface{}{MetadataKeyWorktreeBranch: "feature/provisional-abc"},
|
||||
agentctl: agentctlclient.NewClient(parsed.Hostname(), port, newTestLogger()),
|
||||
}
|
||||
store := NewExecutionStore()
|
||||
@@ -180,7 +180,7 @@ func TestRenameBranchForSessionReportsPrimarySnapshotFailure(t *testing.T) {
|
||||
if result == nil || !result.Success {
|
||||
t.Fatalf("result = %#v, want successful Git result", result)
|
||||
}
|
||||
if got, _ := execution.Metadata[MetadataKeyWorktreeBranch].(string); got != "feature/final-title-abc" {
|
||||
if got := execution.metadataString(MetadataKeyWorktreeBranch); got != "feature/final-title-abc" {
|
||||
t.Fatalf("primary metadata branch = %q, want final branch", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -71,7 +71,7 @@ func buildRunningFromExecution(execution *AgentExecution, prior *models.Executor
|
||||
lastSeenAt = &now
|
||||
}
|
||||
|
||||
metadata := execution.metadataSnapshot()
|
||||
metadata := execution.MetadataSnapshot()
|
||||
running := &models.ExecutorRunning{
|
||||
ID: execution.SessionID,
|
||||
SessionID: execution.SessionID,
|
||||
@@ -120,10 +120,10 @@ func agentctlPortFromExecution(execution *AgentExecution, agentctlURL string) in
|
||||
if execution.standalonePort > 0 {
|
||||
return execution.standalonePort
|
||||
}
|
||||
if port := metadataInt(execution.Metadata, MetadataKeySSHLocalForwardPort); port > 0 {
|
||||
if port := execution.metadataInt(MetadataKeySSHLocalForwardPort); port > 0 {
|
||||
return port
|
||||
}
|
||||
if port := metadataInt(execution.Metadata, MetadataKeyLocalPort); port > 0 {
|
||||
if port := execution.metadataInt(MetadataKeyLocalPort); port > 0 {
|
||||
return port
|
||||
}
|
||||
if agentctlURL == "" {
|
||||
@@ -161,7 +161,7 @@ func agentctlPIDFromExecution(execution *AgentExecution) int {
|
||||
if execution == nil {
|
||||
return 0
|
||||
}
|
||||
return metadataInt(execution.Metadata, MetadataKeySSHRemoteAgentctlPID)
|
||||
return execution.metadataInt(MetadataKeySSHRemoteAgentctlPID)
|
||||
}
|
||||
|
||||
func metadataInt(metadata map[string]interface{}, key string) int {
|
||||
|
||||
@@ -139,7 +139,7 @@ func TestBuildRunningFromExecutionPersistsSSHRuntimePID(t *testing.T) {
|
||||
RuntimeName: agentruntime.RuntimeSSH,
|
||||
Status: v1.AgentStatusRunning,
|
||||
agentctl: client,
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
MetadataKeySSHLocalForwardPort: "43001",
|
||||
MetadataKeySSHRemoteAgentctlPID: "9321",
|
||||
MetadataKeySSHRemoteAgentctlPort: "43000",
|
||||
@@ -258,7 +258,7 @@ func TestSSHRowNeverCarriesLocalPID(t *testing.T) {
|
||||
RuntimeName: agentruntime.RuntimeSSH,
|
||||
Status: v1.AgentStatusRunning,
|
||||
agentctl: client,
|
||||
Metadata: map[string]interface{}{
|
||||
metadata: map[string]interface{}{
|
||||
MetadataKeySSHLocalForwardPort: "43001",
|
||||
MetadataKeySSHRemoteAgentctlPID: "9321",
|
||||
},
|
||||
|
||||
@@ -96,7 +96,7 @@ func (m *Manager) StopProcess(ctx context.Context, processID string) error {
|
||||
}
|
||||
|
||||
func processEnvironment(execution *AgentExecution, requested map[string]string) map[string]string {
|
||||
managedPath, _ := execution.Metadata[managedGoCacheMetadataKey].(string)
|
||||
managedPath := execution.metadataString(managedGoCacheMetadataKey)
|
||||
if managedPath == "" {
|
||||
return requested
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ func TestStartProcessMergesManagedGoCacheAndReleasesTerminalLease(t *testing.T)
|
||||
manager.SetActivityCoordinator(coordinator)
|
||||
execution := &AgentExecution{
|
||||
ID: "exec-1", SessionID: "session-1", agentctl: client,
|
||||
Metadata: map[string]interface{}{managedGoCacheMetadataKey: "/managed/go-cache"},
|
||||
metadata: map[string]interface{}{managedGoCacheMetadataKey: "/managed/go-cache"},
|
||||
}
|
||||
if err := manager.executionStore.Add(execution); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -39,22 +39,18 @@ func (m *Manager) cacheResolvedProfileEnv(execution *AgentExecution, resolved ma
|
||||
if execution == nil || len(resolved) == 0 {
|
||||
return
|
||||
}
|
||||
if execution.Metadata == nil {
|
||||
execution.Metadata = make(map[string]interface{})
|
||||
}
|
||||
execution.Metadata[metadataKeyProfileEnvResolved] = cloneStringMap(resolved)
|
||||
execution.setMetadataValue(metadataKeyProfileEnvResolved, cloneStringMap(resolved))
|
||||
}
|
||||
|
||||
func (m *Manager) mergeAgentProfileEnvForExecution(ctx context.Context, execution *AgentExecution, env map[string]string) {
|
||||
if execution != nil {
|
||||
if cached, ok := execution.Metadata[metadataKeyProfileEnvResolved].(map[string]string); ok && len(cached) > 0 {
|
||||
mergeEnvFillMissing(env, cached)
|
||||
return
|
||||
}
|
||||
}
|
||||
if execution == nil {
|
||||
return
|
||||
}
|
||||
value, _ := execution.metadataValue(metadataKeyProfileEnvResolved)
|
||||
if cached, ok := value.(map[string]string); ok && len(cached) > 0 {
|
||||
mergeEnvFillMissing(env, cached)
|
||||
return
|
||||
}
|
||||
m.mergeAgentProfileEnv(ctx, execution.AgentProfileID, env)
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
package lifecycle
|
||||
|
||||
import agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
|
||||
import (
|
||||
"maps"
|
||||
|
||||
agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
|
||||
)
|
||||
|
||||
// ExecutionStoreForTesting returns the manager's execution store.
|
||||
// Intended for use by test packages that need to inject executions.
|
||||
@@ -13,3 +17,21 @@ func (m *Manager) ExecutionStoreForTesting() *ExecutionStore {
|
||||
func (ae *AgentExecution) SetAgentCtlClientForTesting(client *agentctl.Client) {
|
||||
ae.agentctl = client
|
||||
}
|
||||
|
||||
// SetMetadataForTesting seeds metadata on a synthetic execution built by a test
|
||||
// in a dependent package. The metadata map itself is unexported so that all
|
||||
// production access stays behind the metadataMu helpers.
|
||||
//
|
||||
// The argument is copied rather than adopted, so a caller that keeps reading or
|
||||
// writing the map it passed in cannot reach the execution's store without going
|
||||
// through those helpers.
|
||||
func (ae *AgentExecution) SetMetadataForTesting(metadata map[string]interface{}) {
|
||||
ae.metadataMu.Lock()
|
||||
defer ae.metadataMu.Unlock()
|
||||
if metadata == nil {
|
||||
ae.metadata = nil
|
||||
return
|
||||
}
|
||||
ae.metadata = make(map[string]interface{}, len(metadata))
|
||||
maps.Copy(ae.metadata, metadata)
|
||||
}
|
||||
|
||||
@@ -56,8 +56,11 @@ type AgentExecution struct {
|
||||
ExitCode *int
|
||||
ErrorMessage string
|
||||
ProviderError *streams.ProviderError
|
||||
Metadata map[string]interface{}
|
||||
metadataMu sync.RWMutex
|
||||
// metadata is unexported on purpose: it is touched from the launch, prompt
|
||||
// and stop paths concurrently, so all access must go through the metadataMu
|
||||
// helpers in execution_metadata.go.
|
||||
metadata map[string]interface{}
|
||||
metadataMu sync.RWMutex
|
||||
// runtimeEnv is the effective environment used to create the task's
|
||||
// runtime instance. It is kept in memory only so authorized task-scoped
|
||||
// terminals and passthrough processes can inherit the same credentials and
|
||||
|
||||
@@ -303,8 +303,8 @@ func TestRuntime_GetExecution_ReturnsSnapshot(t *testing.T) {
|
||||
FinishedAt: &finishedAt,
|
||||
ExitCode: &exit,
|
||||
ACPSessionID: "acp-1",
|
||||
Metadata: map[string]interface{}{"k": "v"},
|
||||
}
|
||||
backend.executions["exec-1"].SetMetadataForTesting(map[string]interface{}{"k": "v"})
|
||||
rt := agentruntime.New(backend)
|
||||
got, err := rt.GetExecution(context.Background(), "exec-1")
|
||||
if err != nil {
|
||||
|
||||
@@ -134,15 +134,16 @@ func (a *lifecycleAdapter) LaunchAgent(ctx context.Context, req *executor.Launch
|
||||
}
|
||||
|
||||
// Extract worktree info from metadata if available
|
||||
metadata := execution.MetadataSnapshot()
|
||||
var worktreeID, worktreePath, worktreeBranch string
|
||||
if execution.Metadata != nil {
|
||||
if id, ok := execution.Metadata["worktree_id"].(string); ok {
|
||||
if metadata != nil {
|
||||
if id, ok := metadata["worktree_id"].(string); ok {
|
||||
worktreeID = id
|
||||
}
|
||||
if path, ok := execution.Metadata["worktree_path"].(string); ok {
|
||||
if path, ok := metadata["worktree_path"].(string); ok {
|
||||
worktreePath = path
|
||||
}
|
||||
if branch, ok := execution.Metadata["worktree_branch"].(string); ok {
|
||||
if branch, ok := metadata["worktree_branch"].(string); ok {
|
||||
worktreeBranch = branch
|
||||
}
|
||||
}
|
||||
@@ -173,7 +174,7 @@ func (a *lifecycleAdapter) LaunchAgent(ctx context.Context, req *executor.Launch
|
||||
WorktreePath: worktreePath,
|
||||
WorktreeBranch: worktreeBranch,
|
||||
WorkspacePath: execution.WorkspacePath,
|
||||
Metadata: execution.Metadata,
|
||||
Metadata: metadata,
|
||||
PrepareResult: execution.PrepareResult,
|
||||
Worktrees: worktrees,
|
||||
}, nil
|
||||
|
||||
Reference in New Issue
Block a user