Remove the secretKeyRef env sources feature

cmd/ateapi/internal/controlapi/functional_test.go

@@ -366,7 +366,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 	volPlugins := map[string]volume.VolumePluginControlPlane{
 		mockDriverName: mockPlugin,
 	}
-	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, k8sClient, instruments, "", volPlugins)
+	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins)
 
 	// 5. Start REAL gRPC Server for ATE API
 	grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ateinterceptors.ServerUnaryInterceptor))
@@ -1797,24 +1797,11 @@ func TestResumeActor(t *testing.T) {
 	}
 }
 
-func TestResumeActorResolvesValueFromEnv(t *testing.T) {
-	ns := namespaceForTest("ns-resume-secret-env")
+func TestResumeActorPassesLiteralEnv(t *testing.T) {
+	ns := namespaceForTest("ns-resume-literal-env")
 	tc := setupTest(t, ns)
 	defer tc.cleanup()
 
-	_, err := tc.k8sClient.CoreV1().Secrets(ns).Create(context.Background(), &corev1.Secret{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "api-keys",
-			Namespace: ns,
-		},
-		Data: map[string][]byte{
-			"anthropic": []byte("sk-test"),
-		},
-	}, metav1.CreateOptions{})
-	if err != nil {
-		t.Fatalf("failed to create secret: %v", err)
-	}
-
 	createTemplateWithContainers(t, tc, ns, []atev1alpha1.Container{
 		{
 			Name:    "main",
@@ -1823,23 +1810,14 @@ func TestResumeActorResolvesValueFromEnv(t *testing.T) {
 			Env: []atev1alpha1.EnvVar{
 				{
 					Name:  "LITERAL",
-					Value: ptr.To("plain"),
-				},
-				{
-					Name: "ANTHROPIC_API_KEY",
-					ValueFrom: &atev1alpha1.EnvVarSource{
-						SecretKeyRef: &atev1alpha1.SecretKeySelector{
-							Name: "api-keys",
-							Key:  "anthropic",
-						},
-					},
+					Value: "plain",
 				},
 			},
 		},
 	})
 	createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
 
-	_, err = tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{
+	_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{
 		Metadata:               &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "id1"},
 		ActorTemplateNamespace: ns,
 		ActorTemplateName:      "tmpl1",
@@ -1866,11 +1844,10 @@ func TestResumeActorResolvesValueFromEnv(t *testing.T) {
 		gotEnv[env.GetName()] = env.GetValue()
 	}
 	wantEnv := map[string]string{
-		"LITERAL":           "plain",
-		"ANTHROPIC_API_KEY": "sk-test",
+		"LITERAL": "plain",
 	}
 	if diff := cmp.Diff(wantEnv, gotEnv); diff != "" {
-		t.Errorf("resolved env mismatch (-want +got):\n%s", diff)
+		t.Errorf("env mismatch (-want +got):\n%s", diff)
 	}
 }
 

cmd/ateapi/internal/controlapi/service.go

@@ -24,7 +24,6 @@ import (
 	"github.com/agent-substrate/substrate/internal/volume/csi"
 	listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
-	"k8s.io/client-go/kubernetes"
 	storagev1listers "k8s.io/client-go/listers/storage/v1"
 )
 
@@ -61,7 +60,6 @@ func NewService(
 	csiDriverConfigLister listersv1alpha1.CSIDriverConfigLister,
 	storageClassLister storagev1listers.StorageClassLister,
 	dialer *AteletDialer,
-	kubeClient kubernetes.Interface,
 	instruments *Instruments,
 	egressGatewayAddress string,
 	volumePlugins map[string]volume.VolumePluginControlPlane,
@@ -77,7 +75,7 @@ func NewService(
 		instruments:           instruments,
 		volumePlugins:         volumePlugins,
 	}
-	s.actorWorkflow = NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, kubeClient, instruments, egressGatewayAddress, s)
+	s.actorWorkflow = NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s)
 	return s
 }
 

cmd/ateapi/internal/controlapi/workflow.go

@@ -30,7 +30,6 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	grpcCodes "google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"k8s.io/client-go/kubernetes"
 	storagev1listers "k8s.io/client-go/listers/storage/v1"
 )
 
