diff --git a/apps/backend/AGENTS.md b/apps/backend/AGENTS.md index fa22577edc..8256f0930a 100644 --- a/apps/backend/AGENTS.md +++ b/apps/backend/AGENTS.md @@ -72,7 +72,7 @@ apps/backend/ │ │ └── secretadapter/ # Upsert-style adapter over secrets.SecretStore │ ├── i18n/ # Localization for backend-rendered browser/share artifacts │ ├── jira/ # Jira/Atlassian Cloud integration (config, REST client, poller) -│ ├── kubernetes/ # Kubernetes diagnostics and exact recorded-session status API +│ ├── kubernetes/ # Task-owned Pod/PVC diagnostics; session stop preserves compute. See docs/specs/executors/system-design/kubernetes-task-pod.md for ownership, credentials, and recovery invariants. │ ├── linear/ # Linear integration (config, GraphQL client, poller) │ ├── lsp/ # LSP server │ ├── mcp/ # MCP protocol support @@ -175,7 +175,7 @@ Standalone agentctl is launched in its own process group so terminal Ctrl+C is h - `k8s` - Namespaced Kubernetes Pod with optional PVC workspace - `remote_docker`, `remote_vps` - Planned -**Kubernetes lifecycle:** `executors_running` is the authoritative resource inventory. Persist the exact Pod/PVC names, UIDs, full `kandev.ai/*` identity, workload snapshot, and internal runtime-secret references before reporting a launch as durable. Ordinary stop and backend shutdown preserve resources; terminal cleanup deletes the Pod and only a Kandev-created PVC after exact identity checks and confirmed absence. Reconnect uses the current executor connection config but the recorded workload/resource snapshot, and any ambiguity fails closed. Keep agentctl reachable only through a process-local loopback port-forward; never add a Service or place resolved credentials in a Pod spec. +**Kubernetes lifecycle:** `task_environment_kubernetes` owns shared physical Pod/PVC inventory; `executors_running` records individual sessions and legacy session-owned pods. Persist the exact Pod/PVC names, UIDs, full `kandev.ai/*` identity, workload snapshot, and internal runtime-secret references before reporting a launch as durable. Session stop deletes only its agentctl instance, and backend shutdown preserves resources; task cleanup deletes the Pod and only a Kandev-created PVC after exact identity checks and confirmed absence. Reconnect uses the current executor connection config but the recorded workload/resource snapshot, and any ambiguity fails closed. Keep agentctl reachable only through a process-local loopback port-forward; never add a Service or place resolved credentials in a Pod spec. **Remote SSH executor platforms:** Treat supported remote OS/arch values as an end-to-end contract. Platform probe/normalization, lifecycle support checks, agentctl helper resolution, platform default shell, SSH readiness endpoints, frontend response types, and tests must stay aligned. Preserve raw unsupported platform details in user-facing errors, but use normalized values for supported-platform matching. Keep shell defaults platform-aware: Darwin defaults to `zsh`, Linux defaults to `bash`, unless an explicit shell is saved. diff --git a/apps/backend/internal/agent/kubernetes/compose.go b/apps/backend/internal/agent/kubernetes/compose.go index 3d0743ff0f..9a8490cb50 100644 --- a/apps/backend/internal/agent/kubernetes/compose.go +++ b/apps/backend/internal/agent/kubernetes/compose.go @@ -12,6 +12,9 @@ import ( // Runtime inventory metadata keys are shared by lifecycle persistence and // higher-level fail-closed inventory checks. const ( + MetadataKeyTaskOwned = "kubernetes_task_owned" + MetadataKeyOwnershipVersion = "kubernetes_resource_ownership_version" + TaskOwnershipVersion = "task-v1" MetadataKeyResourceExecutorID = "kubernetes_resource_executor_id" MetadataKeyResourceProfileID = "kubernetes_resource_profile_id" MetadataKeyResourceInstanceID = "kubernetes_resource_instance_id" @@ -21,6 +24,7 @@ const ( ) type ResourceIdentity struct { + TaskOwned bool ExecutorID string ProfileID string InstanceID string @@ -98,6 +102,10 @@ func OwnershipLabels(identity ResourceIdentity) (map[string]string, error) { "app.kubernetes.io/managed-by": "kandev", "app.kubernetes.io/instance": identity.InstanceID, } + if identity.TaskOwned { + labels["kandev.ai/ownership-version"] = TaskOwnershipVersion + labels["app.kubernetes.io/component"] = "task-environment" + } for _, item := range values { if item.value == "" || len(validation.IsValidLabelValue(item.value)) > 0 { return nil, fieldError(item.path, "must be a non-empty Kubernetes label value") diff --git a/apps/backend/internal/agent/runtime/agentctl/control.go b/apps/backend/internal/agent/runtime/agentctl/control.go index 055932bae1..3f9074018e 100644 --- a/apps/backend/internal/agent/runtime/agentctl/control.go +++ b/apps/backend/internal/agent/runtime/agentctl/control.go @@ -344,6 +344,9 @@ func (c *ControlClient) DeleteInstance(ctx context.Context, instanceID string) e return nil } +// ErrInstanceNotFound identifies an absent agentctl instance. +var ErrInstanceNotFound = errors.New("instance not found") + // GetInstance gets information about a specific instance. func (c *ControlClient) GetInstance(ctx context.Context, instanceID string) (*InstanceInfo, error) { req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/instances/"+instanceID, nil) @@ -358,7 +361,7 @@ func (c *ControlClient) GetInstance(ctx context.Context, instanceID string) (*In defer func() { _ = resp.Body.Close() }() if resp.StatusCode == http.StatusNotFound { - return nil, fmt.Errorf("instance %q not found", instanceID) + return nil, fmt.Errorf("%w: %q", ErrInstanceNotFound, instanceID) } if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("failed to get instance: status %d", resp.StatusCode) diff --git a/apps/backend/internal/agent/runtime/agentctl/control_test.go b/apps/backend/internal/agent/runtime/agentctl/control_test.go index 77c58440b7..d02d103390 100644 --- a/apps/backend/internal/agent/runtime/agentctl/control_test.go +++ b/apps/backend/internal/agent/runtime/agentctl/control_test.go @@ -587,7 +587,7 @@ func TestGetInstance_FailureModes(t *testing.T) { body string wantErr string }{ - {"not found", http.StatusNotFound, `{}`, `instance "inst-3" not found`}, + {"not found", http.StatusNotFound, `{}`, `instance not found: "inst-3"`}, {"other status", http.StatusInternalServerError, `{}`, "failed to get instance: status 500"}, {"malformed body", http.StatusOK, `{"id":`, "failed to decode response"}, } @@ -600,6 +600,9 @@ func TestGetInstance_FailureModes(t *testing.T) { if err == nil || !strings.Contains(err.Error(), tc.wantErr) { t.Fatalf("error = %v, want %q", err, tc.wantErr) } + if errors.Is(err, ErrInstanceNotFound) != (tc.status == http.StatusNotFound) { + t.Errorf("not-found classification for status %d: %v", tc.status, err) + } if info != nil { t.Errorf("info = %+v, want nil on failure", info) } diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go b/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go index 63db4c4ece..54f2eb5263 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go @@ -343,34 +343,36 @@ var persistentMetadataKeys = map[string]bool{ MetadataKeySSHReclaimTaskDir: true, // Kubernetes connection and exact resource inventory. - MetadataKeyKubernetesAuthMode: true, - MetadataKeyKubernetesKubeconfigPath: true, - MetadataKeyKubernetesKubeContext: true, - MetadataKeyKubernetesConfigNamespace: true, - MetadataKeyKubernetesRequestTimeoutSeconds: true, - MetadataKeyKubernetesNamespace: true, - MetadataKeyKubernetesPodName: true, - MetadataKeyKubernetesPodUID: true, - MetadataKeyKubernetesMainContainer: true, - MetadataKeyKubernetesPlatform: true, - MetadataKeyKubernetesRuntimeWorkspaceMode: true, - MetadataKeyKubernetesPVCName: true, - MetadataKeyKubernetesPVCUID: true, - MetadataKeyKubernetesPVCCreated: true, - MetadataKeyKubernetesAgentctlRemotePort: true, - MetadataKeyKubernetesAgentctlInstanceID: true, - MetadataKeyKubernetesContainerRestartCount: true, - MetadataKeyKubernetesResourceInstanceID: true, - MetadataKeyKubernetesResourceExecutorID: true, - MetadataKeyKubernetesResourceProfileID: true, - MetadataKeyKubernetesResourceTaskID: true, - MetadataKeyKubernetesResourceSessionID: true, - MetadataKeyKubernetesResourceEnvironmentID: true, - MetadataKeyKubernetesExecutorConfigHash: true, - MetadataKeyKubernetesProfileConfigHash: true, - MetadataKeyKubernetesTemplateHash: true, - MetadataKeyKubernetesProfileSnapshot: true, - MetadataKeyKubernetesInventoryState: true, + MetadataKeyKubernetesAuthMode: true, + MetadataKeyKubernetesKubeconfigPath: true, + MetadataKeyKubernetesKubeContext: true, + MetadataKeyKubernetesConfigNamespace: true, + MetadataKeyKubernetesRequestTimeoutSeconds: true, + MetadataKeyKubernetesNamespace: true, + MetadataKeyKubernetesPodName: true, + MetadataKeyKubernetesPodUID: true, + MetadataKeyKubernetesMainContainer: true, + MetadataKeyKubernetesPlatform: true, + MetadataKeyKubernetesRuntimeWorkspaceMode: true, + MetadataKeyKubernetesPVCName: true, + MetadataKeyKubernetesPVCUID: true, + MetadataKeyKubernetesPVCCreated: true, + MetadataKeyKubernetesAgentctlRemotePort: true, + MetadataKeyKubernetesAgentctlInstanceID: true, + MetadataKeyKubernetesContainerRestartCount: true, + metadataKubernetesTaskOwned: true, + agentkubernetes.MetadataKeyOwnershipVersion: true, + MetadataKeyKubernetesResourceInstanceID: true, + MetadataKeyKubernetesResourceExecutorID: true, + MetadataKeyKubernetesResourceProfileID: true, + MetadataKeyKubernetesResourceTaskID: true, + MetadataKeyKubernetesResourceSessionID: true, + MetadataKeyKubernetesResourceEnvironmentID: true, + MetadataKeyKubernetesExecutorConfigHash: true, + MetadataKeyKubernetesProfileConfigHash: true, + MetadataKeyKubernetesTemplateHash: true, + MetadataKeyKubernetesProfileSnapshot: true, + MetadataKeyKubernetesInventoryState: true, // Executor type marker MetadataKeyIsRemote: true, @@ -417,42 +419,44 @@ var sessionScopedMetadataKeys = map[string]bool{ // must not be inherited by a sibling session sharing the environment. MetadataKeyOfficeAgentProfileID: true, - MetadataKeySSHRemoteSessionDir: true, - MetadataKeySSHRemoteAgentctlPort: true, - MetadataKeySSHRemoteAgentctlPID: true, - MetadataKeySSHAgentctlInstanceID: true, - MetadataKeySSHLocalForwardPort: true, - MetadataKeySSHRemoteAgentctlURL: true, - MetadataKeySSHRuntimeAPILocalURL: true, - MetadataKeySSHRuntimeAPIRemotePort: true, - MetadataKeyKubernetesAuthMode: true, - MetadataKeyKubernetesKubeconfigPath: true, - MetadataKeyKubernetesKubeContext: true, - MetadataKeyKubernetesConfigNamespace: true, - MetadataKeyKubernetesRequestTimeoutSeconds: true, - MetadataKeyKubernetesNamespace: true, - MetadataKeyKubernetesPodName: true, - MetadataKeyKubernetesPodUID: true, - MetadataKeyKubernetesMainContainer: true, - MetadataKeyKubernetesPlatform: true, - MetadataKeyKubernetesRuntimeWorkspaceMode: true, - MetadataKeyKubernetesPVCName: true, - MetadataKeyKubernetesPVCUID: true, - MetadataKeyKubernetesPVCCreated: true, - MetadataKeyKubernetesAgentctlRemotePort: true, - MetadataKeyKubernetesAgentctlInstanceID: true, - MetadataKeyKubernetesContainerRestartCount: true, - MetadataKeyKubernetesResourceInstanceID: true, - MetadataKeyKubernetesResourceExecutorID: true, - MetadataKeyKubernetesResourceProfileID: true, - MetadataKeyKubernetesResourceTaskID: true, - MetadataKeyKubernetesResourceSessionID: true, - MetadataKeyKubernetesResourceEnvironmentID: true, - MetadataKeyKubernetesExecutorConfigHash: true, - MetadataKeyKubernetesProfileConfigHash: true, - MetadataKeyKubernetesTemplateHash: true, - MetadataKeyKubernetesProfileSnapshot: true, - MetadataKeyKubernetesInventoryState: true, + MetadataKeySSHRemoteSessionDir: true, + MetadataKeySSHRemoteAgentctlPort: true, + MetadataKeySSHRemoteAgentctlPID: true, + MetadataKeySSHAgentctlInstanceID: true, + MetadataKeySSHLocalForwardPort: true, + MetadataKeySSHRemoteAgentctlURL: true, + MetadataKeySSHRuntimeAPILocalURL: true, + MetadataKeySSHRuntimeAPIRemotePort: true, + MetadataKeyKubernetesAuthMode: true, + MetadataKeyKubernetesKubeconfigPath: true, + MetadataKeyKubernetesKubeContext: true, + MetadataKeyKubernetesConfigNamespace: true, + MetadataKeyKubernetesRequestTimeoutSeconds: true, + MetadataKeyKubernetesNamespace: true, + MetadataKeyKubernetesPodName: true, + MetadataKeyKubernetesPodUID: true, + MetadataKeyKubernetesMainContainer: true, + MetadataKeyKubernetesPlatform: true, + MetadataKeyKubernetesRuntimeWorkspaceMode: true, + MetadataKeyKubernetesPVCName: true, + MetadataKeyKubernetesPVCUID: true, + MetadataKeyKubernetesPVCCreated: true, + MetadataKeyKubernetesAgentctlRemotePort: true, + MetadataKeyKubernetesAgentctlInstanceID: true, + MetadataKeyKubernetesContainerRestartCount: true, + metadataKubernetesTaskOwned: true, + agentkubernetes.MetadataKeyOwnershipVersion: true, + MetadataKeyKubernetesResourceInstanceID: true, + MetadataKeyKubernetesResourceExecutorID: true, + MetadataKeyKubernetesResourceProfileID: true, + MetadataKeyKubernetesResourceTaskID: true, + MetadataKeyKubernetesResourceSessionID: true, + MetadataKeyKubernetesResourceEnvironmentID: true, + MetadataKeyKubernetesExecutorConfigHash: true, + MetadataKeyKubernetesProfileConfigHash: true, + MetadataKeyKubernetesTemplateHash: true, + MetadataKeyKubernetesProfileSnapshot: true, + MetadataKeyKubernetesInventoryState: true, } // ShouldPersistMetadataKey returns true if the given metadata key should diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go b/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go index 445bd44f9f..f25917f788 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_backend_test.go @@ -39,6 +39,8 @@ func TestShouldPersistMetadataKey(t *testing.T) { func TestKubernetesRuntimeMetadataKeysPersistWithoutLocalForward(t *testing.T) { persistent := []string{ + "kubernetes_task_owned", + "kubernetes_resource_ownership_version", "auth_mode", "kubeconfig_path", "kube_context", diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes.go index 058ebd29aa..597184a153 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes.go @@ -20,6 +20,7 @@ import ( agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" "github.com/kandev/kandev/internal/agentctl/server/process" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/secrets" "github.com/kandev/kandev/internal/task/models" ) @@ -46,6 +47,8 @@ var errKubernetesLifecycleRequestIncomplete = errors.New("kubernetes lifecycle r // KubernetesExecutor owns Kubernetes Pod/PVC lifecycle and process-local // forwards. Cluster clients are initialized lazily from the selected executor. type KubernetesExecutor struct { + environmentStore KubernetesEnvironmentStore + secretStore secrets.SecretStore agentctlResolver *AgentctlResolver logger *logger.Logger clientFactory kubernetesRuntimeClientFactory @@ -53,9 +56,10 @@ type KubernetesExecutor struct { healthRetryDelay time.Duration launchTimingClock func() time.Time - mu sync.Mutex - sessions map[string]*kubernetesSession - locks map[string]*kubernetesInstanceLock + mu sync.Mutex + sessions map[string]*kubernetesSession + locks map[string]*kubernetesInstanceLock + pendingControlTokens map[string]string } type kubernetesInstanceLock struct { @@ -107,6 +111,13 @@ func (r *KubernetesExecutor) Name() executor.Name { return executor.NameKubernet func (r *KubernetesExecutor) HealthCheck(context.Context) error { return nil } func (r *KubernetesExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateRequest) (*ExecutorInstance, error) { + if req != nil && r.environmentStore != nil && req.TaskEnvironmentID != "" { + return r.createTaskInstance(ctx, req) + } + return r.createSessionInstance(ctx, req) +} + +func (r *KubernetesExecutor) createSessionInstance(ctx context.Context, req *ExecutorCreateRequest) (*ExecutorInstance, error) { if req == nil || req.InstanceID == "" { return nil, errKubernetesLifecycleRequestIncomplete } @@ -236,7 +247,7 @@ func newKubernetesFreshLaunch( if err != nil { return nil, err } - podName, pvcName := kubernetesResourceNames(req.InstanceID) + podName, pvcName := kubernetesResourceNames(identity.InstanceID) return &kubernetesFreshLaunch{ executor: executor, runtime: runtime, req: req, executorConfig: executorConfig, profile: profile, identity: identity, podName: podName, pvcName: pvcName, @@ -307,6 +318,7 @@ func (l *kubernetesFreshLaunch) complete( l.executorConfig, l.profile, l.identity, runningPod, l.workspace.claim, l.workspace.createdClaim != nil, remotePort, KubernetesInventoryStateReady, ) + metadata[MetadataKeyKubernetesAgentctlInstanceID] = l.req.InstanceID // A fresh lifecycle-managed launch already has provisional Pod/PVC inventory. // Manager persists the final ready row only after both required runtime secret // references are durable, avoiding a restart-visible ready row that cannot @@ -639,6 +651,11 @@ func kubernetesIdentity(req *ExecutorCreateRequest) (kubeexecutor.ResourceIdenti ExecutorID: getMetadataString(req.Metadata, "executor_id"), ProfileID: getMetadataString(req.Metadata, MetadataKeyExecutorProfileID), InstanceID: req.InstanceID, TaskID: req.TaskID, SessionID: req.SessionID, EnvironmentID: req.TaskEnvironmentID, } + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + identity.TaskOwned = true + identity.InstanceID = req.TaskEnvironmentID + } + if _, err := kubeexecutor.OwnershipLabels(identity); err != nil { return kubeexecutor.ResourceIdentity{}, err } @@ -681,6 +698,11 @@ func kubernetesRuntimeMetadata( "executor_id": identity.ExecutorID, MetadataKeyExecutorProfileID: identity.ProfileID, } + if identity.TaskOwned { + metadata[metadataKubernetesTaskOwned] = true + metadata[kubeexecutor.MetadataKeyOwnershipVersion] = kubeexecutor.TaskOwnershipVersion + } + if pod != nil { metadata[MetadataKeyKubernetesNamespace] = pod.Namespace metadata[MetadataKeyKubernetesPodName] = pod.Name diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_bootstrap.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_bootstrap.go index 8f5526e1cb..c1d906bcd4 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_bootstrap.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_bootstrap.go @@ -56,6 +56,12 @@ func (r *KubernetesExecutor) bootstrapPod( } authEnv := kubernetesAuthEnvironment(req.Env, nonce) + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + authEnv["HOME"] = kubernetesSessionHome(req) + if err := r.writeKubernetesSessionAuth(ctx, runtime, req, pod, profile.MainContainer); err != nil { + return err + } + } authData, err := kubernetesSerializeEnvironment(authEnv) if err != nil { return err @@ -70,6 +76,15 @@ func (r *KubernetesExecutor) bootstrapPod( if err := runKubernetesPrepareScript(ctx, runtime.streams, pod, profile.MainContainer); err != nil { return err } + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + controlEnv, err := kubernetesSerializeEnvironment(kubernetesAuthEnvironment(nil, nonce)) + if err != nil { + return err + } + if err := kubernetesWriteFile(ctx, runtime.streams, pod, profile.MainContainer, kubernetesAuthEnvPath, controlEnv, 0o600); err != nil { + return err + } + } if err := kubernetesWriteFile(ctx, runtime.streams, pod, profile.MainContainer, kubernetesStartPath, nil, 0o600); err != nil { return fmt.Errorf("kubernetes lifecycle: signal bootstrap: %w", err) @@ -163,6 +178,12 @@ func kubernetesRuntimeEnvironment(req *ExecutorCreateRequest) map[string]string kubernetesEnvAgentProfile: req.OfficeAgentProfileID, kubernetesEnvExecutionProfile: req.AgentProfileID, } + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + values[kubernetesEnvInstanceID] = req.TaskEnvironmentID + delete(values, kubernetesEnvSessionID) + delete(values, kubernetesEnvAgentProfile) + delete(values, kubernetesEnvExecutionProfile) + } if selectedCheckoutIsPullRequest(req.Metadata) { values[selectedCheckoutMarker] = "1" } diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_cleanup.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_cleanup.go index ae42473af9..13adf8e2cc 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_cleanup.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_cleanup.go @@ -22,6 +22,12 @@ func (r *KubernetesExecutor) StopInstance(ctx context.Context, instance *Executo if instance == nil { return nil } + if err := r.resolveTaskKubernetesStopOwnership(ctx, instance); err != nil { + return err + } + if getMetadataBool(instance.Metadata, metadataKubernetesTaskOwned) { + return r.stopSharedKubernetesInstance(ctx, instance) + } unlock := r.lockInstance(instance.InstanceID) defer unlock() @@ -283,6 +289,7 @@ func kubernetesCleanupInventory( req := &ExecutorCreateRequest{ TaskID: instance.TaskID, SessionID: instance.SessionID, + TaskEnvironmentID: getMetadataString(instance.Metadata, MetadataKeyKubernetesResourceEnvironmentID), PreviousExecutionID: getMetadataString(metadata, MetadataKeyKubernetesResourceInstanceID), Metadata: metadata, } diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_control_recovery.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_control_recovery.go new file mode 100644 index 0000000000..4eaa12b008 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_control_recovery.go @@ -0,0 +1,51 @@ +package lifecycle + +import ( + "context" + "errors" + "github.com/kandev/kandev/internal/secrets" +) + +// Keep the single-use handshake result until storage acknowledges it. All callers +// hold the environment's control lock, so a sibling retries the same credential. +func (r *KubernetesExecutor) persistSharedKubernetesControlToken(ctx context.Context, id, token string) error { + r.mu.Lock() + if r.pendingControlTokens == nil { + r.pendingControlTokens = make(map[string]string) + } + r.pendingControlTokens[id] = token + r.mu.Unlock() + persistCtx, cancel := kubernetesDurableContext(ctx) + defer cancel() + if err := r.secretStore.Update(persistCtx, id, &secrets.UpdateSecretRequest{Value: &token}); err != nil { + return errors.Join(err, r.saveKubernetesControlRecovery(persistCtx, id, token)) + } + if err := r.secretStore.Delete(persistCtx, kubernetesControlRecoverySecretID(id)); err != nil && !errors.Is(err, secrets.ErrNotFound) { + return err + } + r.mu.Lock() + delete(r.pendingControlTokens, id) + r.mu.Unlock() + return nil +} + +func kubernetesControlRecoverySecretID(id string) string { + if id == "" { + return "" + } + return id + ":recovery" +} + +// A failed canonical update must not discard the credential after nonce +// consumption. The deterministic encrypted recovery record survives restart. +func (r *KubernetesExecutor) saveKubernetesControlRecovery(ctx context.Context, id, token string) error { + recoveryID := kubernetesControlRecoverySecretID(id) + _, err := r.secretStore.Get(ctx, recoveryID) + if errors.Is(err, secrets.ErrNotFound) { + return r.secretStore.Create(ctx, &secrets.SecretWithValue{Secret: secrets.Secret{ID: recoveryID, Name: recoveryID}, Value: token}) + } + if err != nil { + return err + } + return r.secretStore.Update(ctx, recoveryID, &secrets.UpdateSecretRequest{Value: &token}) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_environment.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_environment.go new file mode 100644 index 0000000000..725b901423 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_environment.go @@ -0,0 +1,275 @@ +package lifecycle + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/google/uuid" + "github.com/kandev/kandev/internal/agent/executor" + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + "github.com/kandev/kandev/internal/secrets" + "github.com/kandev/kandev/internal/task/models" +) + +const metadataKubernetesTaskOwned = kubeexecutor.MetadataKeyTaskOwned + +// KubernetesEnvironmentStore separates physical inventory from session execution rows. +type KubernetesEnvironmentStore interface { + GetExecutor(context.Context, string) (*models.Executor, error) + GetTaskEnvironment(context.Context, string) (*models.TaskEnvironment, error) + GetKubernetesEnvironment(context.Context, string) (*models.KubernetesEnvironment, error) + ClaimKubernetesEnvironment(context.Context, string, string, int64, string) (*models.KubernetesEnvironment, error) + ClaimKubernetesEnvironmentCleanup(context.Context, string, string, int64, string) (*models.KubernetesEnvironment, error) + SaveKubernetesEnvironment(context.Context, *models.KubernetesEnvironment, bool) error + DeleteKubernetesEnvironment(context.Context, *models.KubernetesEnvironment) error + ListExecutorsRunningByTaskID(context.Context, string) ([]*models.ExecutorRunning, error) +} + +func (m *Manager) wireKubernetesEnvironmentStore() { + if m.executorRegistry == nil { + return + } + backend, err := m.executorRegistry.GetBackend(executor.NameKubernetes) + if err != nil { + return + } + runtime, ok := backend.(*KubernetesExecutor) + if !ok { + return + } + runtime.environmentStore, _ = m.runningWriter.(KubernetesEnvironmentStore) + runtime.secretStore = m.secretStore +} + +func (r *KubernetesExecutor) createTaskInstance(ctx context.Context, original *ExecutorCreateRequest) (_ *ExecutorInstance, returnedErr error) { + unlock := r.lockInstance("environment:" + original.TaskEnvironmentID) + defer unlock() + legacy, err := r.isUnadoptedKubernetesResume(ctx, original) + if err != nil { + return nil, err + } + if legacy { + return r.createSessionInstance(ctx, original) + } + record, err := r.claimTaskKubernetesEnvironment(ctx, original) + if err != nil { + return nil, err + } + defer func() { + if record.OperationID != "" { + persistCtx, cancel := kubernetesDurableContext(ctx) + defer cancel() + returnedErr = errors.Join(returnedErr, r.environmentStore.SaveKubernetesEnvironment(persistCtx, record, true)) + } + }() + req, err := r.taskKubernetesCreateRequest(ctx, original, record) + if err != nil { + return nil, err + } + instance, err := r.createSessionInstance(ctx, req) + if err != nil { + return nil, err + } + defer func() { + if returnedErr != nil { + rollbackCtx, cancel := kubernetesDurableContext(ctx) + defer cancel() + returnedErr = errors.Join(returnedErr, r.rollbackTaskKubernetesAttachment(rollbackCtx, instance)) + } + }() + instance.Metadata[metadataKubernetesTaskOwned] = true + instance.Metadata[MetadataKeyCleanupScript] = getMetadataString(req.Metadata, MetadataKeyCleanupScript) + record.Metadata = cloneKubernetesMetadata(instance.Metadata) + if err := r.persistKubernetesEnvironmentSecrets(ctx, record, instance); err != nil { + return nil, err + } + if err := r.environmentStore.SaveKubernetesEnvironment(ctx, record, true); err != nil { + return nil, err + } + return instance, nil +} + +func (r *KubernetesExecutor) adoptKubernetesEnvironment(ctx context.Context, req *ExecutorCreateRequest, record *models.KubernetesEnvironment) error { + rows, err := r.environmentStore.ListExecutorsRunningByTaskID(ctx, req.TaskID) + if err != nil { + return err + } + var candidate *models.ExecutorRunning + for _, row := range rows { + if row == nil || row.Runtime != executor.NameKubernetes { + continue + } + if getMetadataString(row.Metadata, MetadataKeyKubernetesResourceEnvironmentID) != req.TaskEnvironmentID { + return models.ErrWorkspaceReuseUnsafe + } + if getMetadataString(row.Metadata, MetadataKeyKubernetesPodUID) == "" { + return models.ErrWorkspaceReuseUnsafe + } + if candidate != nil && !kubernetesLegacyInventoriesAgree(candidate.Metadata, row.Metadata) { + return models.ErrWorkspaceReuseUnsafe + } + candidate = row + } + if candidate == nil { + return nil + } + record.Metadata = cloneKubernetesMetadata(candidate.Metadata) + record.ControlSecretID = getMetadataString(record.Metadata, MetadataKeyAuthTokenSecret) + record.BootstrapSecretID = getMetadataString(record.Metadata, MetadataKeyBootstrapNonceSecret) + return nil +} + +func (r *KubernetesExecutor) attachKubernetesEnvironmentRequest(ctx context.Context, req *ExecutorCreateRequest, record *models.KubernetesEnvironment) error { + if getMetadataString(req.Metadata, "executor_id") != getMetadataString(record.Metadata, MetadataKeyKubernetesResourceExecutorID) || getMetadataString(req.Metadata, MetadataKeyExecutorProfileID) != getMetadataString(record.Metadata, MetadataKeyKubernetesResourceProfileID) { + return models.ErrWorkspaceReuseUnsafe + } + if record.ControlSecretID == "" || record.BootstrapSecretID == "" { + return models.ErrWorkspaceReuseUnsafe + } + token, err := r.secretStore.Reveal(ctx, record.ControlSecretID) + if err != nil { + return err + } + nonce, err := r.secretStore.Reveal(ctx, record.BootstrapSecretID) + if err != nil { + return err + } + if strings.TrimSpace(token) == "" || strings.TrimSpace(nonce) == "" { + return models.ErrWorkspaceReuseUnsafe + } + remoteID := req.InstanceID + if req.PreviousExecutionID != "" { + if previousID := getMetadataString(req.Metadata, MetadataKeyKubernetesAgentctlInstanceID); previousID != "" { + remoteID = previousID + } + } + for key, value := range record.Metadata { + req.Metadata[key] = value + } + req.Metadata[metadataKubernetesTaskOwned] = true + // A resumed session retains its own instance independently of sibling inventory. + req.Metadata[MetadataKeyKubernetesAgentctlInstanceID] = remoteID + req.AuthToken = token + req.BootstrapNonce = nonce + return nil +} + +func (r *KubernetesExecutor) persistKubernetesEnvironmentSecrets(ctx context.Context, record *models.KubernetesEnvironment, instance *ExecutorInstance) error { + refs := []struct { + target *string + name, value, key string + }{ + {&record.ControlSecretID, "agentctl-auth", instance.AuthToken, MetadataKeyAuthTokenSecret}, + {&record.BootstrapSecretID, "agentctl-bootstrap", instance.BootstrapNonce, MetadataKeyBootstrapNonceSecret}, + } + for _, item := range refs { + if item.value == "" { + return errors.New("kubernetes environment control credential is empty") + } + *item.target = kubernetesRuntimeSecretID(record.EnvironmentID, item.name) + name := item.name + "-" + record.EnvironmentID + if _, err := r.secretStore.Get(ctx, *item.target); errors.Is(err, secrets.ErrNotFound) { + if err := r.secretStore.Create(ctx, &secrets.SecretWithValue{Secret: secrets.Secret{ID: *item.target, Name: name}, Value: item.value}); err != nil { + return err + } + } else if err != nil { + return err + } + instance.Metadata[item.key] = *item.target + record.Metadata[item.key] = *item.target + } + return nil +} + +func kubernetesLegacyInventoriesAgree(first, second map[string]interface{}) bool { + keys := []string{ + MetadataKeyKubernetesPodUID, MetadataKeyKubernetesPodName, MetadataKeyKubernetesNamespace, + MetadataKeyKubernetesPVCUID, MetadataKeyKubernetesPVCName, MetadataKeyKubernetesResourceExecutorID, + MetadataKeyKubernetesResourceProfileID, MetadataKeyKubernetesResourceInstanceID, + MetadataKeyKubernetesResourceTaskID, MetadataKeyKubernetesResourceSessionID, + MetadataKeyKubernetesResourceEnvironmentID, MetadataKeyKubernetesProfileSnapshot, + MetadataKeyKubernetesInventoryState, MetadataKeyAuthTokenSecret, MetadataKeyBootstrapNonceSecret, + } + for _, key := range keys { + if getMetadataString(first, key) != getMetadataString(second, key) { + return false + } + } + return getMetadataBool(first, MetadataKeyKubernetesPVCCreated) == getMetadataBool(second, MetadataKeyKubernetesPVCCreated) +} + +func (r *KubernetesExecutor) claimTaskKubernetesEnvironment(ctx context.Context, original *ExecutorCreateRequest) (*models.KubernetesEnvironment, error) { + env, err := r.environmentStore.GetTaskEnvironment(ctx, original.TaskEnvironmentID) + if err != nil { + return nil, fmt.Errorf("load Kubernetes task environment: %w", err) + } + if env == nil || env.TaskID != original.TaskID || env.ExecutorType != "k8s" { + return nil, models.ErrWorkspaceReuseUnsafe + } + if original.WorkspaceReuseRequired && env.Status != models.TaskEnvironmentStatusReady { + if env.Status == models.TaskEnvironmentStatusCreating { + return nil, models.ErrWorkspacePreparing + } + return nil, models.ErrWorkspaceReuseUnsafe + } + if r.secretStore == nil { + return nil, errors.New("kubernetes environment secret store is unavailable") + } + record, err := r.environmentStore.ClaimKubernetesEnvironment(ctx, env.ID, env.TaskID, env.OwnershipGeneration, uuid.NewString()) + if err != nil { + return nil, fmt.Errorf("%w: claim Kubernetes runtime: %w", models.ErrWorkspaceReuseUnsafe, err) + } + return record, nil +} + +func (r *KubernetesExecutor) taskKubernetesCreateRequest(ctx context.Context, original *ExecutorCreateRequest, record *models.KubernetesEnvironment) (*ExecutorCreateRequest, error) { + req := cloneKubernetesCreateRequest(original) + req.Metadata[metadataKubernetesTaskOwned] = true + req.Env = kubernetesSessionEnvironment(req) + if len(record.Metadata) == 0 { + if err := r.adoptKubernetesEnvironment(ctx, req, record); err != nil { + return nil, err + } + } + if len(record.Metadata) > 0 { + if err := r.attachKubernetesEnvironmentRequest(ctx, req, record); err != nil { + return nil, err + } + } else if req.WorkspaceReuseRequired { + return nil, models.ErrWorkspaceReuseUnsafe + } + previousCheckpoint := req.CheckpointRuntimeInventory + req.CheckpointRuntimeInventory = func(checkpointCtx context.Context, metadata map[string]interface{}) error { + metadata[metadataKubernetesTaskOwned] = true + metadata[MetadataKeyCleanupScript] = getMetadataString(req.Metadata, MetadataKeyCleanupScript) + record.Metadata = cloneKubernetesMetadata(metadata) + if err := r.environmentStore.SaveKubernetesEnvironment(checkpointCtx, record, false); err != nil { + return err + } + if previousCheckpoint != nil { + return previousCheckpoint(checkpointCtx, metadata) + } + return nil + } + return req, nil +} + +func (r *KubernetesExecutor) isUnadoptedKubernetesResume(ctx context.Context, req *ExecutorCreateRequest) (bool, error) { + if req.PreviousExecutionID == "" || getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) || getMetadataString(req.Metadata, MetadataKeyKubernetesPodUID) == "" { + return false, nil + } + _, err := r.environmentStore.GetKubernetesEnvironment(ctx, req.TaskEnvironmentID) + if !errors.Is(err, models.ErrKubernetesEnvironmentNotFound) { + return false, err + } + env, err := r.environmentStore.GetTaskEnvironment(ctx, req.TaskEnvironmentID) + if err != nil { + return false, err + } + if env == nil || env.TaskID != req.TaskID || env.ExecutorType != "k8s" { + return false, models.ErrWorkspaceReuseUnsafe + } + return true, nil +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_files.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_files.go index 6dc392f711..6851ade2bd 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_files.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_files.go @@ -68,7 +68,7 @@ func (r *KubernetesExecutor) materializeKubernetesCredentials( if req.AgentConfig == nil { return nil } - home := kubernetesAuthHomePath + home := kubernetesSessionHome(req) catalog := remoteauth.BuildCatalogForHost([]agents.Agent{req.AgentConfig}, runtime.GOOS, mustUserHome()) selectedMethods, err := kubernetesSelectedCredentialMethods(req.Metadata, catalog) if err != nil { @@ -87,7 +87,7 @@ func (r *KubernetesExecutor) materializeKubernetesCredentials( continue } command := "set -eu; export HOME=" + shellQuote(home) + - "; set -a; . " + shellQuote(kubernetesAuthEnvPath) + "; set +a; " + method.SetupScript + "; set -a; . " + shellQuote(kubernetesSessionAuthPath(req)) + "; set +a; " + method.SetupScript if execErr := runtimeClient.streams.Exec(ctx, kubeexecutor.ExecRequest{ Namespace: pod.Namespace, Pod: pod.Name, Container: container, Command: []string{"sh", "-c", command}, diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_reconnect.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_reconnect.go index 1b439a9b4a..b8ee374727 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_reconnect.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_reconnect.go @@ -92,7 +92,7 @@ func kubernetesRecordedInventory( if err != nil { return kubernetesRecordedState{}, kubeexecutor.ResourceIdentity{}, err } - if recorded.agentctlInstanceID != identity.InstanceID { + if !getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) && recorded.agentctlInstanceID != identity.InstanceID { return kubernetesRecordedState{}, kubeexecutor.ResourceIdentity{}, errors.New( "kubernetes lifecycle: recorded agentctl instance does not match resource identity", ) @@ -136,6 +136,7 @@ func kubernetesRecordedResourceIdentity( validateConnectionIdentity bool, ) (kubeexecutor.ResourceIdentity, error) { identity := kubeexecutor.ResourceIdentity{ + TaskOwned: getMetadataString(req.Metadata, kubeexecutor.MetadataKeyOwnershipVersion) == kubeexecutor.TaskOwnershipVersion, ExecutorID: getMetadataString(req.Metadata, MetadataKeyKubernetesResourceExecutorID), ProfileID: getMetadataString(req.Metadata, MetadataKeyKubernetesResourceProfileID), InstanceID: getMetadataString(req.Metadata, MetadataKeyKubernetesResourceInstanceID), @@ -153,7 +154,11 @@ func kubernetesRecordedResourceIdentity( if err := validateKubernetesRecordedIdentityField("task", req.TaskID, identity.TaskID); err != nil { return kubeexecutor.ResourceIdentity{}, err } - if err := validateKubernetesRecordedIdentityField("session", req.SessionID, identity.SessionID); err != nil { + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + if err := validateKubernetesRecordedIdentityField("environment", req.TaskEnvironmentID, identity.EnvironmentID); err != nil { + return kubeexecutor.ResourceIdentity{}, err + } + } else if err := validateKubernetesRecordedIdentityField("session", req.SessionID, identity.SessionID); err != nil { return kubeexecutor.ResourceIdentity{}, err } if validateConnectionIdentity { @@ -184,7 +189,9 @@ func kubernetesRequestWithRecordedIdentity( ) *ExecutorCreateRequest { recorded := *req recorded.TaskID = identity.TaskID - recorded.SessionID = identity.SessionID + if !getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + recorded.SessionID = identity.SessionID + } recorded.TaskEnvironmentID = identity.EnvironmentID return &recorded } @@ -257,6 +264,9 @@ func (r *KubernetesExecutor) reconnectKubernetesAgentctl( pod *corev1.Pod, recorded kubernetesRecordedState, ) (*agentctl.Client, kubeexecutor.PortForwardSession, string, int, error) { + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + return r.connectSharedKubernetesAgentctl(ctx, runtime, req, pod, recorded.agentctlInstanceID) + } if req.AuthToken == "" { return nil, nil, "", 0, errors.New("kubernetes lifecycle: recorded agentctl auth token is unavailable") } @@ -297,6 +307,9 @@ func (r *KubernetesExecutor) connectRestartedKubernetesAgentctl( pod *corev1.Pod, remoteInstanceID string, ) (*agentctl.Client, kubeexecutor.PortForwardSession, string, int, error) { + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + return r.connectSharedKubernetesAgentctl(ctx, runtime, req, pod, remoteInstanceID) + } controlForward, control, err := r.connectHealthyKubernetesControl(ctx, runtime, pod) if err != nil { return nil, nil, "", 0, err @@ -503,7 +516,7 @@ func ValidateKubernetesResumeMetadata( if _, err := kubeexecutor.ParseExecutorConfig(executorConfigValues); err != nil { return fmt.Errorf("kubernetes lifecycle: validate current executor config: %w", err) } - req := &ExecutorCreateRequest{TaskID: taskID, SessionID: sessionID, Metadata: metadata} + req := &ExecutorCreateRequest{TaskID: taskID, SessionID: sessionID, TaskEnvironmentID: getMetadataString(metadata, MetadataKeyKubernetesResourceEnvironmentID), Metadata: metadata} recorded, _, err := kubernetesRecordedInventory(req, true) if err != nil { return err diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_refresh.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_refresh.go index 055b1efff8..f7c9b03fb7 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_refresh.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_refresh.go @@ -67,7 +67,7 @@ func (r *KubernetesExecutor) RefreshRemoteInstance( Instance: &ExecutorInstance{ InstanceID: instance.InstanceID, TaskID: inspection.identity.TaskID, - SessionID: inspection.identity.SessionID, + SessionID: req.SessionID, RuntimeName: r.Name(), Client: client, WorkspacePath: kubernetesWorkspacePath, Metadata: metadata, AuthToken: token, BootstrapNonce: instance.BootstrapNonce, }, @@ -180,6 +180,9 @@ func kubernetesActiveRefreshRequest( req.InstanceID = instance.InstanceID req.TaskID = identity.TaskID req.SessionID = identity.SessionID + if getMetadataBool(instance.Metadata, metadataKubernetesTaskOwned) { + req.SessionID = instance.SessionID + } req.TaskEnvironmentID = identity.EnvironmentID req.Metadata = cloneKubernetesMetadata(instance.Metadata) req.AuthToken = instance.AuthToken @@ -194,6 +197,9 @@ func (r *KubernetesExecutor) connectKubernetesRefresh( req *ExecutorCreateRequest, inspection kubernetesRefreshInspection, ) (*agentctl.Client, kubeexecutor.PortForwardSession, string, int, error) { + if getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + return r.connectSharedKubernetesAgentctl(ctx, freshRuntime, req, inspection.pod, inspection.recorded.agentctlInstanceID) + } if inspection.restarted { return r.connectRestartedKubernetesAgentctl( ctx, freshRuntime, req, inspection.pod, inspection.recorded.agentctlInstanceID, diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_session_env.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_session_env.go new file mode 100644 index 0000000000..596f68fd0f --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_session_env.go @@ -0,0 +1,62 @@ +package lifecycle + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "path" + + "github.com/kandev/kandev/internal/githubauth" + corev1 "k8s.io/api/core/v1" +) + +func kubernetesSessionHome(req *ExecutorCreateRequest) string { + if !getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + return kubernetesAuthHomePath + } + sum := sha256.Sum256([]byte(req.SessionID)) + return path.Join("/run/kandev/sessions", hex.EncodeToString(sum[:16]), "home") +} + +func kubernetesSessionAuthPath(req *ExecutorCreateRequest) string { + if !getMetadataBool(req.Metadata, metadataKubernetesTaskOwned) { + return kubernetesAuthEnvPath + } + return path.Join(kubernetesSessionHome(req), "auth.env") +} + +func kubernetesSessionEnvironment(req *ExecutorCreateRequest) map[string]string { + env := make(map[string]string, len(req.Env)+8) + for key, value := range req.Env { + if !isKubernetesOwnedControlEnvironmentKey(key) { + env[key] = value + } + } + env["HOME"] = kubernetesSessionHome(req) + env[kubernetesEnvSessionID] = req.SessionID + env[kubernetesEnvTaskID] = req.TaskID + env[kubernetesEnvInstanceID] = req.InstanceID + env[kubernetesEnvEnvironmentID] = req.TaskEnvironmentID + env[kubernetesEnvAgentProfile] = req.OfficeAgentProfileID + env[kubernetesEnvExecutionProfile] = req.AgentProfileID + if hasManagedGitCredentialBrokerEnv(req.Env) { + env[githubauth.CredentialHelperPathEnv] = kubernetesAgentctlPath + } + return env +} + +func (r *KubernetesExecutor) writeKubernetesSessionAuth(ctx context.Context, runtime *kubernetesRuntimeClient, req *ExecutorCreateRequest, pod *corev1.Pod, container string) error { + data, err := kubernetesSerializeEnvironment(kubernetesSessionEnvironment(req)) + if err != nil { + return err + } + return kubernetesWriteFile(ctx, runtime.streams, pod, container, kubernetesSessionAuthPath(req), data, 0o600) +} + +func (r *KubernetesExecutor) prepareSharedKubernetesCredentials(ctx context.Context, runtime *kubernetesRuntimeClient, req *ExecutorCreateRequest, pod *corev1.Pod, container string) error { + if err := r.writeKubernetesSessionAuth(ctx, runtime, req, pod, container); err != nil { + return err + } + uploader := kubernetesPodFileUploader{streams: runtime.streams, pod: pod, container: container} + return r.materializeKubernetesCredentials(ctx, uploader, runtime, req, pod, container) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_shared_cleanup.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_shared_cleanup.go new file mode 100644 index 0000000000..9d81fefc79 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_shared_cleanup.go @@ -0,0 +1,174 @@ +package lifecycle + +import ( + "context" + "errors" + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + + "github.com/kandev/kandev/internal/task/models" + apierrors "k8s.io/apimachinery/pkg/api/errors" +) + +func (r *KubernetesExecutor) stopSharedKubernetesInstance(ctx context.Context, instance *ExecutorInstance) error { + unlock := r.lockInstance(instance.InstanceID) + defer unlock() + ctx, cancel := kubernetesDurableContext(ctx) + defer cancel() + session := r.closeSharedKubernetesSessions(instance) + if r.environmentStore == nil || r.secretStore == nil { + return errors.New("kubernetes task environment store is unavailable") + } + environmentID := getMetadataString(instance.Metadata, MetadataKeyKubernetesResourceEnvironmentID) + record, err := r.environmentStore.GetKubernetesEnvironment(ctx, environmentID) + if errors.Is(err, models.ErrKubernetesEnvironmentNotFound) { + return nil + } + if err != nil { + return err + } + if record.TaskID != instance.TaskID { + return models.ErrWorkspaceReuseUnsafe + } + + instance, err = r.taskKubernetesStopInstance(ctx, instance, record) + if err != nil { + return err + } + runtime, err := r.kubernetesCleanupRuntime(instance, session) + if err != nil { + return err + } + recorded, identity, err := kubernetesCleanupInventory(instance) + if err != nil { + return err + } + pod, err := runtime.resources.GetPod(ctx, recorded.namespace, recorded.podName) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return err + } + if err := verifyRecordedPod(pod, recorded.namespace, recorded.podName, recorded.podUID, identity); err != nil { + return err + } + token, err := r.secretStore.Reveal(ctx, record.ControlSecretID) + if err != nil { + return err + } + forward, control, err := r.connectHealthyKubernetesControl(ctx, runtime, pod) + if err != nil { + return err + } + defer func() { _ = forward.Close() }() + control.SetAuthToken(token) + return control.DeleteInstance(ctx, getMetadataString(instance.Metadata, MetadataKeyKubernetesAgentctlInstanceID)) +} + +// Rollback owns only the newly attached agent, even if credential persistence failed. +func (r *KubernetesExecutor) rollbackTaskKubernetesAttachment(ctx context.Context, instance *ExecutorInstance) error { + session := r.closeKubernetesSession(instance.InstanceID) + runtime, err := r.kubernetesCleanupRuntime(instance, session) + if err != nil { + return err + } + recorded, identity, err := kubernetesCleanupInventory(instance) + if err != nil { + return err + } + pod, err := runtime.resources.GetPod(ctx, recorded.namespace, recorded.podName) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return err + } + if err := verifyRecordedPod(pod, recorded.namespace, recorded.podName, recorded.podUID, identity); err != nil { + return err + } + forward, control, err := r.connectHealthyKubernetesControl(ctx, runtime, pod) + if err != nil { + return err + } + defer func() { _ = forward.Close() }() + control.SetAuthToken(instance.AuthToken) + return control.DeleteInstance(ctx, getMetadataString(instance.Metadata, MetadataKeyKubernetesAgentctlInstanceID)) +} + +func (r *KubernetesExecutor) resolveTaskKubernetesStopOwnership(ctx context.Context, instance *ExecutorInstance) error { + if getMetadataBool(instance.Metadata, metadataKubernetesTaskOwned) || r.environmentStore == nil { + return nil + } + envID := getMetadataString(instance.Metadata, MetadataKeyKubernetesResourceEnvironmentID) + if envID == "" { + return nil + } + record, err := r.environmentStore.GetKubernetesEnvironment(ctx, envID) + if errors.Is(err, models.ErrKubernetesEnvironmentNotFound) { + return nil + } + if err != nil { + return err + } + if record.TaskID != instance.TaskID || getMetadataString(record.Metadata, MetadataKeyKubernetesPodUID) != getMetadataString(instance.Metadata, MetadataKeyKubernetesPodUID) { + return models.ErrWorkspaceReuseUnsafe + } + instance.Metadata[metadataKubernetesTaskOwned] = true + return nil +} + +func (r *KubernetesExecutor) taskKubernetesStopInstance(ctx context.Context, instance *ExecutorInstance, record *models.KubernetesEnvironment) (*ExecutorInstance, error) { + current, err := r.environmentStore.GetExecutor(ctx, getMetadataString(record.Metadata, MetadataKeyKubernetesResourceExecutorID)) + if err != nil { + return nil, err + } + if current == nil || current.Type != models.ExecutorTypeKubernetes { + return nil, models.ErrWorkspaceReuseUnsafe + } + config, err := kubeexecutor.ParseExecutorConfig(current.Config) + if err != nil { + return nil, err + } + remoteID := getMetadataString(instance.Metadata, MetadataKeyKubernetesAgentctlInstanceID) + copyInstance := *instance + copyInstance.Metadata = overlayCurrentKubernetesConnectionMetadata(record.Metadata, current.Config, config) + copyInstance.Metadata[MetadataKeyKubernetesAgentctlInstanceID] = remoteID + return ©Instance, nil +} + +// A resumed execution can have a new local ID while agentctl retains the remote +// session ID. Release every local connection for that exact remote session. +func (r *KubernetesExecutor) closeSharedKubernetesSessions(instance *ExecutorInstance) *kubernetesSession { + r.mu.Lock() + var closing []*kubernetesSession + for id, session := range r.sessions { + if id == instance.InstanceID || matchesSharedKubernetesSession(session, instance) { + delete(r.sessions, id) + closing = append(closing, session) + } + } + r.mu.Unlock() + for _, session := range closing { + _ = closeKubernetesSessionResources(session) + } + if len(closing) == 0 { + return nil + } + // Every matching connection uses the same pod and executor configuration; + // any one supplies the fallback runtime client after its forward is closed. + return closing[0] +} + +func matchesSharedKubernetesSession(session *kubernetesSession, instance *ExecutorInstance) bool { + if session == nil || session.request == nil { + return false + } + req := session.request + remoteID := getMetadataString(req.Metadata, MetadataKeyKubernetesAgentctlInstanceID) + if remoteID == "" { + remoteID = req.InstanceID + } + return req.TaskID == instance.TaskID && req.SessionID == instance.SessionID && + req.TaskEnvironmentID == getMetadataString(instance.Metadata, MetadataKeyKubernetesResourceEnvironmentID) && + remoteID == getMetadataString(instance.Metadata, MetadataKeyKubernetesAgentctlInstanceID) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_shared_control.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_shared_control.go new file mode 100644 index 0000000000..942ff42e28 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_shared_control.go @@ -0,0 +1,105 @@ +package lifecycle + +import ( + "context" + "errors" + "fmt" + + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/secrets" + corev1 "k8s.io/api/core/v1" +) + +func (r *KubernetesExecutor) connectSharedKubernetesAgentctl(ctx context.Context, runtime *kubernetesRuntimeClient, req *ExecutorCreateRequest, pod *corev1.Pod, remoteID string) (_ *agentctl.Client, _ kubeexecutor.PortForwardSession, _ string, _ int, returnedErr error) { + if err := r.prepareSharedKubernetesCredentials(ctx, runtime, req, pod, getMetadataString(req.Metadata, MetadataKeyKubernetesMainContainer)); err != nil { + return nil, nil, "", 0, err + } + forward, control, err := r.connectHealthyKubernetesControl(ctx, runtime, pod) + if err != nil { + return nil, nil, "", 0, err + } + defer func() { _ = forward.Close() }() + unlock := r.lockInstance("control:" + req.TaskEnvironmentID) + defer unlock() + token, err := r.sharedKubernetesControlToken(ctx, req) + if err != nil { + return nil, nil, "", 0, err + } + secretID := getMetadataString(req.Metadata, MetadataKeyAuthTokenSecret) + control.SetAuthToken(token) + response, created, err := getOrCreateSharedKubernetesInstance(ctx, control, req, remoteID) + defer func() { + if returnedErr != nil && created { + rollbackCtx, cancel := kubernetesDurableContext(ctx) + defer cancel() + returnedErr = errors.Join(returnedErr, control.DeleteInstance(rollbackCtx, remoteID)) + } + }() + if isAgentctlAuthError(err) && req.BootstrapNonce != "" { + token, err = control.Handshake(ctx, req.BootstrapNonce) + if err == nil && r.secretStore != nil && secretID != "" { + err = r.persistSharedKubernetesControlToken(ctx, secretID, token) + } + if err == nil { + response, created, err = getOrCreateSharedKubernetesInstance(ctx, control, req, remoteID) + } + } + if err != nil { + return nil, nil, "", 0, err + } + instanceForward, err := startKubernetesForward(ctx, runtime.streams, pod, uint16(response.Port)) + if err != nil { + return nil, nil, "", 0, err + } + client := newKubernetesAgentctlClient(r.logger, req, remoteID, token, instanceForward.LocalPort()) + if err := client.Health(ctx); err != nil { + client.Close() + _ = instanceForward.Close() + return nil, nil, "", 0, err + } + return client, instanceForward, token, response.Port, nil +} + +func getOrCreateSharedKubernetesInstance(ctx context.Context, control *agentctl.ControlClient, req *ExecutorCreateRequest, remoteID string) (*agentctl.CreateInstanceResponse, bool, error) { + request := buildReconnectCreateInstanceRequest(req, remoteID) + info, err := control.GetInstance(ctx, remoteID) + if err == nil { + response := &agentctl.CreateInstanceResponse{ID: info.ID, Port: info.Port} + if err := validateKubernetesAgentctlInstanceResponse(response, request); err != nil { + return nil, false, err + } + if info.WorkspacePath != request.WorkspacePath { + return nil, false, fmt.Errorf("kubernetes instance workspace mismatch") + } + return response, false, nil + } + if !errors.Is(err, agentctl.ErrInstanceNotFound) { + return nil, false, err + } + response, createErr := createOrReconcileKubernetesAgentctlInstance(ctx, control, request) + return response, true, createErr +} + +func (r *KubernetesExecutor) sharedKubernetesControlToken(ctx context.Context, req *ExecutorCreateRequest) (string, error) { + id := getMetadataString(req.Metadata, MetadataKeyAuthTokenSecret) + if r.secretStore == nil || id == "" { + return req.AuthToken, nil + } + r.mu.Lock() + pending := r.pendingControlTokens[id] + r.mu.Unlock() + if pending == "" { + var err error + pending, err = r.secretStore.Reveal(ctx, kubernetesControlRecoverySecretID(id)) + if err != nil && !errors.Is(err, secrets.ErrNotFound) { + return "", err + } + } + if pending != "" { + if err := r.persistSharedKubernetesControlToken(ctx, id, pending); err != nil { + return "", err + } + } + return r.secretStore.Reveal(ctx, id) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_failure_test.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_failure_test.go new file mode 100644 index 0000000000..d343642326 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_failure_test.go @@ -0,0 +1,162 @@ +package lifecycle + +import ( + "context" + "errors" + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + "testing" + + "github.com/kandev/kandev/internal/secrets" + "github.com/kandev/kandev/internal/task/models" + "github.com/stretchr/testify/require" +) + +type taskPodFailedSecrets struct{ secrets.SecretStore } + +func (s taskPodFailedSecrets) Get(context.Context, string) (*secrets.Secret, error) { + return nil, errors.New("injected secret storage failure") +} + +func TestKubernetesTaskPodSecretFailureRollsBackAttachment(t *testing.T) { + f := newTaskPodFixture(t) + f.launch(t, 1) + f.runtime.secretStore = taskPodFailedSecrets{f.secretStore} + _, err := f.runtime.CreateInstance(context.Background(), taskPodRequest(2)) + require.ErrorContains(t, err, "injected secret storage failure") + _, active, _ := f.control.snapshot() + require.Len(t, active, 1, "failed attachment must not leave a remote agent") + require.Contains(t, active, "instance-1") + require.Empty(t, f.resources.deletedPods) + require.Empty(t, f.resources.deletedPVCs) + f.runtime.mu.Lock() + _, exists := f.runtime.sessions["instance-2"] + f.runtime.mu.Unlock() + require.False(t, exists, "failed attachment must close its forward") +} + +func TestKubernetesTaskPodPinsCleanupScript(t *testing.T) { + f := newTaskPodFixture(t) + req := taskPodRequest(1) + req.Metadata[MetadataKeyCleanupScript] = "echo clean-task" + _, err := f.runtime.CreateInstance(context.Background(), req) + require.NoError(t, err) + record, err := f.repo.GetKubernetesEnvironment(context.Background(), req.TaskEnvironmentID) + require.NoError(t, err) + require.Equal(t, "echo clean-task", getMetadataString(record.Metadata, MetadataKeyCleanupScript)) +} + +func TestKubernetesTaskPodRejectsDifferentProfile(t *testing.T) { + f := newTaskPodFixture(t) + f.launch(t, 1) + req := taskPodRequest(2) + req.Metadata[MetadataKeyExecutorProfileID] = "different-profile" + _, err := f.runtime.CreateInstance(context.Background(), req) + require.ErrorIs(t, err, models.ErrWorkspaceReuseUnsafe) + require.Len(t, f.resources.createdPods, 1) +} + +type taskPodLegacyStore struct { + KubernetesEnvironmentStore + rows []*models.ExecutorRunning +} + +func (s taskPodLegacyStore) ListExecutorsRunningByTaskID(context.Context, string) ([]*models.ExecutorRunning, error) { + return s.rows, nil +} + +func TestKubernetesTaskPodLegacyInventoryMustAgree(t *testing.T) { + for _, key := range []string{MetadataKeyKubernetesPodUID, MetadataKeyKubernetesNamespace, MetadataKeyKubernetesResourceExecutorID, MetadataKeyKubernetesProfileSnapshot} { + t.Run(key, func(t *testing.T) { + f := newTaskPodFixture(t) + first := f.launch(t, 1) + second := cloneKubernetesMetadata(first.Metadata) + second[key] = "conflicting" + f.runtime.environmentStore = taskPodLegacyStore{KubernetesEnvironmentStore: f.repo, rows: []*models.ExecutorRunning{ + {Runtime: "k8s", Metadata: first.Metadata}, {Runtime: "k8s", Metadata: second}, + }} + record := &models.KubernetesEnvironment{} + err := f.runtime.adoptKubernetesEnvironment(context.Background(), taskPodRequest(2), record) + require.ErrorIs(t, err, models.ErrWorkspaceReuseUnsafe) + }) + } +} + +func TestKubernetesTaskPodLegacySessionCannotDeleteAdoptedPod(t *testing.T) { + f := newTaskPodFixture(t) + first := f.launch(t, 1) + f.launch(t, 2) + delete(first.Metadata, metadataKubernetesTaskOwned) + first.StopReason = StopReasonTaskDeleted + require.NoError(t, f.runtime.StopInstance(context.Background(), first, true)) + require.Empty(t, f.resources.deletedPods) + require.Empty(t, f.resources.deletedPVCs) + _, active, _ := f.control.snapshot() + require.Contains(t, active, "instance-2") +} + +type taskPodFailedReleaseStore struct{ KubernetesEnvironmentStore } + +func (s taskPodFailedReleaseStore) SaveKubernetesEnvironment(ctx context.Context, record *models.KubernetesEnvironment, release bool) error { + if release { + return errors.New("injected release failure") + } + return s.KubernetesEnvironmentStore.SaveKubernetesEnvironment(ctx, record, release) +} + +func TestKubernetesTaskPodReleaseFailureRollsBackAttachment(t *testing.T) { + f := newTaskPodFixture(t) + f.launch(t, 1) + f.runtime.environmentStore = taskPodFailedReleaseStore{f.repo} + _, err := f.runtime.CreateInstance(context.Background(), taskPodRequest(2)) + require.ErrorContains(t, err, "injected release failure") + _, active, _ := f.control.snapshot() + require.Len(t, active, 1) + require.Contains(t, active, "instance-1") +} + +func TestKubernetesTaskPodConnectionFailureRollsBackOnlyNewInstance(t *testing.T) { + for _, resume := range []bool{false, true} { + t.Run(map[bool]string{false: "new attachment", true: "existing instance"}[resume], func(t *testing.T) { + f := newTaskPodFixture(t) + first := f.launch(t, 1) + attaching := newFakeKubernetesExecutor(t, f.resources, f.execs, map[uint16]uint16{uint16(kubeexecutor.DefaultAgentctlPort): f.port}) + attaching.environmentStore = f.repo + attaching.secretStore = f.secretStore + t.Cleanup(func() { require.NoError(t, attaching.Close()) }) + req := taskPodRequest(2) + if resume { + req.SessionID = first.SessionID + req.PreviousExecutionID = first.InstanceID + req.Metadata = cloneKubernetesMetadata(first.Metadata) + } + _, err := attaching.CreateInstance(context.Background(), req) + require.Error(t, err) + _, active, _ := f.control.snapshot() + require.Len(t, active, 1) + require.Contains(t, active, "instance-1") + }) + } +} + +func TestKubernetesTaskPodLegacyResumeDoesNotRequireConsolidation(t *testing.T) { + f := newTaskPodFixture(t) + ctx := context.Background() + req := taskPodRequest(1) + first, err := f.runtime.createSessionInstance(ctx, req) + require.NoError(t, err) + other := cloneKubernetesMetadata(first.Metadata) + other[MetadataKeyKubernetesPodUID] = "different-legacy-pod" + f.runtime.environmentStore = taskPodLegacyStore{KubernetesEnvironmentStore: f.repo, rows: []*models.ExecutorRunning{{Runtime: "k8s", Metadata: first.Metadata}, {Runtime: "k8s", Metadata: other}}} + resumed := taskPodRequest(1) + resumed.InstanceID = "resumed-legacy" + resumed.PreviousExecutionID = first.InstanceID + resumed.AuthToken = first.AuthToken + resumed.BootstrapNonce = first.BootstrapNonce + for key, value := range first.Metadata { + resumed.Metadata[key] = value + } + instance, err := f.runtime.CreateInstance(ctx, resumed) + require.NoError(t, err, "an existing legacy session can resume its exact pod without consolidating other pods") + require.Equal(t, first.Metadata[MetadataKeyKubernetesPodUID], instance.Metadata[MetadataKeyKubernetesPodUID]) + require.Len(t, f.resources.createdPods, 1) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_fixture_test.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_fixture_test.go new file mode 100644 index 0000000000..6966030609 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_fixture_test.go @@ -0,0 +1,172 @@ +package lifecycle + +import ( + "context" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + + "github.com/jmoiron/sqlx" + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/secrets" + "github.com/kandev/kandev/internal/task/models" + tasksqlite "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/stretchr/testify/require" +) + +type taskPodControl struct { + mu sync.Mutex + active map[string]agentctl.CreateInstanceRequest + created []agentctl.CreateInstanceRequest + handshakes int + token string + failID string + beforeDelete func() +} + +func (c *taskPodControl) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodDelete && c.beforeDelete != nil { + c.beforeDelete() + } + c.mu.Lock() + defer c.mu.Unlock() + if r.URL.Path == "/health" { + w.WriteHeader(http.StatusOK) + return + } + if r.URL.Path == "/auth/handshake" { + c.handshakes++ + if c.handshakes > 1 { + http.Error(w, "nonce already used", http.StatusUnauthorized) + return + } + _ = json.NewEncoder(w).Encode(map[string]string{"token": c.token}) + return + } + if r.Header.Get("Authorization") != "Bearer "+c.token { + w.WriteHeader(http.StatusUnauthorized) + _ = json.NewEncoder(w).Encode(map[string]string{"error": "unauthorized"}) + return + } + if r.URL.Path == "/api/v1/instances" { + var request agentctl.CreateInstanceRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + http.Error(w, err.Error(), 400) + return + } + if request.ID == c.failID { + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]string{"error": "injected create failure"}) + return + } + c.created = append(c.created, request) + c.active[request.ID] = request + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(agentctl.CreateInstanceResponse{ID: request.ID, Port: 41001}) + return + } + id := strings.TrimPrefix(r.URL.Path, "/api/v1/instances/") + request, exists := c.active[id] + if !exists { + http.NotFound(w, r) + return + } + if r.Method == http.MethodDelete { + delete(c.active, id) + w.WriteHeader(http.StatusNoContent) + return + } + _ = json.NewEncoder(w).Encode(agentctl.InstanceInfo{ID: id, Port: 41001, WorkspacePath: request.WorkspacePath}) +} + +func (c *taskPodControl) snapshot() ([]agentctl.CreateInstanceRequest, map[string]agentctl.CreateInstanceRequest, int) { + c.mu.Lock() + defer c.mu.Unlock() + active := make(map[string]agentctl.CreateInstanceRequest, len(c.active)) + for k, v := range c.active { + active[k] = v + } + return append([]agentctl.CreateInstanceRequest(nil), c.created...), active, c.handshakes +} + +func (c *taskPodControl) reboot() { + c.mu.Lock() + defer c.mu.Unlock() + c.active = map[string]agentctl.CreateInstanceRequest{} + c.handshakes = 0 + c.token += "-restarted" +} + +type taskPodFixture struct { + runtime *KubernetesExecutor + manager *Manager + repo *tasksqlite.Repository + database *sqlx.DB + resources *fakeKubernetesResources + execs *recordingKubernetesExec + control *taskPodControl + secretStore secrets.SecretStore + port uint16 +} + +func newTaskPodFixture(t *testing.T) *taskPodFixture { + t.Helper() + repo, database := runOwnerTestRepository(t) + ctx := context.Background() + require.NoError(t, repo.CreateTask(ctx, &models.Task{ID: "task-1", Title: "shared pod"})) + require.NoError(t, repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{ID: "environment-1", TaskID: "task-1", ExecutorType: "k8s", ExecutorID: "executor-1", ExecutorProfileID: "profile-1", Status: models.TaskEnvironmentStatusCreating, OwnershipGeneration: 1})) + require.NoError(t, repo.CreateExecutor(ctx, &models.Executor{ID: "executor-1", Name: "cluster", Type: models.ExecutorTypeKubernetes, Config: map[string]string{"auth_mode": "in_cluster", "namespace": "kandev-agents", "request_timeout_seconds": "30"}})) + control := &taskPodControl{active: map[string]agentctl.CreateInstanceRequest{}, token: "shared-control-token"} + server := httptest.NewServer(control) + t.Cleanup(server.Close) + _, raw, err := net.SplitHostPort(server.Listener.Addr().String()) + require.NoError(t, err) + port, err := strconv.Atoi(raw) + require.NoError(t, err) + f := &taskPodFixture{repo: repo, database: database, resources: &fakeKubernetesResources{}, execs: &recordingKubernetesExec{}, control: control, secretStore: newInMemorySecretStore(), port: uint16(port)} + f.restartBackend(t) + return f +} + +func (f *taskPodFixture) restartBackend(t *testing.T) { + t.Helper() + if f.runtime != nil { + require.NoError(t, f.runtime.Close()) + } + f.runtime = newFakeKubernetesExecutor(t, f.resources, f.execs, map[uint16]uint16{uint16(kubeexecutor.DefaultAgentctlPort): f.port, 41001: f.port}) + current := f.runtime + t.Cleanup(func() { require.NoError(t, current.Close()) }) + f.manager = newTestManager(t) + f.manager.executorRegistry = NewExecutorRegistry(f.manager.logger) + f.manager.executorRegistry.Register(f.runtime) + f.manager.SetExecutorRunningWriter(f.repo) + f.manager.SetSecretStore(f.secretStore) +} + +func taskPodRequest(number int) *ExecutorCreateRequest { + req := validKubernetesCreateRequest() + setManagedKubernetesWorkspace(req) + req.InstanceID = "instance-" + strconv.Itoa(number) + req.SessionID = "session-" + strconv.Itoa(number) + req.AgentProfileID = "agent-profile-" + strconv.Itoa(number) + req.Env = map[string]string{"SESSION_CREDENTIAL": "credential-" + strconv.Itoa(number)} + req.WorkspaceReuseRequired = number > 1 + return req +} + +func (f *taskPodFixture) launch(t *testing.T, number int) *ExecutorInstance { + t.Helper() + instance, err := f.runtime.CreateInstance(context.Background(), taskPodRequest(number)) + require.NoError(t, err) + if number == 1 { + _, err = f.database.Exec(`UPDATE task_environments SET status = 'ready' WHERE id = 'environment-1'`) + require.NoError(t, err) + } + return instance +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_recovery_test.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_recovery_test.go new file mode 100644 index 0000000000..863423b28a --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_recovery_test.go @@ -0,0 +1,231 @@ +package lifecycle + +import ( + "context" + "github.com/kandev/kandev/internal/secrets" + "testing" + "time" + + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + "github.com/kandev/kandev/internal/task/models" + "github.com/stretchr/testify/require" +) + +func TestKubernetesTaskPodBackendRestartKeepsSessionInstances(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + f.launch(t, 2) + f.restartBackend(t) + req := taskPodRequest(1) + req.InstanceID = "resumed-execution" + req.PreviousExecutionID = a.InstanceID + req.Metadata = cloneKubernetesMetadata(a.Metadata) + resumed, err := f.runtime.CreateInstance(context.Background(), req) + require.NoError(t, err) + require.Equal(t, a.Metadata[MetadataKeyKubernetesPodUID], resumed.Metadata[MetadataKeyKubernetesPodUID]) + require.Equal(t, "instance-1", resumed.Metadata[MetadataKeyKubernetesAgentctlInstanceID]) + _, active, _ := f.control.snapshot() + require.Len(t, active, 2, "resume must not leave a second agent process for the same session") + require.Len(t, f.resources.createdPods, 1) +} + +func TestKubernetesTaskPodRefreshKeepsSiblingSessionIdentity(t *testing.T) { + req := taskPodRequest(2) + req.Metadata[metadataKubernetesTaskOwned] = true + instance := &ExecutorInstance{InstanceID: req.InstanceID, TaskID: req.TaskID, SessionID: req.SessionID, Metadata: req.Metadata} + identity := kubeexecutor.ResourceIdentity{TaskID: req.TaskID, SessionID: "session-1", EnvironmentID: req.TaskEnvironmentID} + refreshed := kubernetesActiveRefreshRequest(req, instance, identity) + require.Equal(t, "session-2", refreshed.SessionID) +} + +func TestKubernetesTaskPodContainerRestartSharesRotatedControlToken(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + b := f.launch(t, 2) + f.control.reboot() + f.resources.mu.Lock() + f.resources.pod.Status.ContainerStatuses[0].RestartCount = 1 + f.resources.mu.Unlock() + ctx := context.Background() + for _, instance := range []*ExecutorInstance{a, b} { + refreshed, err := f.runtime.RefreshRemoteInstance(ctx, kubernetesRefreshInstance(instance, taskPodRequest(1).Metadata)) + require.NoError(t, err) + require.NotNil(t, refreshed) + require.Equal(t, instance.SessionID, refreshed.Instance.SessionID) + require.NoError(t, refreshed.Commit(nil)) + } + _, active, handshakes := f.control.snapshot() + require.Len(t, active, 2) + require.Equal(t, 1, handshakes) +} + +func TestKubernetesTaskPodRejectsPreparingEnvironment(t *testing.T) { + f := newTaskPodFixture(t) + _, err := f.runtime.CreateInstance(context.Background(), taskPodRequest(2)) + require.ErrorIs(t, err, models.ErrWorkspacePreparing) + require.Empty(t, f.resources.createdPods) + _, err = f.repo.GetKubernetesEnvironment(context.Background(), "environment-1") + require.ErrorIs(t, err, models.ErrKubernetesEnvironmentNotFound) +} + +func TestKubernetesTaskPodConcurrentAttachAndFailure(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + results := make(chan error, 2) + for _, n := range []int{2, 3} { + go func(number int) { + _, err := f.runtime.CreateInstance(context.Background(), taskPodRequest(number)) + results <- err + }(n) + } + require.NoError(t, <-results) + require.NoError(t, <-results) + require.Len(t, f.resources.createdPods, 1) + require.Len(t, f.resources.createdPVCs, 1) + f.control.mu.Lock() + f.control.failID = "instance-4" + f.control.mu.Unlock() + _, err := f.runtime.CreateInstance(context.Background(), taskPodRequest(4)) + require.Error(t, err) + _, active, _ := f.control.snapshot() + require.Len(t, active, 3) + require.Contains(t, active, a.InstanceID) + require.Empty(t, f.resources.deletedPods) +} + +func TestKubernetesTaskPodResumeValidationAllowsSibling(t *testing.T) { + f := newTaskPodFixture(t) + f.launch(t, 1) + b := f.launch(t, 2) + require.NoError(t, ValidateKubernetesResumeMetadata(b.Metadata, b.TaskID, b.SessionID, map[string]string{"auth_mode": "in_cluster", "namespace": "kandev-agents"})) +} + +func TestKubernetesTaskPodMissingPodReplacementSharedAcrossResumes(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + b := f.launch(t, 2) + f.restartBackend(t) + f.control.reboot() + f.resources.mu.Lock() + f.resources.pod = nil + f.resources.nextPodUID = "replacement-uid" + f.resources.mu.Unlock() + for _, instance := range []*ExecutorInstance{a, b} { + req := taskPodRequest(2) + req.SessionID = instance.SessionID + req.InstanceID = instance.InstanceID + "-resumed" + req.PreviousExecutionID = instance.InstanceID + req.Metadata = cloneKubernetesMetadata(instance.Metadata) + resumed, err := f.runtime.CreateInstance(context.Background(), req) + require.NoError(t, err) + require.Equal(t, "replacement-uid", resumed.Metadata[MetadataKeyKubernetesPodUID]) + } + require.Len(t, f.resources.createdPods, 2) + require.Len(t, f.resources.createdPVCs, 1) + _, active, handshakes := f.control.snapshot() + require.Len(t, active, 2) + require.Equal(t, 1, handshakes) + require.NoError(t, f.runtime.StopInstance(context.Background(), a, true)) + _, active, _ = f.control.snapshot() + require.NotContains(t, active, "instance-1") + require.Contains(t, active, "instance-2") +} + +func TestKubernetesTaskPodStaleSessionCannotOverwriteRotatedToken(t *testing.T) { + for _, layer := range []string{"lifecycle", "manager"} { + t.Run(layer, func(t *testing.T) { + f := newTaskPodFixture(t) + instance := f.launch(t, 1) + ctx := context.Background() + id := getMetadataString(instance.Metadata, MetadataKeyAuthTokenSecret) + rotated := "newer-sibling-token" + require.NoError(t, f.secretStore.Update(ctx, id, &secrets.UpdateSecretRequest{Value: &rotated})) + if layer == "lifecycle" { + record, err := f.repo.GetKubernetesEnvironment(ctx, "environment-1") + require.NoError(t, err) + require.NoError(t, f.runtime.persistKubernetesEnvironmentSecrets(ctx, record, instance)) + } else { + execution := &AgentExecution{ID: instance.InstanceID, metadata: cloneKubernetesMetadata(instance.Metadata)} + _, err := f.manager.persistRequiredKubernetesRuntimeSecrets(ctx, instance, execution) + require.NoError(t, err) + } + got, err := f.secretStore.Reveal(ctx, id) + require.NoError(t, err) + require.Equal(t, rotated, got) + }) + } +} + +func TestKubernetesTaskPodRetriesRotatedTokenPersistence(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + f.launch(t, 2) + id := getMetadataString(a.Metadata, MetadataKeyAuthTokenSecret) + store := &failingUpdateSecretStore{inMemorySecretStore: f.secretStore.(*inMemorySecretStore), failUpdateFor: id} + f.runtime.secretStore = store + f.control.reboot() + f.resources.mu.Lock() + f.resources.pod.Status.ContainerStatuses[0].RestartCount = 1 + f.resources.mu.Unlock() + ctx := context.Background() + _, err := f.runtime.RefreshRemoteInstance(ctx, kubernetesRefreshInstance(a, taskPodRequest(1).Metadata)) + require.ErrorContains(t, err, "injected secret update failure") + store.failUpdateFor = "" + f.restartBackend(t) + req := taskPodRequest(1) + req.PreviousExecutionID = a.InstanceID + req.Metadata = cloneKubernetesMetadata(a.Metadata) + _, err = f.runtime.CreateInstance(ctx, req) + require.NoError(t, err) + _, _, handshakes := f.control.snapshot() + require.Equal(t, 1, handshakes, "storage retry must preserve the single-use handshake result") + f.restartBackend(t) + f.launch(t, 3) + require.Len(t, f.resources.createdPods, 1) +} + +// Reviewer-requested contract coverage: attaching a different session may overlap +// deletion of the stopped session's remote agent without deleting shared storage. +func TestKubernetesTaskPodConcurrentStopAndAttach(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + f.launch(t, 2) + entered, release := make(chan struct{}), make(chan struct{}) + f.control.beforeDelete = func() { close(entered); <-release } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + stopped := make(chan error, 1) + go func() { stopped <- f.runtime.StopInstance(ctx, a, true) }() + select { + case <-entered: + case <-ctx.Done(): + close(release) + t.Fatal("stop did not reach remote deletion") + } + _, attachErr := f.runtime.CreateInstance(ctx, taskPodRequest(3)) + close(release) + require.NoError(t, attachErr) + require.NoError(t, <-stopped) + _, active, _ := f.control.snapshot() + require.NotContains(t, active, "instance-1") + require.Contains(t, active, "instance-2") + require.Contains(t, active, "instance-3") + require.Len(t, f.resources.createdPods, 1) + require.Empty(t, f.resources.deletedPods) + require.Empty(t, f.resources.deletedPVCs) +} + +func TestKubernetesTaskPodStopOutlivesCanceledCaller(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + f.launch(t, 2) + f.resources.rejectCanceledGetContexts = true + f.secretStore.(*inMemorySecretStore).rejectCanceledCalls = true + ctx, cancel := context.WithCancel(context.Background()) + cancel() + require.NoError(t, f.runtime.StopInstance(ctx, a, true)) + _, active, _ := f.control.snapshot() + require.NotContains(t, active, "instance-1") + require.Contains(t, active, "instance-2") + require.Empty(t, f.resources.deletedPods) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_test.go b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_test.go new file mode 100644 index 0000000000..0fc7ae5c7a --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_kubernetes_task_pod_test.go @@ -0,0 +1,146 @@ +package lifecycle + +import ( + "context" + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + "github.com/kandev/kandev/internal/task/models" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "path" + "strings" + "testing" +) + +func TestKubernetesTaskPodAdditionalSession(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + b := f.launch(t, 2) + require.Len(t, f.resources.createdPods, 1) + require.Len(t, f.resources.createdPVCs, 1) + require.Equal(t, a.Metadata[MetadataKeyKubernetesPodUID], b.Metadata[MetadataKeyKubernetesPodUID]) + instances, active, handshakes := f.control.snapshot() + require.Len(t, instances, 2) + require.Len(t, active, 2) + require.Equal(t, 1, handshakes) + require.NotEqual(t, instances[0].ID, instances[1].ID) + require.Equal(t, "session-2", instances[1].SessionID) + require.NotEmpty(t, instances[0].Env["HOME"]) + require.NotEqual(t, instances[0].Env["HOME"], instances[1].Env["HOME"]) + require.Equal(t, "agent-profile-1", instances[0].Env[kubernetesEnvExecutionProfile]) + require.Equal(t, "agent-profile-2", instances[1].Env[kubernetesEnvExecutionProfile]) + require.Equal(t, "credential-1", instances[0].Env["SESSION_CREDENTIAL"]) + require.Equal(t, "credential-2", instances[1].Env["SESSION_CREDENTIAL"]) + require.Equal(t, "session-2", instances[1].Env[kubernetesEnvSessionID]) +} + +func TestKubernetesTaskPodStopPreservesSibling(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + b := f.launch(t, 2) + ctx := context.Background() + b.StopReason = StopReasonTaskDeleted + require.NoError(t, f.runtime.StopInstance(ctx, b, true)) + require.Empty(t, f.resources.deletedPods) + require.Empty(t, f.resources.deletedPVCs) + require.NoError(t, a.Client.Health(ctx)) + require.NoError(t, f.manager.deleteKubernetesRuntimeSecrets(ctx, b.Metadata)) + _, err := f.secretStore.Reveal(ctx, getMetadataString(a.Metadata, MetadataKeyAuthTokenSecret)) + require.NoError(t, err) + _, active, _ := f.control.snapshot() + require.Contains(t, active, "instance-1") + require.NotContains(t, active, "instance-2") +} + +func TestKubernetesTaskPodTerminalCleanup(t *testing.T) { + f := newTaskPodFixture(t) + a := f.launch(t, 1) + b := f.launch(t, 2) + ctx := context.Background() + env, err := f.repo.GetTaskEnvironment(ctx, "environment-1") + require.NoError(t, err) + require.Error(t, f.manager.DestroyKubernetesEnvironment(ctx, env), "a live sibling blocks physical teardown") + require.Empty(t, f.resources.deletedPods) + require.NoError(t, f.runtime.StopInstance(ctx, a, true)) + require.NoError(t, f.runtime.StopInstance(ctx, b, true)) + require.NoError(t, f.manager.DestroyKubernetesEnvironment(ctx, env)) + require.Len(t, f.resources.deletedPods, 1) + require.Len(t, f.resources.deletedPVCs, 1) + _, err = f.repo.GetKubernetesEnvironment(ctx, env.ID) + require.ErrorIs(t, err, models.ErrKubernetesEnvironmentNotFound) + require.NoError(t, f.manager.DestroyKubernetesEnvironment(ctx, env)) + require.NoError(t, f.runtime.StopInstance(ctx, a, true), "session stop remains idempotent after task cleanup") + require.Len(t, f.resources.deletedPods, 1) +} + +func TestKubernetesTaskPodControlEnvironmentIsTaskScoped(t *testing.T) { + req := validKubernetesCreateRequest() + req.Metadata[metadataKubernetesTaskOwned] = true + env := kubernetesRuntimeEnvironment(req) + require.Empty(t, env[kubernetesEnvSessionID]) + require.Empty(t, env[kubernetesEnvExecutionProfile]) + require.Equal(t, req.TaskEnvironmentID, env[kubernetesEnvInstanceID]) +} + +func TestKubernetesTaskPodBootstrapDropsAgentCredentials(t *testing.T) { + req := validKubernetesCreateRequest() + req.Metadata[metadataKubernetesTaskOwned] = true + req.Env = map[string]string{"SESSION_CREDENTIAL": "must-not-be-inherited"} + execs := &recordingKubernetesExec{} + runtime := newFakeKubernetesExecutor(t, &fakeKubernetesResources{}, execs, nil) + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "kandev-agents", Name: "task-pod"}} + require.NoError(t, runtime.bootstrapPod(context.Background(), &kubernetesRuntimeClient{streams: kubeexecutor.NewStreamOperations(execs, nil)}, req, pod, kubeexecutor.ProfileConfig{MainContainer: "kandev-agent"}, "bootstrap-nonce", []byte("agentctl"))) + var controlEnv string + for _, call := range execs.requests { + if strings.Contains(strings.Join(call.request.Command, " "), kubernetesAuthEnvPath) && len(call.stdin) > 0 { + controlEnv = string(call.stdin) + } + } + require.NotEmpty(t, controlEnv) + require.NotContains(t, controlEnv, "must-not-be-inherited") + require.Contains(t, controlEnv, "bootstrap-nonce") +} + +func TestKubernetesTaskPodResourceIdentityIsEnvironmentScoped(t *testing.T) { + req := validKubernetesCreateRequest() + req.Metadata[metadataKubernetesTaskOwned] = true + identity, err := kubernetesIdentity(req) + require.NoError(t, err) + require.Equal(t, req.TaskEnvironmentID, identity.InstanceID) + labels, err := kubeexecutor.OwnershipLabels(identity) + require.NoError(t, err) + require.Equal(t, "task-v1", labels["kandev.ai/ownership-version"]) +} + +func TestKubernetesTaskPodCleanupAfterTaskDeletion(t *testing.T) { + f := newTaskPodFixture(t) + instance := f.launch(t, 1) + ctx := context.Background() + env, err := f.repo.GetTaskEnvironment(ctx, "environment-1") + require.NoError(t, err) + require.NoError(t, f.repo.DeleteTask(ctx, env.TaskID)) + require.NoError(t, f.runtime.StopInstance(ctx, instance, true)) + require.NoError(t, f.manager.DestroyKubernetesEnvironment(ctx, env)) + require.Len(t, f.resources.deletedPods, 1) + require.Len(t, f.resources.deletedPVCs, 1) +} + +func TestKubernetesTaskPodAuthUploadCreatesSessionHome(t *testing.T) { + req := validKubernetesCreateRequest() + req.Metadata[metadataKubernetesTaskOwned] = true + // The hardened uploader creates destination parents before preparation or + // agent startup uses HOME, including sessions with no credential files. + require.Equal(t, kubernetesSessionHome(req), path.Dir(kubernetesSessionAuthPath(req))) +} + +func TestKubernetesTaskPodStopAfterExecutionIdentityChanges(t *testing.T) { + f := newTaskPodFixture(t) + instance := f.launch(t, 1) + rebound := *instance + rebound.InstanceID = "resumed-local-execution" + require.NoError(t, f.runtime.StopInstance(context.Background(), &rebound, true)) + env, err := f.repo.GetTaskEnvironment(context.Background(), "environment-1") + require.NoError(t, err) + require.NoError(t, f.manager.DestroyKubernetesEnvironment(context.Background(), env), "stopping the same remote session must release its original local connection") + require.Len(t, f.resources.deletedPods, 1) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager.go b/apps/backend/internal/agent/runtime/lifecycle/manager.go index 0462dc504c..581f738871 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager.go @@ -836,6 +836,7 @@ func (m *Manager) SetPreparerRegistry(registry *PreparerRegistry) { // SetSecretStore sets the secret store for encrypting runtime auth tokens. func (m *Manager) SetSecretStore(store secrets.SecretStore) { m.secretStore = store + m.wireKubernetesEnvironmentStore() } // SetAgentProfileReader wires the reader the launch-prep SkillDeployer uses diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go b/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go index ba1fcfc290..02268d692d 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_execution.go @@ -1474,6 +1474,19 @@ func (m *Manager) persistRuntimeSecretResult( if instance == nil || execution == nil || m.secretStore == nil { return false, errors.New("runtime secret persistence is unavailable") } + if getMetadataBool(instance.Metadata, metadataKubernetesTaskOwned) { + // The shared control connection owns token rotation. A session snapshot + // must not overwrite a newer token published by a sibling refresh. + secretID := getMetadataString(instance.Metadata, metadataKey) + if secretID == "" { + return false, errors.New("kubernetes environment secret reference is missing") + } + if _, err := m.secretStore.Reveal(ctx, secretID); err != nil { + return false, err + } + execution.setMetadataValue(metadataKey, secretID) + return false, nil + } secretID := execution.metadataString(metadataKey) if secretID == "" { resourceInstanceID := execution.metadataString(MetadataKeyKubernetesResourceInstanceID) @@ -1535,6 +1548,9 @@ func (m *Manager) deleteCreatedRuntimeSecrets( } func (m *Manager) deleteKubernetesRuntimeSecrets(ctx context.Context, metadata map[string]interface{}) error { + if getMetadataBool(metadata, metadataKubernetesTaskOwned) { + return nil + } secretIDs := kubernetesRuntimeSecretIDs(metadata) if secretIDs[0] == "" && secretIDs[1] == "" { return nil diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_environment_cleanup.go b/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_environment_cleanup.go new file mode 100644 index 0000000000..c1323a6201 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_environment_cleanup.go @@ -0,0 +1,136 @@ +package lifecycle + +import ( + "context" + "errors" + "fmt" + + "github.com/google/uuid" + "github.com/kandev/kandev/internal/agent/executor" + kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes" + "github.com/kandev/kandev/internal/secrets" + "github.com/kandev/kandev/internal/task/models" +) + +// DestroyKubernetesEnvironment is called only by task resource teardown after +// session termination and the task cleanup barrier have excluded new launches. +func (m *Manager) DestroyKubernetesEnvironment(ctx context.Context, env *models.TaskEnvironment) error { + if env == nil { + return nil + } + if m.executorRegistry == nil { + return errors.New("kubernetes runtime registry is unavailable") + } + backend, err := m.executorRegistry.GetBackend(executor.NameKubernetes) + if err != nil { + return err + } + runtime, ok := backend.(*KubernetesExecutor) + if !ok { + return errors.New("kubernetes runtime is unavailable") + } + return runtime.destroyTaskEnvironment(ctx, env) +} + +func (r *KubernetesExecutor) destroyTaskEnvironment(ctx context.Context, env *models.TaskEnvironment) (returnedErr error) { + if r.environmentStore == nil { + return errors.New("kubernetes environment inventory is unavailable") + } + unlock := r.lockInstance("environment:" + env.ID) + defer unlock() + if _, err := r.environmentStore.GetKubernetesEnvironment(ctx, env.ID); errors.Is(err, models.ErrKubernetesEnvironmentNotFound) { + return nil + } else if err != nil { + return err + } + if err := r.ensureKubernetesEnvironmentStopped(ctx, env); err != nil { + return err + } + record, err := r.environmentStore.ClaimKubernetesEnvironmentCleanup(ctx, env.ID, env.TaskID, env.OwnershipGeneration, uuid.NewString()) + if err != nil { + return err + } + defer func() { + if returnedErr != nil { + persistCtx, cancel := kubernetesDurableContext(ctx) + defer cancel() + returnedErr = errors.Join(returnedErr, r.environmentStore.SaveKubernetesEnvironment(persistCtx, record, true)) + } + }() + if err := r.deleteTaskKubernetesResources(ctx, record); err != nil { + return err + } + for _, id := range []string{record.ControlSecretID, record.BootstrapSecretID, kubernetesControlRecoverySecretID(record.ControlSecretID)} { + if id == "" { + continue + } + if r.secretStore == nil { + return errors.New("kubernetes environment secret store is unavailable") + } + if err := r.secretStore.Delete(ctx, id); err != nil && !errors.Is(err, secrets.ErrNotFound) { + return err + } + } + return r.environmentStore.DeleteKubernetesEnvironment(ctx, record) +} + +func (r *KubernetesExecutor) ensureKubernetesEnvironmentStopped(ctx context.Context, env *models.TaskEnvironment) error { + r.mu.Lock() + for _, session := range r.sessions { + if session.request != nil && session.request.TaskEnvironmentID == env.ID { + r.mu.Unlock() + return models.ErrWorkspaceReuseUnsafe + } + } + r.mu.Unlock() + rows, err := r.environmentStore.ListExecutorsRunningByTaskID(ctx, env.TaskID) + if err != nil { + return err + } + for _, row := range rows { + if row == nil || getMetadataString(row.Metadata, MetadataKeyKubernetesResourceEnvironmentID) != env.ID { + continue + } + switch row.Status { + case models.ExecutorRunningStatusStopped, models.ExecutorRunningStatusFailed, models.ExecutorRunningStatusComplete: + default: + return models.ErrWorkspaceReuseUnsafe + } + } + return nil +} + +func (r *KubernetesExecutor) deleteTaskKubernetesResources(ctx context.Context, record *models.KubernetesEnvironment) error { + if len(record.Metadata) == 0 { + return nil + } + current, err := r.environmentStore.GetExecutor(ctx, getMetadataString(record.Metadata, MetadataKeyKubernetesResourceExecutorID)) + if err != nil { + return err + } + if current == nil || current.Type != models.ExecutorTypeKubernetes { + return models.ErrWorkspaceReuseUnsafe + } + config, err := kubeexecutor.ParseExecutorConfig(current.Config) + if err != nil { + return err + } + metadata := overlayCurrentKubernetesConnectionMetadata(record.Metadata, current.Config, config) + instance := &ExecutorInstance{TaskID: record.TaskID, SessionID: getMetadataString(metadata, MetadataKeyKubernetesResourceSessionID), Metadata: metadata, StopReason: StopReasonTaskDeleted} + runtime, err := r.kubernetesCleanupRuntime(instance, nil) + if err != nil { + return err + } + recorded, identity, err := kubernetesCleanupInventory(instance) + if err != nil { + return err + } + if err := verifyKubernetesCleanupTargets(ctx, runtime.resources, recorded, identity); err != nil { + return err + } + r.runKubernetesTerminalCleanupScript(ctx, runtime.streams, instance, recorded) + if err := deleteKubernetesResources(ctx, runtime.resources, recorded, identity); err != nil { + return fmt.Errorf("delete task Kubernetes resources: %w", err) + } + return nil +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_restart_cleanup.go b/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_restart_cleanup.go index 19b4853ff3..3031501bb4 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_restart_cleanup.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_restart_cleanup.go @@ -115,7 +115,7 @@ func resolvePersistedKubernetesCleanupMetadata( if err != nil { return nil, err } - req := &ExecutorCreateRequest{TaskID: row.TaskID, SessionID: row.SessionID, Metadata: metadata} + req := &ExecutorCreateRequest{TaskID: row.TaskID, SessionID: row.SessionID, TaskEnvironmentID: getMetadataString(metadata, MetadataKeyKubernetesResourceEnvironmentID), Metadata: metadata} if _, _, err := kubernetesRecordedCleanupInventory(req, currentConfig, true); err != nil { return nil, fmt.Errorf("validate persisted Kubernetes runtime inventory: %w", err) } diff --git a/apps/backend/internal/agent/runtime/lifecycle/persistence.go b/apps/backend/internal/agent/runtime/lifecycle/persistence.go index 695058bd2d..42a3289595 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/persistence.go +++ b/apps/backend/internal/agent/runtime/lifecycle/persistence.go @@ -53,6 +53,7 @@ type ExecutorRunningWriter interface { // Optional only for tests that don't exercise the persistence path. func (m *Manager) SetExecutorRunningWriter(w ExecutorRunningWriter) { m.runningWriter = w + m.wireKubernetesEnvironmentStore() } // buildRunningFromExecution maps an in-memory execution into the persistence diff --git a/apps/backend/internal/agentctl/server/api/ownership_test.go b/apps/backend/internal/agentctl/server/api/ownership_test.go index dd9d493f69..f84d395a5e 100644 --- a/apps/backend/internal/agentctl/server/api/ownership_test.go +++ b/apps/backend/internal/agentctl/server/api/ownership_test.go @@ -1,6 +1,7 @@ package api import ( + "errors" "net/http/httptest" "testing" "time" @@ -261,10 +262,9 @@ func TestNonRenewingControlOperationsDoNotRenewOwnership(t *testing.T) { }}, {"instance operation", func(t *testing.T) { _, err := client.GetInstance(t.Context(), "no-such-instance") - // The not-found wording is the handler's 404, distinct from the - // "failed to get instance" a transport error produces, so this - // asserts the request was served rather than merely refused. - if err == nil || err.Error() != `instance "no-such-instance" not found` { + // Only the handler's 404 maps to this sentinel; a transport failure + // must not make the ownership assertion pass vacuously. + if !errors.Is(err, agentctl.ErrInstanceNotFound) { t.Fatalf("GetInstance error = %v, want the handler's not-found response", err) } }}, diff --git a/apps/backend/internal/backendapp/environment_destroyer_kubernetes_test.go b/apps/backend/internal/backendapp/environment_destroyer_kubernetes_test.go new file mode 100644 index 0000000000..4653961b63 --- /dev/null +++ b/apps/backend/internal/backendapp/environment_destroyer_kubernetes_test.go @@ -0,0 +1,15 @@ +package backendapp + +import ( + "context" + "github.com/kandev/kandev/internal/task/models" + "github.com/stretchr/testify/require" + "testing" +) + +func TestKubernetesEnvironmentDestroyerWiring(t *testing.T) { + _, ok := interface{}(&environmentDestroyerAdapter{}).(interface { + DestroyKubernetesEnvironment(context.Context, *models.TaskEnvironment) error + }) + require.True(t, ok, "task cleanup adapter must expose shared Kubernetes teardown") +} diff --git a/apps/backend/internal/backendapp/storage.go b/apps/backend/internal/backendapp/storage.go index 2016ee5eee..32b2cd1675 100644 --- a/apps/backend/internal/backendapp/storage.go +++ b/apps/backend/internal/backendapp/storage.go @@ -83,6 +83,9 @@ func provideRepositories(ctx context.Context, cfg *config.Config, log *logger.Lo if err := recordRequiredStore(ctx, tracker, "schema-meta", nil); err != nil { return nil, nil, nil, err } + if taskRepoErr == nil { + taskRepoErr = taskRepoImpl.RecoverInterruptedKubernetesOperations(ctx) + } if err := recordRequiredStore(ctx, tracker, "task", taskRepoErr); err != nil { return nil, nil, nil, fmt.Errorf("task store: %w", err) } diff --git a/apps/backend/internal/backendapp/worktree.go b/apps/backend/internal/backendapp/worktree.go index 43f2bc32fb..c24fb2a222 100644 --- a/apps/backend/internal/backendapp/worktree.go +++ b/apps/backend/internal/backendapp/worktree.go @@ -109,6 +109,10 @@ type environmentDestroyerAdapter struct { worktrees *worktree.Manager } +func (a *environmentDestroyerAdapter) DestroyKubernetesEnvironment(ctx context.Context, env *models.TaskEnvironment) error { + return a.lifecycle.DestroyKubernetesEnvironment(ctx, env) +} + func (a *environmentDestroyerAdapter) DestroyContainer(ctx context.Context, containerID string) error { return a.lifecycle.DestroyContainer(ctx, containerID) } diff --git a/apps/backend/internal/kubernetes/sessions.go b/apps/backend/internal/kubernetes/sessions.go index 5907cfca55..0da56ff365 100644 --- a/apps/backend/internal/kubernetes/sessions.go +++ b/apps/backend/internal/kubernetes/sessions.go @@ -8,7 +8,6 @@ import ( corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" kubeclient "k8s.io/client-go/kubernetes" agentkubernetes "github.com/kandev/kandev/internal/agent/kubernetes" @@ -100,11 +99,12 @@ func (h *Handler) listSessions( return nil, err } rows := make([]SessionRow, 0, len(runs)) + cache := newSessionStatusCache(client) for _, run := range runs { if !filter.matches(run) { continue } - row, visible, rowErr := h.sessionRow(ctx, client, executorID, run) + row, visible, rowErr := h.sessionRow(ctx, cache, executorID, run) if rowErr != nil { return nil, rowErr } @@ -112,7 +112,7 @@ func (h *Handler) listSessions( rows = append(rows, row) } } - return rows, nil + return h.appendRetainedTaskPods(ctx, client, executorID, filter, rows) } func (f SessionFilter) matches(run *models.ExecutorRunning) bool { @@ -162,7 +162,7 @@ func (h *Handler) sessionStatusSource( func (h *Handler) sessionRow( ctx context.Context, - client kubeclient.Interface, + cache *sessionStatusCache, executorID string, run *models.ExecutorRunning, ) (SessionRow, bool, error) { @@ -188,14 +188,19 @@ func (h *Handler) sessionRow( } return SessionRow{}, false, err } + run, inventoryErr := h.canonicalTaskPodInventory(ctx, run, session, cache) row := newInventorySessionRow(run) row.SessionState = projectedTaskSessionState(session.State) + if inventoryErr != nil { + row.FailureReason = "Kubernetes task inventory is unavailable" + return row, true, nil + } if inventoryFailure := validateSessionInventory(run, executorID, row); inventoryFailure != "" { row.FailureReason = inventoryFailure return row, true, nil } namespace := metadataString(run.Metadata, metadataNamespace) - pod, err := client.CoreV1().Pods(namespace).Get(ctx, row.PodName, metav1.GetOptions{}) + pod, err := cache.pod(ctx, namespace, row.PodName) if err != nil { row.FailureReason = podLookupFailure(err) if apierrors.IsNotFound(err) { @@ -233,7 +238,7 @@ func validateSessionInventory( identity, validIdentity := recordedResourceIdentity(run.Metadata) if !validIdentity || run.ID != run.SessionID || run.ExecutorID != executorID || identity.ExecutorID != executorID || identity.TaskID != run.TaskID || - identity.SessionID != run.SessionID || metadataString(run.Metadata, metadataNamespace) == "" || + (!isTaskPodInventory(run.Metadata) && identity.SessionID != run.SessionID) || metadataString(run.Metadata, metadataNamespace) == "" || row.PodName == "" || metadataString(run.Metadata, metadataPodUID) == "" || metadataString(run.Metadata, metadataMainContainer) == "" || row.WorkspaceKind == "" { return "Kubernetes runtime inventory is incomplete" @@ -274,6 +279,7 @@ func matchesSessionIdentity(pod *corev1.Pod, run *models.ExecutorRunning) bool { func recordedResourceIdentity(metadata map[string]interface{}) (agentkubernetes.ResourceIdentity, bool) { identity := agentkubernetes.ResourceIdentity{ + TaskOwned: metadataString(metadata, agentkubernetes.MetadataKeyOwnershipVersion) == agentkubernetes.TaskOwnershipVersion, ExecutorID: metadataString(metadata, metadataResourceExecutor), ProfileID: metadataString(metadata, metadataResourceProfile), InstanceID: metadataString(metadata, metadataResourceInstance), diff --git a/apps/backend/internal/kubernetes/sessions_cache.go b/apps/backend/internal/kubernetes/sessions_cache.go new file mode 100644 index 0000000000..656372f77c --- /dev/null +++ b/apps/backend/internal/kubernetes/sessions_cache.go @@ -0,0 +1,39 @@ +package kubernetes + +import ( + "context" + "github.com/kandev/kandev/internal/task/models" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + kubeclient "k8s.io/client-go/kubernetes" +) + +type environmentStatusResult struct { + record *models.KubernetesEnvironment + err error +} +type podStatusResult struct { + pod *corev1.Pod + err error +} + +// A request projects one observed physical state onto every authorized session. +// Identity and task authorization remain checked separately for every row. +type sessionStatusCache struct { + client kubeclient.Interface + environments map[string]environmentStatusResult + pods map[string]podStatusResult +} + +func newSessionStatusCache(client kubeclient.Interface) *sessionStatusCache { + return &sessionStatusCache{client: client, environments: make(map[string]environmentStatusResult), pods: make(map[string]podStatusResult)} +} +func (c *sessionStatusCache) pod(ctx context.Context, namespace, name string) (*corev1.Pod, error) { + key := namespace + "/" + name + result, exists := c.pods[key] + if !exists { + result.pod, result.err = c.client.CoreV1().Pods(namespace).Get(ctx, name, metav1.GetOptions{}) + c.pods[key] = result + } + return result.pod, result.err +} diff --git a/apps/backend/internal/kubernetes/sessions_retention_test.go b/apps/backend/internal/kubernetes/sessions_retention_test.go index 349f74bbb6..715571783e 100644 --- a/apps/backend/internal/kubernetes/sessions_retention_test.go +++ b/apps/backend/internal/kubernetes/sessions_retention_test.go @@ -66,7 +66,7 @@ func TestKubernetesSessionRetentionProjectionTable(t *testing.T) { clientset := kubernetesfake.NewSimpleClientset(pod) handler := retentionProjectionHandler(repo, clientset) - row, visible, err := handler.sessionRow(context.Background(), clientset, "executor-1", run) + row, visible, err := handler.sessionRow(context.Background(), newSessionStatusCache(clientset), "executor-1", run) require.NoError(t, err) require.True(t, visible) @@ -120,7 +120,7 @@ func TestKubernetesRetentionProjectionOmitsUnverifiedRequests(t *testing.T) { } handler := retentionProjectionHandler(repo, clientset) - row, visible, err := handler.sessionRow(context.Background(), clientset, "executor-1", run) + row, visible, err := handler.sessionRow(context.Background(), newSessionStatusCache(clientset), "executor-1", run) require.NoError(t, err) require.True(t, visible) @@ -153,7 +153,7 @@ func TestKubernetesMainContainerRequestsPreserveExplicitZero(t *testing.T) { clientset := kubernetesfake.NewSimpleClientset(pod) handler := retentionProjectionHandler(repo, clientset) - row, visible, err := handler.sessionRow(context.Background(), clientset, "executor-1", run) + row, visible, err := handler.sessionRow(context.Background(), newSessionStatusCache(clientset), "executor-1", run) require.NoError(t, err) require.True(t, visible) diff --git a/apps/backend/internal/kubernetes/sessions_task_pod.go b/apps/backend/internal/kubernetes/sessions_task_pod.go new file mode 100644 index 0000000000..ac52b0051b --- /dev/null +++ b/apps/backend/internal/kubernetes/sessions_task_pod.go @@ -0,0 +1,104 @@ +package kubernetes + +import ( + "context" + "errors" + "github.com/kandev/kandev/internal/task/repository/repoerrors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + kubeclient "k8s.io/client-go/kubernetes" + + agentkubernetes "github.com/kandev/kandev/internal/agent/kubernetes" + "github.com/kandev/kandev/internal/task/models" +) + +func isTaskPodInventory(metadata map[string]interface{}) bool { + shared, _ := metadata[agentkubernetes.MetadataKeyTaskOwned].(bool) + return shared +} + +// Resolve physical inventory only after authorizing the session's owning task. +func (h *Handler) canonicalTaskPodInventory(ctx context.Context, run *models.ExecutorRunning, session *models.TaskSession, cache *sessionStatusCache) (*models.ExecutorRunning, error) { + if !isTaskPodInventory(run.Metadata) { + return run, nil + } + store, ok := h.repo.(interface { + GetKubernetesEnvironment(context.Context, string) (*models.KubernetesEnvironment, error) + }) + if !ok { + return run, errors.New("kubernetes environment store unavailable") + } + cached, exists := cache.environments[session.TaskEnvironmentID] + if !exists { + cached.record, cached.err = store.GetKubernetesEnvironment(ctx, session.TaskEnvironmentID) + cache.environments[session.TaskEnvironmentID] = cached + } + record, err := cached.record, cached.err + if err != nil { + return run, err + } + if record == nil || record.TaskID != run.TaskID || record.EnvironmentID != session.TaskEnvironmentID || metadataString(record.Metadata, metadataResourceEnv) != record.EnvironmentID { + return run, models.ErrWorkspaceReuseUnsafe + } + copyRun := *run + copyRun.Metadata = record.Metadata + return ©Run, nil +} + +// Retained physical resources stay visible when their last session is removed. +func (h *Handler) appendRetainedTaskPods(ctx context.Context, client kubeclient.Interface, executorID string, filter SessionFilter, rows []SessionRow) ([]SessionRow, error) { + if filter.SessionID != "" { + return rows, nil + } + store, ok := h.repo.(interface { + ListKubernetesEnvironments(context.Context) ([]*models.KubernetesEnvironment, error) + }) + if !ok { + return rows, nil + } + records, err := store.ListKubernetesEnvironments(ctx) + if err != nil { + return nil, err + } + visibleTasks := make(map[string]bool, len(rows)) + for _, row := range rows { + visibleTasks[row.TaskID] = true + } + for _, record := range records { + if record == nil || visibleTasks[record.TaskID] || (filter.TaskID != "" && filter.TaskID != record.TaskID) || metadataString(record.Metadata, metadataResourceExecutor) != executorID { + continue + } + if err := h.access.AuthorizeTaskAccess(ctx, record.TaskID); err != nil { + if errors.Is(err, repoerrors.ErrTaskNotFound) { + continue + } + return nil, err + } + row := retainedTaskPodRow(ctx, client, executorID, record) + rows = append(rows, row) + } + return rows, nil +} + +func retainedTaskPodRow(ctx context.Context, client kubeclient.Interface, executorID string, record *models.KubernetesEnvironment) SessionRow { + run := &models.ExecutorRunning{TaskID: record.TaskID, ExecutorID: executorID, Metadata: record.Metadata} + row := newInventorySessionRow(run) + if !isTaskPodInventory(record.Metadata) || metadataString(record.Metadata, metadataResourceEnv) != record.EnvironmentID { + row.FailureReason = "Kubernetes task inventory is incomplete" + return row + } + if failure := validateSessionInventory(run, executorID, row); failure != "" { + row.FailureReason = failure + return row + } + pod, err := client.CoreV1().Pods(metadataString(record.Metadata, metadataNamespace)).Get(ctx, row.PodName, metav1.GetOptions{}) + if err != nil { + row.FailureReason = podLookupFailure(err) + return row + } + if !matchesSessionIdentity(pod, run) { + row.FailureReason = "Pod identity does not match runtime inventory" + return row + } + populatePodStatus(&row, pod, metadataString(record.Metadata, metadataMainContainer), models.TaskSessionStateIdle) + return row +} diff --git a/apps/backend/internal/kubernetes/sessions_task_pod_test.go b/apps/backend/internal/kubernetes/sessions_task_pod_test.go new file mode 100644 index 0000000000..44ff9ee58a --- /dev/null +++ b/apps/backend/internal/kubernetes/sessions_task_pod_test.go @@ -0,0 +1,89 @@ +package kubernetes + +import ( + "context" + "testing" + "time" + + agentkubernetes "github.com/kandev/kandev/internal/agent/kubernetes" + "github.com/kandev/kandev/internal/task/models" + "github.com/stretchr/testify/require" + kubernetesfake "k8s.io/client-go/kubernetes/fake" +) + +type taskPodStatusRepository struct { + *fakeResourceRepository + environment *models.KubernetesEnvironment + reads int +} + +func (r *taskPodStatusRepository) GetKubernetesEnvironment(context.Context, string) (*models.KubernetesEnvironment, error) { + r.reads++ + return r.environment, nil +} + +func TestSharedTaskPodStatusUsesCanonicalInventory(t *testing.T) { + run := kubernetesRunningRow("environment-1", "session-2", "task-1", "pod-1", "old-uid", time.Now()) + run.Metadata[agentkubernetes.MetadataKeyTaskOwned] = true + run.Metadata[agentkubernetes.MetadataKeyOwnershipVersion] = agentkubernetes.TaskOwnershipVersion + run.Metadata[metadataResourceSession] = "session-1" + metadata := make(map[string]interface{}, len(run.Metadata)) + for k, v := range run.Metadata { + metadata[k] = v + } + metadata[metadataPodUID] = "new-uid" + session := kubernetesTaskSession("session-2", "task-1", "executor-1", "profile-1") + repo := &taskPodStatusRepository{fakeResourceRepository: &fakeResourceRepository{ + executor: &models.Executor{ID: "executor-1", Type: models.ExecutorTypeKubernetes, Config: validHandlerExecutorConfig()}, + runs: []*models.ExecutorRunning{run}, sessions: map[string]*models.TaskSession{session.ID: session}, + }, environment: &models.KubernetesEnvironment{EnvironmentID: "environment-1", TaskID: "task-1", Metadata: metadata}} + pod := kubernetesOwnedPodForIdentity("pod-1", "new-uid", agentkubernetes.ResourceIdentity{ExecutorID: "executor-1", ProfileID: "profile-1", InstanceID: "environment-1", TaskID: "task-1", SessionID: "session-1", EnvironmentID: "environment-1", TaskOwned: true}) + clientset := kubernetesfake.NewSimpleClientset(pod) + handler := NewHandler(repo, &fakeAccessChecker{}, func(agentkubernetes.ExecutorConfig) (*agentkubernetes.Client, error) { + return &agentkubernetes.Client{Clientset: clientset}, nil + }) + sibling := *run + sibling.ID, sibling.SessionID = "session-3", "session-3" + repo.runs = append(repo.runs, &sibling) + repo.sessions["session-3"] = kubernetesTaskSession("session-3", "task-1", "executor-1", "profile-1") + rows, err := handler.listSessions(context.Background(), "executor-1", SessionFilter{}) + require.NoError(t, err) + require.Len(t, rows, 2) + require.Equal(t, 1, repo.reads) + require.Len(t, clientset.Actions(), 1, "probe shared Pod once per request") + require.Empty(t, rows[0].FailureReason) + require.Equal(t, "session-2", rows[0].SessionID) + require.Equal(t, "Running", rows[0].PodPhase) + repo.environment.TaskID = "foreign-task" + rows, err = handler.listSessions(context.Background(), "executor-1", SessionFilter{}) + require.NoError(t, err) + require.NotEmpty(t, rows[0].FailureReason) +} + +func (r *taskPodStatusRepository) ListKubernetesEnvironments(context.Context) ([]*models.KubernetesEnvironment, error) { + return []*models.KubernetesEnvironment{r.environment}, nil +} + +func TestSharedTaskPodStatusRetainedWithoutSessions(t *testing.T) { + run := kubernetesRunningRow("environment-1", "session-1", "task-1", "pod-1", "pod-uid", time.Now()) + run.Metadata[agentkubernetes.MetadataKeyTaskOwned] = true + run.Metadata[agentkubernetes.MetadataKeyOwnershipVersion] = agentkubernetes.TaskOwnershipVersion + repo := &taskPodStatusRepository{fakeResourceRepository: &fakeResourceRepository{ + executor: &models.Executor{ID: "executor-1", Type: models.ExecutorTypeKubernetes, Config: validHandlerExecutorConfig()}, + }, environment: &models.KubernetesEnvironment{EnvironmentID: "environment-1", TaskID: "task-1", Metadata: run.Metadata}} + identity, _ := recordedResourceIdentity(run.Metadata) + pod := kubernetesOwnedPodForIdentity("pod-1", "pod-uid", identity) + clientset := kubernetesfake.NewSimpleClientset(pod) + access := &fakeAccessChecker{} + handler := NewHandler(repo, access, func(agentkubernetes.ExecutorConfig) (*agentkubernetes.Client, error) { + return &agentkubernetes.Client{Clientset: clientset}, nil + }) + rows, err := handler.listSessions(context.Background(), "executor-1", SessionFilter{}) + require.NoError(t, err) + require.Len(t, rows, 1, "retained task pod must remain visible after deleting every session") + require.Empty(t, rows[0].SessionID) + require.Equal(t, "task-1", rows[0].TaskID) + require.Equal(t, "retained", rows[0].RetentionState) + require.Equal(t, "Running", rows[0].PodPhase) + require.NotEmpty(t, access.calls) +} diff --git a/apps/backend/internal/orchestrator/executor/session_coresidency_test.go b/apps/backend/internal/orchestrator/executor/session_coresidency_test.go index 8a9abd11f0..a6fbb16113 100644 --- a/apps/backend/internal/orchestrator/executor/session_coresidency_test.go +++ b/apps/backend/internal/orchestrator/executor/session_coresidency_test.go @@ -211,8 +211,7 @@ func TestLaunchPreparedSession_ObservesWorkingSiblingOnAgentStart(t *testing.T) } // TestResumeSession_ObservesWorkingSiblingOnAgentStart pins the wiring -// half of AC-004.1 for the resume seam. Ordering is enforced by -// production code structure, not a synchronized assertion here. +// half of AC-004.1 for resume admission and asynchronous process startup. func TestResumeSession_ObservesWorkingSiblingOnAgentStart(t *testing.T) { repo := newMockRepository() setupLiveResumeTestFixture(repo) @@ -226,25 +225,42 @@ func TestResumeSession_ObservesWorkingSiblingOnAgentStart(t *testing.T) { if err != nil { t.Fatalf("NewFromZap: %v", err) } + observedAtStart := make(chan int, 1) + finished := make(chan struct{}) agentMgr := &mockAgentManager{ launchAgentFunc: func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { return &LaunchAgentResponse{AgentExecutionID: "exec-new"}, nil }, + startAgentProcessFunc: func(context.Context, string) error { + observedAtStart <- logs.FilterLevelExact(zapcore.WarnLevel).Len() + return nil + }, } exec := NewExecutor(agentMgr, repo, log, ExecutorConfig{ShellPrefs: &mockShellPrefs{}}) exec.SetCapabilities(&mockCapabilities{}) + exec.SetOnAgentProcessStarted(func(context.Context, string, string, string) { close(finished) }) before := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteResume) if _, err := exec.ResumeSession(context.Background(), repo.sessions["sess-1"], true); err != nil { t.Fatalf("ResumeSession: %v", err) } - if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteResume); after != before+1 { - t.Fatalf("admitted[resume] counter = %d, want %d", after, before+1) + select { + case <-finished: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for resumed agent startup") + } + if observed := <-observedAtStart; observed != 2 { + t.Fatalf("observations before process start = %d, want launch admission and process startup", observed) + } + + // Other tests can still emit asynchronous observations into the global counter. + if after := counterValue(sessionCoresidencyAdmittedTotalVar, sessionCoresidencySiteResume); after < before+2 { + t.Fatalf("admitted[resume] counter = %d, want at least %d", after, before+2) } warnings := logs.FilterLevelExact(zapcore.WarnLevel).All() - if len(warnings) != 1 { - t.Fatalf("warning entries = %d, want 1; all=%v", len(warnings), logs.All()) + if len(warnings) != 2 { + t.Fatalf("warning entries = %d, want 2; all=%v", len(warnings), logs.All()) } fields := warnings[0].ContextMap() if fields["site"] != sessionCoresidencySiteResume { diff --git a/apps/backend/internal/persistence/requiredstores/catalog.go b/apps/backend/internal/persistence/requiredstores/catalog.go index 4d058e0aed..fb49c011e7 100644 --- a/apps/backend/internal/persistence/requiredstores/catalog.go +++ b/apps/backend/internal/persistence/requiredstores/catalog.go @@ -63,7 +63,7 @@ type Descriptor struct { // of the database contract and must be initialized before readiness. var catalog = []Descriptor{ {ID: storeIDSchemaMeta, OwnerPackage: "internal/persistence", RequiredTables: []string{"kandev_meta"}, Sweep: startup.StepStoresRepositories}, - {ID: storeIDTask, OwnerPackage: "internal/task/repository/sqlite", RequiredTables: []string{"workspaces", "tasks", "task_workflow_session_bindings"}, DependsOn: []string{storeIDSchemaMeta}, Capabilities: []Capability{CapabilityBoolean, CapabilityTimestamp, CapabilityConflict, CapabilityTransaction}, Sweep: startup.StepStoresRepositories}, + {ID: storeIDTask, OwnerPackage: "internal/task/repository/sqlite", RequiredTables: []string{"workspaces", "tasks", "task_workflow_session_bindings", "task_environment_kubernetes"}, DependsOn: []string{storeIDSchemaMeta}, Capabilities: []Capability{CapabilityBoolean, CapabilityTimestamp, CapabilityConflict, CapabilityTransaction}, Sweep: startup.StepStoresRepositories}, {ID: storeIDWorkflow, OwnerPackage: "internal/workflow/repository", RequiredTables: []string{"workflow_templates", "workflow_steps"}, DependsOn: []string{storeIDTask}, Capabilities: []Capability{CapabilityTimestamp, CapabilityConflict, CapabilityTransaction}, Sweep: startup.StepStoresRepositories}, {ID: "analytics", OwnerPackage: "internal/analytics/repository", RequiredTables: []string{"tasks"}, DependsOn: []string{storeIDTask, storeIDWorkflow}, Capabilities: []Capability{CapabilityTimestamp}, Sweep: startup.StepStoresRepositories}, {ID: storeIDAgentSettings, OwnerPackage: "internal/agent/settings/store", RequiredTables: []string{"agents", "agent_profiles"}, DependsOn: []string{storeIDSchemaMeta}, Capabilities: []Capability{CapabilityBoolean, CapabilityTimestamp, CapabilityConflict}, Sweep: startup.StepStoresRepositories}, diff --git a/apps/backend/internal/task/models/kubernetes_environment.go b/apps/backend/internal/task/models/kubernetes_environment.go new file mode 100644 index 0000000000..fc3725afb4 --- /dev/null +++ b/apps/backend/internal/task/models/kubernetes_environment.go @@ -0,0 +1,22 @@ +package models + +import "errors" + +var ( + ErrKubernetesEnvironmentNotFound = errors.New("kubernetes environment inventory not found") + ErrKubernetesEnvironmentConflict = errors.New("kubernetes environment ownership or operation changed") +) + +// KubernetesEnvironment owns physical resource inventory independently of a session. +// Metadata contains the admitted resource identity and workload snapshot. Secrets +// are encrypted store references and must never be serialized in public responses. +type KubernetesEnvironment struct { + EnvironmentID string + TaskID string + OwnershipGeneration int64 + Revision int64 + OperationID string + Metadata map[string]interface{} + ControlSecretID string + BootstrapSecretID string +} diff --git a/apps/backend/internal/task/repository/sqlite/base_migrations.go b/apps/backend/internal/task/repository/sqlite/base_migrations.go index a45189fae8..617084c696 100644 --- a/apps/backend/internal/task/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/task/repository/sqlite/base_migrations.go @@ -482,6 +482,10 @@ func (r *Repository) runMigrations(ctx context.Context) error { return fmt.Errorf("required task migration: %w", err) } + if _, err := r.db.ExecContext(ctx, kubernetesEnvironmentSchemaDDL); err != nil { + return fmt.Errorf("create Kubernetes environment inventory: %w", err) + } + return nil } diff --git a/apps/backend/internal/task/repository/sqlite/environment_kubernetes.go b/apps/backend/internal/task/repository/sqlite/environment_kubernetes.go new file mode 100644 index 0000000000..e6245b87cc --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/environment_kubernetes.go @@ -0,0 +1,220 @@ +package sqlite + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + + "github.com/jmoiron/sqlx" + "github.com/kandev/kandev/internal/db/dialect" + "github.com/kandev/kandev/internal/task/models" +) + +const kubernetesEnvironmentSchemaDDL = `CREATE TABLE IF NOT EXISTS task_environment_kubernetes ( + environment_id TEXT PRIMARY KEY, + task_id TEXT NOT NULL UNIQUE, + ownership_generation BIGINT NOT NULL, + revision BIGINT NOT NULL DEFAULT 1, + operation_id TEXT NOT NULL DEFAULT '', + metadata TEXT NOT NULL DEFAULT '{}', + control_secret_id TEXT NOT NULL DEFAULT '', + bootstrap_secret_id TEXT NOT NULL DEFAULT '' +)` + +// ClaimKubernetesEnvironment serializes physical resource operations while leaving +// the database transaction closed during remote Kubernetes API calls. +func (r *Repository) ClaimKubernetesEnvironment(ctx context.Context, environmentID, taskID string, generation int64, operationID string) (*models.KubernetesEnvironment, error) { + if environmentID == "" || taskID == "" || generation < 1 || operationID == "" { + return nil, models.ErrKubernetesEnvironmentConflict + } + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback() }() + if err := r.validateKubernetesEnvironmentOwner(ctx, tx, environmentID, taskID, generation); err != nil { + return nil, err + } + _, err = tx.ExecContext(ctx, r.db.Rebind(`INSERT INTO task_environment_kubernetes (environment_id, task_id, ownership_generation) + VALUES (?, ?, ?) ON CONFLICT DO NOTHING`), environmentID, taskID, generation) + if err != nil { + return nil, err + } + record, err := scanKubernetesEnvironment(tx.QueryRowContext(ctx, r.db.Rebind(kubernetesEnvironmentSelect+` WHERE environment_id = ?`), environmentID)) + if err != nil { + return nil, err + } + if record.TaskID != taskID || record.OwnershipGeneration != generation || (record.OperationID != "" && record.OperationID != operationID) { + return nil, models.ErrKubernetesEnvironmentConflict + } + if record.OperationID == "" { + result, err := tx.ExecContext(ctx, r.db.Rebind(`UPDATE task_environment_kubernetes SET operation_id = ?, revision = revision + 1 WHERE environment_id = ? AND revision = ? AND operation_id = ''`), operationID, environmentID, record.Revision) + if err := requireKubernetesEnvironmentWrite(result, err); err != nil { + return nil, err + } + record.OperationID = operationID + record.Revision++ + } + return record, tx.Commit() +} + +const kubernetesEnvironmentSelect = `SELECT environment_id, task_id, ownership_generation, revision, operation_id, metadata, control_secret_id, bootstrap_secret_id FROM task_environment_kubernetes` + +type kubernetesInventoryScanner interface{ Scan(...interface{}) error } + +func scanKubernetesEnvironment(row kubernetesInventoryScanner) (*models.KubernetesEnvironment, error) { + record := &models.KubernetesEnvironment{} + var raw string + err := row.Scan(&record.EnvironmentID, &record.TaskID, &record.OwnershipGeneration, &record.Revision, &record.OperationID, &raw, &record.ControlSecretID, &record.BootstrapSecretID) + if errors.Is(err, sql.ErrNoRows) { + return nil, models.ErrKubernetesEnvironmentNotFound + } + if err != nil { + return nil, err + } + if err := json.Unmarshal([]byte(raw), &record.Metadata); err != nil { + return nil, fmt.Errorf("decode Kubernetes environment inventory: %w", err) + } + return record, nil +} + +func (r *Repository) GetKubernetesEnvironment(ctx context.Context, environmentID string) (*models.KubernetesEnvironment, error) { + return scanKubernetesEnvironment(r.db.QueryRowContext(ctx, r.db.Rebind(kubernetesEnvironmentSelect+` WHERE environment_id = ?`), environmentID)) +} + +// SaveKubernetesEnvironment checkpoints only the current operation and revision. +// Releasing its operation allows the next attachment, recovery, or cleanup. +func (r *Repository) SaveKubernetesEnvironment(ctx context.Context, record *models.KubernetesEnvironment, release bool) error { + if record == nil || record.OperationID == "" { + return models.ErrKubernetesEnvironmentConflict + } + raw, err := json.Marshal(record.Metadata) + if err != nil { + return err + } + operation := record.OperationID + if release { + operation = "" + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(`UPDATE task_environment_kubernetes SET metadata = ?, control_secret_id = ?, bootstrap_secret_id = ?, operation_id = ?, revision = revision + 1 + WHERE environment_id = ? AND task_id = ? AND ownership_generation = ? AND revision = ? AND operation_id = ? AND (`+kubernetesEnvironmentWritable+`)`), string(raw), record.ControlSecretID, record.BootstrapSecretID, operation, record.EnvironmentID, record.TaskID, record.OwnershipGeneration, record.Revision, record.OperationID) + if err := requireKubernetesEnvironmentWrite(result, err); err != nil { + return err + } + record.Revision++ + record.OperationID = operation + return nil +} + +func requireKubernetesEnvironmentWrite(result sql.Result, err error) error { + if err != nil { + return err + } + n, err := result.RowsAffected() + if err != nil { + return err + } + if n != 1 { + return models.ErrKubernetesEnvironmentConflict + } + return nil +} + +// DeleteKubernetesEnvironment releases inventory after exact remote cleanup. +func (r *Repository) DeleteKubernetesEnvironment(ctx context.Context, record *models.KubernetesEnvironment) error { + if record == nil || record.OperationID == "" { + return models.ErrKubernetesEnvironmentConflict + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM task_environment_kubernetes + WHERE environment_id = ? AND task_id = ? AND ownership_generation = ? AND revision = ? AND operation_id = ? + AND (`+kubernetesEnvironmentWritable+`)`), + record.EnvironmentID, record.TaskID, record.OwnershipGeneration, record.Revision, record.OperationID) + return requireKubernetesEnvironmentWrite(result, err) +} + +// RecoverInterruptedKubernetesOperations fences claims from a terminated backend. +// Call only at startup under exclusive runtime-state ownership, before launching +// workers. Inventory remains intact for exact-resource reconciliation or cleanup. +func (r *Repository) RecoverInterruptedKubernetesOperations(ctx context.Context) error { + _, err := r.db.ExecContext(ctx, `UPDATE task_environment_kubernetes SET operation_id = '', revision = revision + 1 WHERE operation_id <> ''`) + return err +} + +func (r *Repository) ListKubernetesEnvironments(ctx context.Context) ([]*models.KubernetesEnvironment, error) { + rows, err := r.db.QueryContext(ctx, kubernetesEnvironmentSelect+` ORDER BY environment_id`) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + records := make([]*models.KubernetesEnvironment, 0) + for rows.Next() { + record, err := scanKubernetesEnvironment(rows) + if err != nil { + return nil, err + } + records = append(records, record) + } + return records, rows.Err() +} + +func (r *Repository) validateKubernetesEnvironmentOwner(ctx context.Context, tx *sqlx.Tx, environmentID, taskID string, generation int64) error { + if dialect.IsPostgres(r.db.DriverName()) { + if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, "kubernetes-environment:"+taskID); err != nil { + return err + } + } + if err := r.lockTaskRowInTx(ctx, tx, taskID); err != nil { + return err + } + var owner string + var currentGeneration int64 + if err := tx.QueryRowContext(ctx, r.db.Rebind(`SELECT task_id, ownership_generation FROM task_environments WHERE id = ?`), environmentID).Scan(&owner, ¤tGeneration); err != nil { + return err + } + if owner != taskID || generation != currentGeneration { + return models.ErrKubernetesEnvironmentConflict + } + return nil +} + +// Physical inventory survives task/environment cascades until durable cleanup +// finishes. Only cleanup claims may operate after both owner rows are gone. +const kubernetesEnvironmentWritable = `EXISTS ( + SELECT 1 FROM task_environments e WHERE e.id = task_environment_kubernetes.environment_id + AND e.task_id = task_environment_kubernetes.task_id + AND e.ownership_generation = task_environment_kubernetes.ownership_generation +) OR (operation_id LIKE 'cleanup:%' + AND NOT EXISTS (SELECT 1 FROM task_environments e WHERE e.id = task_environment_kubernetes.environment_id) + AND NOT EXISTS (SELECT 1 FROM tasks t WHERE t.id = task_environment_kubernetes.task_id))` + +func (r *Repository) ClaimKubernetesEnvironmentCleanup(ctx context.Context, environmentID, taskID string, generation int64, operationID string) (*models.KubernetesEnvironment, error) { + if operationID == "" { + return nil, models.ErrKubernetesEnvironmentConflict + } + operationID = "cleanup:" + operationID + // An existing record is mandatory: deletion never creates resource authority. + result, err := r.db.ExecContext(ctx, r.db.Rebind(`UPDATE task_environment_kubernetes + SET operation_id = ?, revision = revision + 1 + WHERE environment_id = ? AND task_id = ? AND ownership_generation = ? AND operation_id = '' + AND (EXISTS (SELECT 1 FROM task_environments e WHERE e.id = task_environment_kubernetes.environment_id + AND e.task_id = task_environment_kubernetes.task_id AND e.ownership_generation = task_environment_kubernetes.ownership_generation) + OR (NOT EXISTS (SELECT 1 FROM task_environments e WHERE e.id = task_environment_kubernetes.environment_id) + AND NOT EXISTS (SELECT 1 FROM tasks t WHERE t.id = task_environment_kubernetes.task_id)))`), operationID, environmentID, taskID, generation) + if err := requireKubernetesEnvironmentWrite(result, err); err != nil { + return nil, err + } + return r.GetKubernetesEnvironment(ctx, environmentID) +} + +func (r *Repository) ensureKubernetesInventoryAbsent(ctx context.Context, tx *sqlx.Tx, environmentID string) error { + var count int + if err := tx.QueryRowContext(ctx, r.db.Rebind(`SELECT COUNT(*) FROM task_environment_kubernetes WHERE environment_id = ?`), environmentID).Scan(&count); err != nil { + return err + } + if count != 0 { + return models.ErrKubernetesEnvironmentConflict + } + return nil +} diff --git a/apps/backend/internal/task/repository/sqlite/environment_kubernetes_test.go b/apps/backend/internal/task/repository/sqlite/environment_kubernetes_test.go new file mode 100644 index 0000000000..c05ac9f1ba --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/environment_kubernetes_test.go @@ -0,0 +1,179 @@ +package sqlite + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/task/models" + "github.com/stretchr/testify/require" +) + +func TestKubernetesEnvironmentInventorySurvivesSessionRemoval(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-kube", "env-kube") + // The physical resource record must not depend on a session's lifetime. + _, err := repo.db.ExecContext(ctx, `INSERT INTO task_environment_kubernetes + (environment_id, task_id, ownership_generation, metadata, control_secret_id, bootstrap_secret_id) + VALUES ('env-kube', 'task-kube', 1, '{"pod_uid":"exact-uid"}', 'control-secret', 'bootstrap-secret')`) + require.NoError(t, err) + session := &models.TaskSession{ID: "session-kube", TaskID: "task-kube", TaskEnvironmentID: "env-kube"} + require.NoError(t, repo.CreateTaskSession(ctx, session)) + require.NoError(t, repo.DeleteTaskSession(ctx, session)) + require.Error(t, repo.DeleteTaskEnvironment(ctx, "env-kube"), "remote inventory must be cleaned before its environment can be deleted") + var metadata string + require.NoError(t, repo.db.QueryRowContext(ctx, `SELECT metadata FROM task_environment_kubernetes WHERE environment_id = 'env-kube'`).Scan(&metadata)) + require.JSONEq(t, `{"pod_uid":"exact-uid"}`, metadata) +} + +func TestKubernetesEnvironmentClaimExclusion(t *testing.T) { + repo := newRepoForEntityTests(t) + store, ok := interface{}(repo).(interface { + ClaimKubernetesEnvironment(context.Context, string, string, int64, string) (*models.KubernetesEnvironment, error) + SaveKubernetesEnvironment(context.Context, *models.KubernetesEnvironment, bool) error + GetKubernetesEnvironment(context.Context, string) (*models.KubernetesEnvironment, error) + }) + require.True(t, ok, "repository must support durable Kubernetes inventory claims") + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-claims", "env-claims") + claim, err := store.ClaimKubernetesEnvironment(ctx, "env-claims", "task-claims", 1, "create") + require.NoError(t, err) + _, err = store.ClaimKubernetesEnvironment(ctx, "env-claims", "task-claims", 1, "delete") + require.Error(t, err, "a competing operation must not acquire the resource") + claim.Metadata = map[string]interface{}{"pod_uid": "original"} + require.NoError(t, store.SaveKubernetesEnvironment(ctx, claim, true)) + replacement, err := store.ClaimKubernetesEnvironment(ctx, "env-claims", "task-claims", 1, "delete") + require.NoError(t, err) + require.Error(t, store.SaveKubernetesEnvironment(ctx, claim, true), "stale operation must not overwrite a replacement") + got, err := store.GetKubernetesEnvironment(ctx, "env-claims") + require.NoError(t, err) + require.Equal(t, replacement.OperationID, got.OperationID) + require.Equal(t, "original", got.Metadata["pod_uid"]) + _, err = store.ClaimKubernetesEnvironment(ctx, "env-claims", "foreign-task", 1, "foreign") + require.Error(t, err) + _, err = store.ClaimKubernetesEnvironment(ctx, "env-claims", "task-claims", 2, "stale-generation") + require.Error(t, err) +} + +func TestKubernetesEnvironmentRestartReplay(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-replay-kube", "env-replay-kube") + claim, err := repo.ClaimKubernetesEnvironment(ctx, "env-replay-kube", "task-replay-kube", 1, "operation") + require.NoError(t, err) + claim.Metadata = map[string]interface{}{"pod_uid": "retained"} + claim.ControlSecretID = "encrypted-control-reference" + claim.BootstrapSecretID = "encrypted-bootstrap-reference" + require.NoError(t, repo.SaveKubernetesEnvironment(ctx, claim, false)) + require.NoError(t, repo.runMigrations(ctx)) + recovered, err := repo.ClaimKubernetesEnvironment(ctx, claim.EnvironmentID, claim.TaskID, 1, "operation") + require.NoError(t, err) + require.Equal(t, claim, recovered) +} + +func TestPostgresKubernetesEnvironmentClaimExclusion(t *testing.T) { + first, second, _ := newTaskPostgresRepoPair(t) + ctx := context.Background() + seedRecoveryClaimEnvironment(t, first, "task-pg-kube", "env-pg-kube") + claim, err := first.ClaimKubernetesEnvironment(ctx, "env-pg-kube", "task-pg-kube", 1, "create") + require.NoError(t, err) + _, err = second.ClaimKubernetesEnvironment(ctx, "env-pg-kube", "task-pg-kube", 1, "delete") + require.ErrorIs(t, err, models.ErrKubernetesEnvironmentConflict) + require.NoError(t, first.SaveKubernetesEnvironment(ctx, claim, true)) + _, err = second.ClaimKubernetesEnvironment(ctx, "env-pg-kube", "task-pg-kube", 1, "delete") + require.NoError(t, err) +} + +func TestKubernetesEnvironmentRejectsChangedOwnerCheckpoint(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-owner-kube", "env-owner-kube") + claim, err := repo.ClaimKubernetesEnvironment(ctx, "env-owner-kube", "task-owner-kube", 1, "create") + require.NoError(t, err) + _, err = repo.db.ExecContext(ctx, `UPDATE task_environments SET ownership_generation = 2 WHERE id = 'env-owner-kube'`) + require.NoError(t, err) + require.ErrorIs(t, repo.SaveKubernetesEnvironment(ctx, claim, true), models.ErrKubernetesEnvironmentConflict) +} + +func TestKubernetesEnvironmentDeletionRequiresCurrentClaim(t *testing.T) { + repo := newRepoForEntityTests(t) + deleter, ok := interface{}(repo).(interface { + DeleteKubernetesEnvironment(context.Context, *models.KubernetesEnvironment) error + }) + require.True(t, ok, "resource cleanup requires a conditional inventory release") + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-delete-kube", "env-delete-kube") + claim, err := repo.ClaimKubernetesEnvironment(ctx, "env-delete-kube", "task-delete-kube", 1, "delete") + require.NoError(t, err) + stale := *claim + stale.Revision-- + require.ErrorIs(t, deleter.DeleteKubernetesEnvironment(ctx, &stale), models.ErrKubernetesEnvironmentConflict) + require.NoError(t, deleter.DeleteKubernetesEnvironment(ctx, claim)) + require.NoError(t, repo.DeleteTaskEnvironment(ctx, "env-delete-kube")) +} + +func TestKubernetesEnvironmentInterruptedOperationRecovery(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-interrupted", "env-interrupted") + record, err := repo.ClaimKubernetesEnvironment(ctx, "env-interrupted", "task-interrupted", 1, "old-process") + require.NoError(t, err) + record.Metadata = map[string]interface{}{"pod_uid": "retained"} + require.NoError(t, repo.SaveKubernetesEnvironment(ctx, record, false)) + recovery, ok := interface{}(repo).(interface{ RecoverInterruptedKubernetesOperations(context.Context) error }) + require.True(t, ok, "startup under exclusive runtime ownership must fence interrupted operations") + require.NoError(t, recovery.RecoverInterruptedKubernetesOperations(ctx)) + require.ErrorIs(t, repo.SaveKubernetesEnvironment(ctx, record, true), models.ErrKubernetesEnvironmentConflict) + resumed, err := repo.ClaimKubernetesEnvironment(ctx, record.EnvironmentID, record.TaskID, 1, "new-process") + require.NoError(t, err) + require.Equal(t, "retained", resumed.Metadata["pod_uid"]) +} + +func TestKubernetesEnvironmentInventoryListsWithoutSessions(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-retained", "env-retained") + _, err := repo.ClaimKubernetesEnvironment(ctx, "env-retained", "task-retained", 1, "create") + require.NoError(t, err) + lister, ok := interface{}(repo).(interface { + ListKubernetesEnvironments(context.Context) ([]*models.KubernetesEnvironment, error) + }) + require.True(t, ok, "retained pods must be discoverable without a session row") + records, err := lister.ListKubernetesEnvironments(ctx) + require.NoError(t, err) + require.Len(t, records, 1) + require.Equal(t, "env-retained", records[0].EnvironmentID) +} + +func TestKubernetesEnvironmentInventorySurvivesTaskDeletion(t *testing.T) { + testKubernetesEnvironmentInventorySurvivesTaskDeletion(t, newRepoForEntityTests(t)) +} + +func TestPostgresKubernetesEnvironmentInventorySurvivesTaskDeletion(t *testing.T) { + repo, _, _ := newTaskPostgresRepoPair(t) + testKubernetesEnvironmentInventorySurvivesTaskDeletion(t, repo) +} + +func testKubernetesEnvironmentInventorySurvivesTaskDeletion(t *testing.T, repo *Repository) { + t.Helper() + ctx := context.Background() + seedRecoveryClaimEnvironment(t, repo, "task-deleted-kube", "env-deleted-kube") + record, err := repo.ClaimKubernetesEnvironment(ctx, "env-deleted-kube", "task-deleted-kube", 1, "create") + require.NoError(t, err) + record.Metadata = map[string]interface{}{"pod_uid": "retained"} + require.NoError(t, repo.SaveKubernetesEnvironment(ctx, record, true)) + require.NoError(t, repo.DeleteTask(ctx, record.TaskID)) + retained, err := repo.GetKubernetesEnvironment(ctx, record.EnvironmentID) + require.NoError(t, err) + require.Equal(t, record.Metadata, retained.Metadata) + _, err = repo.ClaimKubernetesEnvironment(ctx, record.EnvironmentID, record.TaskID, 1, "launch") + require.Error(t, err, "deleted owners cannot launch") + _, err = repo.ClaimKubernetesEnvironmentCleanup(ctx, record.EnvironmentID, record.TaskID, 2, "wrong-generation") + require.Error(t, err) + cleanup, err := repo.ClaimKubernetesEnvironmentCleanup(ctx, record.EnvironmentID, record.TaskID, 1, "delete") + require.NoError(t, err) + require.NoError(t, repo.SaveKubernetesEnvironment(ctx, cleanup, true)) + cleanup, err = repo.ClaimKubernetesEnvironmentCleanup(ctx, record.EnvironmentID, record.TaskID, 1, "retry") + require.NoError(t, err) + require.NoError(t, repo.DeleteKubernetesEnvironment(ctx, cleanup)) +} diff --git a/apps/backend/internal/task/repository/sqlite/task_environment.go b/apps/backend/internal/task/repository/sqlite/task_environment.go index 7caf4adfc4..147ee5287c 100644 --- a/apps/backend/internal/task/repository/sqlite/task_environment.go +++ b/apps/backend/internal/task/repository/sqlite/task_environment.go @@ -878,6 +878,9 @@ func (r *Repository) DeleteTaskEnvironment(ctx context.Context, id string) error if err := recoveryclaim.EnsureAvailableTx(ctx, r.db, tx, id); err != nil { return err } + if err := r.ensureKubernetesInventoryAbsent(ctx, tx, id); err != nil { + return err + } result, err := tx.ExecContext(ctx, r.db.Rebind(`DELETE FROM task_environments WHERE id = ?`), id) if err != nil { return err @@ -919,6 +922,9 @@ func (r *Repository) DeleteTaskEnvironmentsByTask(ctx context.Context, taskID st return err } for _, environmentID := range environmentIDs { + if err := r.ensureKubernetesInventoryAbsent(ctx, tx, environmentID); err != nil { + return err + } if err := recoveryclaim.EnsureAvailableTx(ctx, r.db, tx, environmentID); err != nil { return err } diff --git a/apps/backend/internal/task/service/kubernetes_environment_cleanup.go b/apps/backend/internal/task/service/kubernetes_environment_cleanup.go new file mode 100644 index 0000000000..cfbabd305d --- /dev/null +++ b/apps/backend/internal/task/service/kubernetes_environment_cleanup.go @@ -0,0 +1,63 @@ +package service + +import ( + "context" + "errors" + "fmt" + agentkubernetes "github.com/kandev/kandev/internal/agent/kubernetes" + "strings" + + "github.com/kandev/kandev/internal/task/models" +) + +func (s *Service) teardownKubernetesEnvironment(ctx context.Context, env *models.TaskEnvironment) error { + if env.ExecutorType != string(models.ExecutorTypeKubernetes) { + return nil + } + inventory, ok := s.taskEnvironments.(interface { + GetKubernetesEnvironment(context.Context, string) (*models.KubernetesEnvironment, error) + }) + if !ok { + return nil + } + record, err := inventory.GetKubernetesEnvironment(ctx, env.ID) + if errors.Is(err, models.ErrKubernetesEnvironmentNotFound) { + return nil + } + if err != nil { + return err + } + if record == nil || record.TaskID != env.TaskID || record.EnvironmentID != env.ID { + return models.ErrWorkspaceReuseUnsafe + } + destroyer, ok := s.envDestroyer.(interface { + DestroyKubernetesEnvironment(context.Context, *models.TaskEnvironment) error + }) + if !ok { + return fmt.Errorf("kubernetes environment destroyer is unavailable") + } + return destroyer.DestroyKubernetesEnvironment(ctx, env) +} + +func (s *Service) hasKubernetesEnvironmentInventory(ctx context.Context, executorID string) (bool, error) { + store, ok := s.taskEnvironments.(interface { + ListKubernetesEnvironments(context.Context) ([]*models.KubernetesEnvironment, error) + }) + if !ok { + return false, nil + } + records, err := store.ListKubernetesEnvironments(ctx) + if err != nil { + return false, err + } + for _, record := range records { + if record == nil { + continue + } + id, _ := record.Metadata[agentkubernetes.MetadataKeyResourceExecutorID].(string) + if strings.TrimSpace(id) == "" || id == executorID { + return true, nil + } + } + return false, nil +} diff --git a/apps/backend/internal/task/service/service_kubernetes_environment_cleanup_test.go b/apps/backend/internal/task/service/service_kubernetes_environment_cleanup_test.go new file mode 100644 index 0000000000..4e0e746940 --- /dev/null +++ b/apps/backend/internal/task/service/service_kubernetes_environment_cleanup_test.go @@ -0,0 +1,62 @@ +package service + +import ( + "context" + "errors" + "testing" + + "github.com/kandev/kandev/internal/task/models" + "github.com/stretchr/testify/require" +) + +type kubernetesInventoryEnvRepo struct { + stubEnvRepo + inventory *models.KubernetesEnvironment + inventoryErr error +} + +func (r *kubernetesInventoryEnvRepo) GetKubernetesEnvironment(context.Context, string) (*models.KubernetesEnvironment, error) { + return r.inventory, r.inventoryErr +} + +type kubernetesEnvironmentDestroyer struct { + stubDestroyer + deleted string + err error +} + +func (d *kubernetesEnvironmentDestroyer) DestroyKubernetesEnvironment(_ context.Context, env *models.TaskEnvironment) error { + d.deleted = env.ID + return d.err +} + +func TestKubernetesEnvironmentCleanupUsesTaskOwner(t *testing.T) { + for _, fail := range []bool{false, true} { + t.Run(map[bool]string{false: "successful teardown", true: "failed teardown retains inventory"}[fail], func(t *testing.T) { + env := &models.TaskEnvironment{ID: "env-kube", TaskID: "task-kube", ExecutorType: "k8s", OwnershipGeneration: 1} + destroyer := &kubernetesEnvironmentDestroyer{} + if fail { + destroyer.err = errors.New("cluster unavailable") + } + service := &Service{taskEnvironments: &kubernetesInventoryEnvRepo{inventory: &models.KubernetesEnvironment{EnvironmentID: env.ID, TaskID: env.TaskID}}, envDestroyer: destroyer} + err := service.teardownEnvironmentRuntimeResources(context.Background(), env) + require.Equal(t, env.ID, destroyer.deleted) + require.ErrorIs(t, err, destroyer.err) + }) + } +} + +func (r *kubernetesInventoryEnvRepo) ListKubernetesEnvironments(context.Context) ([]*models.KubernetesEnvironment, error) { + return []*models.KubernetesEnvironment{r.inventory}, r.inventoryErr +} + +func TestKubernetesRetainedEnvironmentBlocksExecutorDeletion(t *testing.T) { + service := &Service{taskEnvironments: &kubernetesInventoryEnvRepo{inventory: &models.KubernetesEnvironment{Metadata: map[string]interface{}{"kubernetes_resource_executor_id": "executor-kube"}}}} + checker, ok := interface{}(service).(interface { + hasKubernetesEnvironmentInventory(context.Context, string) (bool, error) + }) + require.True(t, ok, "executor mutation guard must include retained environments") + retained, err := checker.hasKubernetesEnvironmentInventory(context.Background(), "executor-kube") + require.NoError(t, err) + require.True(t, retained) +} diff --git a/apps/backend/internal/task/service/service_resources.go b/apps/backend/internal/task/service/service_resources.go index 737acb980d..edfae416d7 100644 --- a/apps/backend/internal/task/service/service_resources.go +++ b/apps/backend/internal/task/service/service_resources.go @@ -1955,6 +1955,9 @@ func (s *Service) DeleteExecutor(ctx context.Context, id string) error { } func (s *Service) hasExecutorRunningInventory(ctx context.Context, executorID string) (bool, error) { + if retained, err := s.hasKubernetesEnvironmentInventory(ctx, executorID); err != nil || retained { + return retained, err + } running, err := s.executors.ListExecutorsRunning(ctx) if err != nil { return false, err diff --git a/apps/backend/internal/task/service/service_task_environments.go b/apps/backend/internal/task/service/service_task_environments.go index 297da2a56e..fbb27c1ec7 100644 --- a/apps/backend/internal/task/service/service_task_environments.go +++ b/apps/backend/internal/task/service/service_task_environments.go @@ -394,6 +394,10 @@ func (s *Service) teardownEnvironmentResourcesWithWorktrees( if cause := context.Cause(ctx); cause != nil { return cause } + if err := s.teardownKubernetesEnvironment(ctx, env); err != nil { + return err + } + worktreeIDs := environmentWorktreeIDs(env) if !environmentHasResources(env, includeWorktrees, worktreeIDs) { return nil diff --git a/apps/web/components/settings/kubernetes-session-lists.tsx b/apps/web/components/settings/kubernetes-session-lists.tsx index d1f3e87d08..c447af5586 100644 --- a/apps/web/components/settings/kubernetes-session-lists.tsx +++ b/apps/web/components/settings/kubernetes-session-lists.tsx @@ -28,7 +28,7 @@ export function MobileSessionList({ sessions }: { sessions: KubernetesSession[]
{sessions.map((session) => ( {sessions.map((session) => { - const expanded = expandedSessionId === session.session_id; + const rowKey = `${session.task_id}:${session.session_id}`; + const expanded = expandedSessionId === rowKey; return ( - + setExpandedSessionId(expanded ? null : session.session_id)} + onClick={() => setExpandedSessionId(expanded ? null : rowKey)} >