fix(k8s): restore executor profile environment on resume (#3835)

2026-09-21 11:18:08 +01:00
parent db8131436f
commit 3ebd7381bb
9 changed files with 417 additions and 2 deletions
@@ -1530,7 +1530,11 @@ func (e *Executor) applyRecordedKubernetesExecutorConfigToResumeRequest(
 		ExecutorID: executorID, ExecutorType: string(current.Type), ExecutorCfg: current.Config,
 		Metadata: metadata, Resumable: current.Resumable, RuntimeName: string(current.Type),
 	}
+	if err := e.restoreKubernetesProfileEnvironment(ctx, &config, running.Metadata); err != nil {
+		return executorConfig{}, err
+	}
 	session.ExecutorID = executorID
+	session.ExecutorProfileID, _ = running.Metadata[lifecycle.MetadataKeyExecutorProfileID].(string)
 	req.ExecutorType = config.ExecutorType
 	req.ExecutorConfig = config.ExecutorCfg
 	req.Metadata = metadata
@@ -0,0 +1,36 @@
+package executor
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/repository/repoerrors"
+)
+
+// Reload environment sources independently of the immutable workload snapshot.
+// Secret references remain unresolved until lifecycle's launch checkpoint.
+func (e *Executor) restoreKubernetesProfileEnvironment(
+	ctx context.Context, config *executorConfig, recorded map[string]interface{},
+) error {
+	profileID, _ := recorded["executor_profile_id"].(string)
+	if profileID == "" {
+		return nil
+	}
+	profile, err := e.repo.GetExecutorProfile(ctx, profileID)
+	if errors.Is(err, repoerrors.ErrExecutorProfileNotFound) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("resume Kubernetes executor: load recorded profile environment: %w", err)
+	}
+	if profile == nil {
+		return nil
+	}
+	if profile.ExecutorID != config.ExecutorID {
+		return errors.New("resume Kubernetes executor: recorded profile belongs to a different executor")
+	}
+	config.ProfileEnvVars = append([]models.ProfileEnvVar(nil), profile.EnvVars...)
+	return nil
+}
@@ -0,0 +1,109 @@
+package executor
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+
+	runtimeenv "github.com/kandev/kandev/internal/agent/runtime/environment"
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/agentruntime"
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/repository/repoerrors"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+func TestKubernetesResumeRestoresRecordedProfileEnvironment(t *testing.T) {
+	repo := newMockRepository()
+	setupLiveResumeTestFixture(repo)
+	session := repo.sessions["sess-1"]
+	session.ExecutorProfileID = "untrusted-profile"
+	repo.executors["recorded-kubernetes-executor"] = &models.Executor{ID: "recorded-kubernetes-executor", Type: models.ExecutorTypeKubernetes, Config: map[string]string{"auth_mode": "kubeconfig", "kubeconfig_path": "/tmp/test-config", "namespace": "kandev-agents"}}
+	repo.executorsRunning[session.ID] = &models.ExecutorRunning{
+		ID: session.ID, SessionID: session.ID, TaskID: session.TaskID,
+		ExecutorID: "recorded-kubernetes-executor", Runtime: agentruntime.RuntimeKubernetes, Metadata: recordedKubernetesResumeMetadata(),
+	}
+	repo.executorProfiles["recorded-profile"] = &models.ExecutorProfile{
+		ID: "recorded-profile", ExecutorID: "recorded-kubernetes-executor",
+		Config:  map[string]string{"pod_template_yaml": "edited template must not replace snapshot"},
+		EnvVars: []models.ProfileEnvVar{{Key: "PYTHONPATH", Value: "/artifacts/runtime"}, {Key: "INTEL_DATABASE_URL", SecretID: "database-secret"}},
+	}
+	repo.executorProfiles["untrusted-profile"] = &models.ExecutorProfile{
+		ID: "untrusted-profile", ExecutorID: "recorded-kubernetes-executor", EnvVars: []models.ProfileEnvVar{{Key: "INJECTED", Value: "bad"}},
+	}
+	exec := newTestExecutor(t, &mockAgentManager{}, repo)
+	req, _, _, _, _, err := exec.buildResumeRequest(context.Background(), &v1.Task{ID: session.TaskID, WorkspaceID: "workspace-1"}, session, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []runtimeenv.Definition
+	for _, def := range req.EnvironmentDefinitions {
+		if def.Origin == runtimeenv.OriginExecutorProfile {
+			got = append(got, def)
+		}
+	}
+	want := []runtimeenv.Definition{
+		{Key: "PYTHONPATH", Literal: "/artifacts/runtime", Origin: runtimeenv.OriginExecutorProfile},
+		{Key: "INTEL_DATABASE_URL", SecretID: "database-secret", Origin: runtimeenv.OriginExecutorProfile},
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("profile environment = %#v, want %#v", got, want)
+	}
+	if req.Metadata[lifecycle.MetadataKeyKubernetesProfileSnapshot] != recordedKubernetesResumeMetadata()[lifecycle.MetadataKeyKubernetesProfileSnapshot] {
+		t.Fatal("recorded workload snapshot changed")
+	}
+}
+
+func TestKubernetesResumeProfileEnvironmentFailures(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		profile   *models.ExecutorProfile
+		lookupErr error
+		wantError bool
+	}{
+		{name: "deleted profile", lookupErr: repoerrors.ErrExecutorProfileNotFound},
+		{name: "absent profile"},
+		{name: "database failure", lookupErr: errors.New("database unavailable"), wantError: true},
+		{name: "different executor", profile: &models.ExecutorProfile{ExecutorID: "other"}, wantError: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			repo := &resumeProfileRepository{mockRepository: newMockRepository(), profile: tc.profile, err: tc.lookupErr}
+			exec := newTestExecutor(t, &mockAgentManager{}, repo.mockRepository)
+			exec.repo = repo
+			config := executorConfig{ExecutorID: "recorded"}
+			err := exec.restoreKubernetesProfileEnvironment(context.Background(), &config, map[string]interface{}{lifecycle.MetadataKeyExecutorProfileID: "profile"})
+			if (err != nil) != tc.wantError {
+				t.Fatalf("error = %v, want error %v", err, tc.wantError)
+			}
+			if len(config.ProfileEnvVars) != 0 {
+				t.Fatal("loaded environment from unavailable or foreign profile")
+			}
+		})
+	}
+}
+
+func TestKubernetesResumeWithoutRecordedProfileSkipsLookup(t *testing.T) {
+	for _, recorded := range []map[string]interface{}{nil, {}, {lifecycle.MetadataKeyExecutorProfileID: ""}} {
+		repo := &resumeProfileRepository{mockRepository: newMockRepository(), err: errors.New("unexpected profile lookup")}
+		exec := newTestExecutor(t, &mockAgentManager{}, repo.mockRepository)
+		exec.repo = repo
+		config := executorConfig{ExecutorID: "recorded"}
+		if err := exec.restoreKubernetesProfileEnvironment(context.Background(), &config, recorded); err != nil {
+			t.Fatalf("resume without a recorded profile: %v", err)
+		}
+		if len(config.ProfileEnvVars) != 0 {
+			t.Fatal("loaded environment without a recorded profile")
+		}
+	}
+}
+
+type resumeProfileRepository struct {
+	*mockRepository
+	profile *models.ExecutorProfile
+	err     error
+}
+
+func (r *resumeProfileRepository) GetExecutorProfile(context.Context, string) (*models.ExecutorProfile, error) {
+	return r.profile, r.err
+}
@@ -0,0 +1,69 @@
+package executor
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/agentruntime"
+	"github.com/kandev/kandev/internal/task/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+func TestKubernetesResumePreservesProfileForExistingWorkspaceStart(t *testing.T) {
+	ctx := context.Background()
+	repo := newMockRepository()
+	setupLiveResumeTestFixture(repo)
+	session := cloneMockTaskSession(repo.sessions["sess-1"])
+	session.ExecutorProfileID = "untrusted-profile"
+	repo.sessions[session.ID] = cloneMockTaskSession(session)
+	repo.executors["recorded-kubernetes-executor"] = &models.Executor{ID: "recorded-kubernetes-executor", Type: models.ExecutorTypeKubernetes, Config: map[string]string{"auth_mode": "kubeconfig", "kubeconfig_path": "/tmp/test-config", "namespace": "kandev-agents"}}
+	repo.executorsRunning[session.ID] = &models.ExecutorRunning{
+		SessionID: session.ID, TaskID: session.TaskID, ExecutorID: "recorded-kubernetes-executor",
+		Runtime: agentruntime.RuntimeKubernetes, Metadata: recordedKubernetesResumeMetadata(),
+	}
+	repo.executorProfiles["recorded-profile"] = &models.ExecutorProfile{
+		ID: "recorded-profile", ExecutorID: "recorded-kubernetes-executor",
+		EnvVars: []models.ProfileEnvVar{{Key: "PYTHONPATH", Value: "/runtime"}, {Key: "DATABASE_URL", SecretID: "database"}},
+	}
+	repo.executorProfiles["untrusted-profile"] = &models.ExecutorProfile{
+		ID: "untrusted-profile", EnvVars: []models.ProfileEnvVar{{Key: "PYTHONPATH", Value: "/wrong"}, {Key: "INJECTED", Value: "bad"}},
+	}
+	profileManager := &lifecycle.Manager{}
+	profileManager.SetExecutorProfileReader(&resumeEnvironmentProfileReader{repo})
+	profileManager.SetSecretStore(&mockSecretStore{secrets: map[string]string{"database": "synthetic-database"}})
+	var delivered map[string]string
+	manager := &mockAgentManager{
+		executorProfileEnvFunc: profileManager.ExecutorProfileEnvForSession,
+		setExecutionEnvFunc: func(_ context.Context, _ string, env map[string]string) error {
+			delivered = env
+			return nil
+		},
+	}
+	exec := newTestExecutor(t, manager, repo)
+	task := &v1.Task{ID: session.TaskID, WorkspaceID: "workspace-1"}
+	if _, _, _, _, _, err := exec.buildResumeRequest(ctx, task, session, true); err != nil {
+		t.Fatal(err)
+	}
+	if err := exec.persistSessionFullRowIfCurrentState(ctx, session, session.State); err != nil {
+		t.Fatal(err)
+	}
+	reloaded, err := repo.GetTaskSession(ctx, session.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	request := &LaunchAgentRequest{ExecutorType: string(models.ExecutorTypeKubernetes)}
+	if err := exec.configureExistingWorkspace(ctx, task, reloaded, "execution", "", request); err != nil {
+		t.Fatal(err)
+	}
+	want := map[string]string{"PYTHONPATH": "/runtime", "DATABASE_URL": "synthetic-database"}
+	if delivered["PYTHONPATH"] != want["PYTHONPATH"] || delivered["DATABASE_URL"] != want["DATABASE_URL"] || delivered["INJECTED"] != "" {
+		t.Fatalf("next subprocess environment = %#v, want %#v", delivered, want)
+	}
+}
+
+type resumeEnvironmentProfileReader struct{ *mockRepository }
+
+func (*resumeEnvironmentProfileReader) HasActiveTaskResourceCleanupJob(context.Context, string) (bool, error) {
+	return false, nil
+}