@@ -76,8 +75,6 @@ type ActorWorkflow struct {
 	workerPoolLister     listersv1alpha1.WorkerPoolLister
 	sandboxConfigLister  listersv1alpha1.SandboxConfigLister
 	storageClassLister   storagev1listers.StorageClassLister
-	kubeClient           kubernetes.Interface
-	secretCache          *envSecretCache
 	instruments          *Instruments
 	egressGatewayAddress string
 	pluginRegistry       VolumePluginRegistry
@@ -92,7 +89,6 @@ func NewActorWorkflow(
 	workerPoolLister listersv1alpha1.WorkerPoolLister,
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	storageClassLister storagev1listers.StorageClassLister,
-	kubeClient kubernetes.Interface,
 	instruments *Instruments,
 	egressGatewayAddress string,
 	pluginRegistry VolumePluginRegistry,
@@ -106,8 +102,6 @@ func NewActorWorkflow(
 		workerPoolLister:     workerPoolLister,
 		sandboxConfigLister:  sandboxConfigLister,
 		storageClassLister:   storageClassLister,
-		kubeClient:           kubeClient,
-		secretCache:          newEnvSecretCache(envSecretCacheTTL),
 		instruments:          instruments,
 		egressGatewayAddress: egressGatewayAddress,
 		pluginRegistry:       pluginRegistry,

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -613,7 +613,7 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou
 	}
 	client := ateletpb.NewAteomHerderClient(ateletConn)
 
-	workloadSpec, err := workloadSpecFromActorTemplateWithEnv(ctx, w.kubeClient, w.secretCache, actorTemplate, actor)
+	workloadSpec, err := workloadSpecFromActorTemplate(actorTemplate, actor)
 	if err != nil {
 		return tele, err
 	}

cmd/ateapi/internal/controlapi/workflow_suspend_test.go

@@ -679,7 +679,7 @@ func TestSuspendActor_PausedWithoutLocalSnapshotCrashes(t *testing.T) {
 	}); err != nil {
 		t.Fatalf("add template to indexer: %v", err)
 	}
-	w := NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, nil, "", nil)
+	w := NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil)
 
 	seedWorkflowActor(t, ctx, st, resources.ActorRef{Atespace: "team-a", Name: "id1"}, "ns", "tmpl1", ateapipb.Actor_STATUS_PAUSED)
 

cmd/ateapi/internal/controlapi/workflow_testutil_test.go

@@ -42,7 +42,7 @@ func newTestActorWorkflow(t *testing.T, st store.Interface, tmplNamespace, tmplN
 	}); err != nil {
 		t.Fatalf("add template to indexer: %v", err)
 	}
-	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, nil, "", nil)
+	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil)
 }
 
 // seedWorkflowActor stores an actor with the given status, bound to the given

cmd/ateapi/internal/controlapi/workload_spec.go

@@ -15,27 +15,15 @@
 package controlapi
 
 import (
-	"context"
 	"fmt"
-	"sync"
-	"time"
 
 	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-	corev1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 )
 
