From 4ef775afc93f10195c113cc1d5d945fea4b870e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Almeida?= <57680069+zeval@users.noreply.github.com> Date: Tue, 15 Sep 2026 22:09:33 +0200 Subject: [PATCH] fix(kubernetes): recover sessions after lost Pods (#3703) --- .../runtime/lifecycle/bootstrap_failure.go | 4 + .../runtime/lifecycle/manager_interaction.go | 24 ++++- ...ubernetes_resume_bootstrap_cleanup_test.go | 88 +++++++++++++++++++ .../agent/runtime/lifecycle/session.go | 63 ++++++++++++- .../lifecycle/session_load_failure_test.go | 14 +++ .../lifecycle/session_missing_rollout_test.go | 67 ++++++++++++++ .../orchestrator/executor/executor_execute.go | 8 +- .../orchestrator/executor/executor_test.go | 7 +- 8 files changed, 264 insertions(+), 11 deletions(-) create mode 100644 apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_resume_bootstrap_cleanup_test.go create mode 100644 apps/backend/internal/agent/runtime/lifecycle/session_missing_rollout_test.go diff --git a/apps/backend/internal/agent/runtime/lifecycle/bootstrap_failure.go b/apps/backend/internal/agent/runtime/lifecycle/bootstrap_failure.go index 43d53d6d07..6d4bbde397 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/bootstrap_failure.go +++ b/apps/backend/internal/agent/runtime/lifecycle/bootstrap_failure.go @@ -7,6 +7,10 @@ import ( "github.com/kandev/kandev/internal/task/models" ) +// StopReasonAgentBootstrapFailed lets the lifecycle manager reclaim a fresh +// failed launch while preserving an already-retained Kubernetes runtime. +const StopReasonAgentBootstrapFailed = "agent bootstrap failed" + // BootstrapFailure carries safe, operation-boundary evidence for a failure // before the agent becomes ready. Cause remains available to backend logging // and errors.Is/errors.As callers; user-facing projections must use Code and diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go index e0ccde0744..ae1c4ad0eb 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go @@ -1062,7 +1062,15 @@ func (m *Manager) StopAgentWithReason(ctx context.Context, executionID string, r if current, currentExists := m.executionStore.Get(executionID); !currentExists || current != execution { return fmt.Errorf("execution %q not found: %w", executionID, ErrExecutionNotFound) } - activityLease, err := m.acquireActivity(ctx, activity.KindExecutionStopping) + backendForce := force + stopCtx := ctx + if shouldPreserveFailedKubernetesResume(execution, reason) { + backendForce = false + var cancelStop context.CancelFunc + stopCtx, cancelStop = kubernetesDurableContext(ctx) + defer cancelStop() + } + activityLease, err := m.acquireActivity(stopCtx, activity.KindExecutionStopping) if err != nil { return err } @@ -1093,18 +1101,21 @@ func (m *Manager) StopAgentWithReason(ctx context.Context, executionID string, r zap.String("execution_id", executionID), zap.String("reason", reason), zap.Bool("force", force), + zap.Bool("runtime_force", backendForce), zap.Stringer("runtime", execution.RuntimeName)) // Try to gracefully stop via agentctl first, then always close connections. - agentStopFailed := m.stopExecutionAgentctl(ctx, executionID, execution, force) + // A retained Kubernetes resume gets a bounded non-cancelled opportunity to + // stop the failed process before its Pod is preserved for another retry. + agentStopFailed := m.stopExecutionAgentctl(stopCtx, executionID, execution, backendForce) // Stop the agent execution via the runtime that created it. A failed stop // must remain tracked: removing it here would turn a retryable cleanup into // an unobservable orphan process. - if err := m.stopAgentViaBackend(ctx, executionID, execution, reason, force, agentStopFailed); err != nil { + if err := m.stopAgentViaBackend(stopCtx, executionID, execution, reason, backendForce, agentStopFailed); err != nil { return fmt.Errorf("stop runtime for execution %q: %w", executionID, err) } - if execution.RuntimeName == executor.NameKubernetes && (force || shouldRunExecutorCleanup(reason)) { + if execution.RuntimeName == executor.NameKubernetes && (backendForce || shouldRunExecutorCleanup(reason)) { cleanupCtx, cancelCleanup := kubernetesDurableContext(ctx) err := m.deleteKubernetesRuntimeSecrets(cleanupCtx, execution.MetadataSnapshot()) cancelCleanup() @@ -1144,6 +1155,11 @@ func (m *Manager) StopAgentWithReason(ctx context.Context, executionID string, r return nil } +func shouldPreserveFailedKubernetesResume(execution *AgentExecution, reason string) bool { + return execution != nil && execution.RuntimeName == executor.NameKubernetes && + execution.isResumedSession && reason == StopReasonAgentBootstrapFailed +} + // detachAgentExecution implements the AC-EXECUTORS-SURVIVAL survivable-detach // branch of StopAgentWithReason (design 01/02 kill-path #4): when the // agent-survival capability is enabled and the stop reason is graceful diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_resume_bootstrap_cleanup_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_resume_bootstrap_cleanup_test.go new file mode 100644 index 0000000000..bacb5e66a6 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_kubernetes_resume_bootstrap_cleanup_test.go @@ -0,0 +1,88 @@ +package lifecycle + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/agent/executor" + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/secrets" +) + +func TestFailedKubernetesResumePreservesRetainedRuntimeAndSecrets(t *testing.T) { + log := newTestRegistryLogger() + stopRequests := make(chan string, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + stopRequests <- r.Method + " " + r.URL.Path + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true}`)) + })) + t.Cleanup(server.Close) + host, port := parseHTTPURL(t, server.URL) + + executorRegistry := NewExecutorRegistry(log) + backend := &stopTrackingExecutor{MockExecutor: MockExecutor{name: executor.NameKubernetes}} + executorRegistry.Register(backend) + mgr := NewManager(newTestRegistry(), &MockEventBus{}, executorRegistry, nil, nil, nil, ExecutorFallbackWarn, "", log) + cleanupManagerStopCh(t, mgr) + + secretStore := newInMemorySecretStore() + for _, id := range []string{"runtime-auth", "runtime-bootstrap"} { + require.NoError(t, secretStore.Create(context.Background(), &secrets.SecretWithValue{ + Secret: secrets.Secret{ID: id, Name: id}, Value: "secret", + })) + } + mgr.SetSecretStore(secretStore) + require.NoError(t, mgr.executionStore.Add(&AgentExecution{ + ID: "execution-1", TaskID: "task-1", SessionID: "session-1", + RuntimeName: executor.NameKubernetes, isResumedSession: true, + agentctl: agentctl.NewClient(host, port, log), + metadata: map[string]interface{}{ + MetadataKeyAuthTokenSecret: "runtime-auth", + MetadataKeyBootstrapNonceSecret: "runtime-bootstrap", + }, + })) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := mgr.StopAgentWithReason(ctx, "execution-1", StopReasonAgentBootstrapFailed, true) + + require.NoError(t, err) + select { + case request := <-stopRequests: + require.Equal(t, "POST /api/v1/stop", request) + case <-time.After(time.Second): + t.Fatal("the failed process was not stopped before its Kubernetes runtime was retained") + } + require.Equal(t, []bool{false}, backend.forces, + "failed resume cleanup must only close process-local Kubernetes connections") + require.Len(t, secretStore.store, 2, "retained runtime secrets must remain available for retry") + _, exists := mgr.executionStore.Get("execution-1") + require.False(t, exists, "failed execution must release its in-memory session slot") +} + +func TestFailedFreshKubernetesLaunchStillCleansItsRuntime(t *testing.T) { + log := newTestRegistryLogger() + executorRegistry := NewExecutorRegistry(log) + backend := &stopTrackingExecutor{MockExecutor: MockExecutor{name: executor.NameKubernetes}} + executorRegistry.Register(backend) + mgr := NewManager(newTestRegistry(), &MockEventBus{}, executorRegistry, nil, nil, nil, ExecutorFallbackWarn, "", log) + cleanupManagerStopCh(t, mgr) + require.NoError(t, mgr.executionStore.Add(&AgentExecution{ + ID: "execution-1", TaskID: "task-1", SessionID: "session-1", + RuntimeName: executor.NameKubernetes, + })) + + err := mgr.StopAgentWithReason( + context.Background(), "execution-1", StopReasonAgentBootstrapFailed, true, + ) + + require.NoError(t, err) + require.Equal(t, []bool{true}, backend.forces, + "a failed fresh launch must reclaim the runtime it created") +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/session.go b/apps/backend/internal/agent/runtime/lifecycle/session.go index b3c5fc7647..a6d595a57b 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session.go @@ -4,6 +4,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "encoding/json" "errors" "fmt" "sort" @@ -230,7 +231,7 @@ func (sm *SessionManager) createOrLoadSession( // the provider conversation. An internal error, timeout, cancellation, // authentication failure, or unknown transport error is inconclusive: // preserve the stored identity so a later retry can load it. - if !isSessionLoadFallbackErr(err) { + if !isSessionLoadFallbackErr(err, existingSessionID) { sm.logger.Warn("session/load failed with an inconclusive error, preserving session identity", zap.String("agent_type", agentConfig.ID()), zap.String("existing_session_id", existingSessionID), @@ -249,7 +250,8 @@ func (sm *SessionManager) createOrLoadSession( zap.String("reason", err.Error()), zap.Bool("method_not_found", isMethodNotFoundErr(err)), zap.Bool("capability_mismatch", hasCanonicalSessionLoadMessage(err, "agent does not support session loading (LoadSession capability is false)")), - zap.Bool("session_unknown", isSessionUnknownErr(err))) + zap.Bool("session_unknown", isSessionUnknownErr(err)), + zap.Bool("provider_session_missing", isMissingProviderRolloutErr(err, existingSessionID))) return sm.createNewSession(ctx, client, agentConfig, workspacePath, mcpServers) } return sm.createNewSession(ctx, client, agentConfig, workspacePath, mcpServers) @@ -1887,11 +1889,12 @@ func isSessionUnknownErr(err error) bool { // which replacing the provider conversation is known to be safe. Errors from // the agentctl WebSocket boundary are message-only, so retain the structured // ACP checks and match only their canonical projected messages here. -func isSessionLoadFallbackErr(err error) bool { +func isSessionLoadFallbackErr(err error, expectedSessionID string) bool { if err == nil { return false } - if isMethodNotFoundErr(err) || isSessionUnknownErr(err) { + if isMethodNotFoundErr(err) || isSessionUnknownErr(err) || + isMissingProviderRolloutErr(err, expectedSessionID) { return true } return hasCanonicalSessionLoadMessage(err, "Method not found") || @@ -1899,6 +1902,58 @@ func isSessionLoadFallbackErr(err error) bool { hasCanonicalSessionLoadMessage(err, "Resource not found") } +const ( + jsonRPCInternalError = -32603 + missingProviderRolloutPrefix = "no rollout found for thread id " +) + +type sessionLoadRequestError struct { + Code int `json:"code"` + Message string `json:"message"` + Data struct { + Details string `json:"details"` + } `json:"data"` +} + +// isMissingProviderRolloutErr recognizes Codex's explicit not-found response +// after its process-local rollout state disappeared. The session ID must match +// the one Kandev attempted to load; unrelated internal errors remain fatal. +func isMissingProviderRolloutErr(err error, expectedSessionID string) bool { + if err == nil || strings.TrimSpace(expectedSessionID) == "" { + return false + } + var reqErr *acp.RequestError + if errors.As(err, &reqErr) { + encoded, marshalErr := json.Marshal(reqErr) + if marshalErr == nil && matchesMissingProviderRollout(encoded, expectedSessionID) { + return true + } + } + message := err.Error() + for offset := strings.IndexByte(message, '{'); offset >= 0; { + candidate := message[offset:] + if matchesMissingProviderRollout([]byte(candidate), expectedSessionID) { + return true + } + next := strings.IndexByte(candidate[1:], '{') + if next < 0 { + break + } + offset += next + 1 + } + return false +} + +func matchesMissingProviderRollout(encoded []byte, expectedSessionID string) bool { + var projected sessionLoadRequestError + if err := json.Unmarshal(encoded, &projected); err != nil { + return false + } + return projected.Code == jsonRPCInternalError && + projected.Message == "Internal error" && + projected.Data.Details == missingProviderRolloutPrefix+expectedSessionID +} + func hasCanonicalSessionLoadMessage(err error, canonical string) bool { if err == nil { return false diff --git a/apps/backend/internal/agent/runtime/lifecycle/session_load_failure_test.go b/apps/backend/internal/agent/runtime/lifecycle/session_load_failure_test.go index 821af93ac0..171576411c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session_load_failure_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session_load_failure_test.go @@ -32,6 +32,15 @@ func TestInitializeSession_LoadFailureDoesNotCreateReplacement(t *testing.T) { name: "authentication failure", message: "authentication required", }, + { + name: "missing rollout for a different session", + message: `load session failed: failed to load session: {"code":-32603,"message":"Internal error",` + + `"data":{"details":"no rollout found for thread id another-session"}}`, + }, + { + name: "unstructured missing rollout phrase", + message: "internal error: no rollout found for thread id saved-session", + }, } for _, tt := range tests { @@ -106,6 +115,11 @@ func TestInitializeSession_LoadCompatibilityFailureCreatesReplacement(t *testing {name: "method not found", message: "method not found"}, {name: "capability mismatch", message: "agent does not support session loading (LoadSession capability is false)"}, {name: "unknown session", message: "Resource not found"}, + { + name: "missing provider rollout", + message: `load session failed: failed to load session: {"code":-32603,"message":"Internal error",` + + `"data":{"details":"no rollout found for thread id saved-session"}}`, + }, } for _, tt := range tests { diff --git a/apps/backend/internal/agent/runtime/lifecycle/session_missing_rollout_test.go b/apps/backend/internal/agent/runtime/lifecycle/session_missing_rollout_test.go new file mode 100644 index 0000000000..45fbadda36 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/session_missing_rollout_test.go @@ -0,0 +1,67 @@ +package lifecycle + +import ( + "errors" + "fmt" + "testing" + + "github.com/coder/acp-go-sdk" +) + +func TestIsMissingProviderRolloutErrRequiresMatchingStructuredEvidence(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + { + name: "matching wrapped request error", + err: fmt.Errorf("load failed: %w", &acp.RequestError{ + Code: -32603, + Message: "Internal error", + Data: map[string]any{ + "details": "no rollout found for thread id saved-session", + }, + }), + want: true, + }, + { + name: "different request code", + err: &acp.RequestError{ + Code: -32002, + Message: "Internal error", + Data: map[string]any{"details": "no rollout found for thread id saved-session"}, + }, + }, + { + name: "different request message", + err: &acp.RequestError{ + Code: -32603, + Message: "Provider failed", + Data: map[string]any{"details": "no rollout found for thread id saved-session"}, + }, + }, + { + name: "different session", + err: &acp.RequestError{ + Code: -32603, + Message: "Internal error", + Data: map[string]any{"details": "no rollout found for thread id other-session"}, + }, + }, + { + name: "matching JSON with trailing text", + err: errors.New( + `{"code":-32603,"message":"Internal error","data":{"details":"no rollout found for thread id saved-session"}} trailing`, + ), + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := isMissingProviderRolloutErr(test.err, "saved-session"); got != test.want { + t.Fatalf("isMissingProviderRolloutErr() = %t, want %t", got, test.want) + } + }) + } +} diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index efd089c9cd..81a2ed342e 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -331,7 +331,9 @@ func (e *Executor) claimForcedExecutionCleanup(sessionID, agentExecutionID strin } func (e *Executor) stopFailedStartExecution(ctx context.Context, agentExecutionID, phase string) { - if stopErr := e.agentManager.StopAgent(ctx, agentExecutionID, true); stopErr != nil { + if stopErr := e.agentManager.StopAgentWithReason( + ctx, agentExecutionID, lifecycle.StopReasonAgentBootstrapFailed, true, + ); stopErr != nil { e.logger.Warn("failed to clean up agent after "+phase, zap.String("agent_execution_id", agentExecutionID), zap.Error(stopErr)) @@ -424,7 +426,9 @@ func (e *Executor) stopUnstartedExecution(ctx context.Context, sessionID, agentE } stopCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) defer cancel() - if stopErr := e.agentManager.StopAgent(stopCtx, agentExecutionID, true); stopErr != nil { + if stopErr := e.agentManager.StopAgentWithReason( + stopCtx, agentExecutionID, lifecycle.StopReasonAgentBootstrapFailed, true, + ); stopErr != nil { e.logger.Warn("failed to stop unstarted agent execution", zap.String("session_id", sessionID), zap.String("agent_execution_id", agentExecutionID), diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index 99bec3ddb8..9744d0a8a7 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -2116,15 +2116,17 @@ func TestRunAgentProcessAsync_CleansUpOnStartFailure(t *testing.T) { var stopCalled atomic.Bool var stopForce atomic.Bool var stoppedExecutionID atomic.Value + var stopReason atomic.Value agentManager := &mockAgentManager{ startAgentProcessFunc: func(ctx context.Context, agentExecutionID string) error { return fmt.Errorf("ACP initialize handshake failed: context deadline exceeded") }, - stopAgentFunc: func(ctx context.Context, agentExecutionID string, force bool) error { + stopAgentWithReasonFunc: func(ctx context.Context, agentExecutionID, reason string, force bool) error { stopCalled.Store(true) stopForce.Store(force) stoppedExecutionID.Store(agentExecutionID) + stopReason.Store(reason) return nil }, } @@ -2189,6 +2191,9 @@ verified: if id, ok := stoppedExecutionID.Load().(string); !ok || id != "exec-456" { t.Errorf("expected StopAgent called with execution ID exec-456, got %v", stoppedExecutionID.Load()) } + if reason, ok := stopReason.Load().(string); !ok || reason != "agent bootstrap failed" { + t.Errorf("expected bootstrap-failure stop reason, got %v", stopReason.Load()) + } // Verify session was marked as FAILED session := repo.sessions["session-123"]