-const envSecretCacheTTL = 30 * time.Second
-
-// workloadSpecFromActorTemplate builds a WorkloadSpec without resolving
-// container env vars. Use this when downstream consumers (e.g. checkpoint
-// requests) don't need env entries materialized.
+// workloadSpecFromActorTemplate builds a WorkloadSpec from the template;
+// container env is copied verbatim.
 func workloadSpecFromActorTemplate(actorTemplate *atev1alpha1.ActorTemplate, actor *ateapipb.Actor) (*ateletpb.WorkloadSpec, error) {
 	workloadSpec := &ateletpb.WorkloadSpec{}
 
@@ -67,6 +55,12 @@ func workloadSpecFromActorTemplate(actorTemplate *atev1alpha1.ActorTemplate, act
 			Args:    ctr.Args,
 			Readyz:  toAteletReadyz(ctr.Readyz),
 		}
+		for _, env := range ctr.Env {
+			ateletCtr.Env = append(ateletCtr.Env, &ateletpb.EnvEntry{
+				Name:  env.Name,
+				Value: env.Value,
+			})
+		}
 		for _, mount := range ctr.VolumeMounts {
 			ateletCtr.VolumeMounts = append(ateletCtr.VolumeMounts, &ateletpb.VolumeMount{
 				Name:      mount.Name,
@@ -79,36 +73,6 @@ func workloadSpecFromActorTemplate(actorTemplate *atev1alpha1.ActorTemplate, act
 	return workloadSpec, nil
 }
 
-// workloadSpecFromActorTemplateWithEnv builds a WorkloadSpec and resolves each
-// container's env vars against the cluster. kubeClient must be non-nil;
-// secretCache is optional and, when supplied, deduplicates Secret reads.
-func workloadSpecFromActorTemplateWithEnv(ctx context.Context, kubeClient kubernetes.Interface, secretCache *envSecretCache, actorTemplate *atev1alpha1.ActorTemplate, actor *ateapipb.Actor) (*ateletpb.WorkloadSpec, error) {
-	workloadSpec, err := workloadSpecFromActorTemplate(actorTemplate, actor)
-	if err != nil {
-		return nil, err
-	}
-
-	resolver := envResolver{
-		kubeClient: kubeClient,
-		namespace:  actorTemplate.Namespace,
-		cache:      secretCache,
-	}
-
-	for i, ctr := range actorTemplate.Spec.Containers {
-		for _, env := range ctr.Env {
-			ateletEnv, err := resolver.resolve(ctx, ctr.Name, env)
-			if err != nil {
-				return nil, err
-			}
-			if ateletEnv != nil {
-				workloadSpec.Containers[i].Env = append(workloadSpec.Containers[i].Env, ateletEnv)
-			}
-		}
-	}
-
-	return workloadSpec, nil
-}
-
 // appendExternalVolumes maps template external volumes to resolved actor volumes and appends them to workloadSpec
 // if they are referenced in container volumeMounts.
 func appendExternalVolumes(workloadSpec *ateletpb.WorkloadSpec, template *atev1alpha1.ActorTemplate, actor *ateapipb.Actor) error {
@@ -181,136 +145,3 @@ func toAteletReadyz(in *atev1alpha1.ContainerReadyz) *ateletpb.Readyz {
 	}
 	return out
 }
-
-type envResolver struct {
-	kubeClient kubernetes.Interface
-	namespace  string
-	cache      *envSecretCache
-}
-
-func (r *envResolver) resolve(ctx context.Context, containerName string, env atev1alpha1.EnvVar) (*ateletpb.EnvEntry, error) {
-	envID := fmt.Sprintf("container %q env %q", containerName, env.Name)
-
-	switch {
-	case env.Value != nil:
-		return &ateletpb.EnvEntry{
-			Name:  env.Name,
-			Value: *env.Value,
-		}, nil
-	case env.ValueFrom != nil:
-		value, include, err := r.resolveValueFrom(ctx, envID, env.ValueFrom)
-		if err != nil {
-			return nil, err
-		}
-		if !include {
-			return nil, nil
-		}
-		return &ateletpb.EnvEntry{
-			Name:  env.Name,
-			Value: value,
-		}, nil
-	}
-	return nil, status.Errorf(codes.FailedPrecondition, "%s has unknown value source", envID)
-}
-
-func (r *envResolver) resolveValueFrom(ctx context.Context, envID string, valueFrom *atev1alpha1.EnvVarSource) (string, bool, error) {
-	if ref := valueFrom.SecretKeyRef; ref != nil {
-		return r.resolveSecretKeyRef(ctx, envID, ref)
-	}
-	return "", false, status.Errorf(codes.FailedPrecondition, "%s uses unsupported valueFrom source; only secretKeyRef is supported", envID)
-}
-
-func (r *envResolver) resolveSecretKeyRef(ctx context.Context, envID string, ref *atev1alpha1.SecretKeySelector) (string, bool, error) {
-	if r.kubeClient == nil {
-		return "", false, status.Errorf(codes.FailedPrecondition, "%s cannot resolve secretKeyRef because Kubernetes client is unavailable", envID)
-	}
-
-	secret, err := r.secret(ctx, ref.Name)
-	if err != nil {
-		if apierrors.IsNotFound(err) {
-			if isOptional(ref.Optional) {
-				return "", false, nil
-			}
-			return "", false, status.Errorf(codes.FailedPrecondition, "%s references missing secret %s/%s", envID, r.namespace, ref.Name)
-		}
-		return "", false, status.Errorf(codes.Internal, "while resolving %s secretKeyRef %s/%s: %v", envID, r.namespace, ref.Name, err)
-	}
-
-	value, ok := secret.Data[ref.Key]
-	if !ok {
-		if isOptional(ref.Optional) {
-			return "", false, nil
-		}
-		return "", false, status.Errorf(codes.FailedPrecondition, "%s references missing key %q in secret %s/%s", envID, ref.Key, r.namespace, ref.Name)
-	}
-
-	return string(value), true, nil
-}
-
-func (r *envResolver) secret(ctx context.Context, name string) (*corev1.Secret, error) {
-	if r.cache != nil {
-		return r.cache.get(ctx, r.kubeClient, r.namespace, name)
-	}
-	return r.kubeClient.CoreV1().Secrets(r.namespace).Get(ctx, name, metav1.GetOptions{})
-}
-
-type envSecretCache struct {
-	mu      sync.RWMutex
-	ttl     time.Duration
-	entries map[envSecretCacheKey]envSecretCacheEntry
-}
-
-type envSecretCacheKey struct {
-	namespace string
-	name      string
-}
-
-type envSecretCacheEntry struct {
-	secret    *corev1.Secret
-	expiresAt time.Time
-}
-
-func newEnvSecretCache(ttl time.Duration) *envSecretCache {
-	return &envSecretCache{
-		ttl:     ttl,
-		entries: map[envSecretCacheKey]envSecretCacheEntry{},
-	}
-}
-
-func (c *envSecretCache) get(ctx context.Context, kubeClient kubernetes.Interface, namespace, name string) (*corev1.Secret, error) {
-	key := envSecretCacheKey{
-		namespace: namespace,
-		name:      name,
-	}
-	now := time.Now()
-
-	c.mu.RLock()
-	entry, ok := c.entries[key]
-	if ok && now.Before(entry.expiresAt) {
-		secret := entry.secret.DeepCopy()
-		c.mu.RUnlock()
-		return secret, nil
-	}
-	c.mu.RUnlock()
-
-	// TODO: Make refresh smarter if this pattern sticks, for example by
-	// refreshing asynchronously or watching referenced Secrets.
-	secret, err := kubeClient.CoreV1().Secrets(namespace).Get(ctx, name, metav1.GetOptions{})
-	if err != nil {
-		return nil, err
-	}
-
-	secret = secret.DeepCopy()
-	c.mu.Lock()
-	c.entries[key] = envSecretCacheEntry{
-		secret:    secret,
-		expiresAt: time.Now().Add(c.ttl),
-	}
-	c.mu.Unlock()
-
-	return secret.DeepCopy(), nil
-}
-
-func isOptional(optional *bool) bool {
-	return optional != nil && *optional
-}