From c6bf43ded9c839616b7deaa8c0f7702260946b9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 26 Mar 2026 15:11:23 +0000 Subject: [PATCH] fix: resolve stale execution ID after backend restart (#473) --- apps/backend/cmd/kandev/adapters.go | 4 + .../agent/lifecycle/manager_execution.go | 9 ++ .../simulated_agent_manager_test.go | 10 +++ .../orchestrator/event_handlers_test.go | 3 + .../orchestrator/executor/executor.go | 5 ++ .../orchestrator/executor/executor_execute.go | 35 ++++++-- .../orchestrator/executor/executor_test.go | 83 +++++++++++++++++-- .../orchestrator/scheduler/scheduler_test.go | 4 + apps/web/e2e/tests/executor-not-found.spec.ts | 75 +++++++++++++++++ 9 files changed, 215 insertions(+), 13 deletions(-) create mode 100644 apps/web/e2e/tests/executor-not-found.spec.ts diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index d60022937b..f149866c48 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -298,6 +298,10 @@ func (a *lifecycleAdapter) EnsureWorkspaceExecutionForSession(ctx context.Contex return err } +func (a *lifecycleAdapter) GetExecutionIDForSession(ctx context.Context, sessionID string) (string, error) { + return a.mgr.GetExecutionIDForSession(ctx, sessionID) +} + func (a *lifecycleAdapter) GetRemoteRuntimeStatusBySession(ctx context.Context, sessionID string) (*executor.RemoteRuntimeStatus, error) { status, ok := a.mgr.GetRemoteStatusBySessionID(ctx, sessionID) if !ok || status == nil { diff --git a/apps/backend/internal/agent/lifecycle/manager_execution.go b/apps/backend/internal/agent/lifecycle/manager_execution.go index 410a3dc4fe..a349ba46b2 100644 --- a/apps/backend/internal/agent/lifecycle/manager_execution.go +++ b/apps/backend/internal/agent/lifecycle/manager_execution.go @@ -77,6 +77,15 @@ func (m *Manager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID return execution, nil } +// GetExecutionIDForSession returns the execution ID for a session from the in-memory +// execution store. Returns empty string and error if no execution is found. +func (m *Manager) GetExecutionIDForSession(_ context.Context, sessionID string) (string, error) { + if execution, exists := m.executionStore.GetBySessionID(sessionID); exists { + return execution.ID, nil + } + return "", fmt.Errorf("no execution found for session %s", sessionID) +} + // EnsurePassthroughExecution ensures an execution exists for a passthrough session // and starts the passthrough process if needed. This is called when the terminal // handler receives a connection for a session that might need recovery after backend restart. diff --git a/apps/backend/internal/integration/simulated_agent_manager_test.go b/apps/backend/internal/integration/simulated_agent_manager_test.go index 9aff1fb793..7083b0c9f1 100644 --- a/apps/backend/internal/integration/simulated_agent_manager_test.go +++ b/apps/backend/internal/integration/simulated_agent_manager_test.go @@ -442,6 +442,16 @@ func (s *SimulatedAgentManagerClient) CleanupStaleExecutionBySessionID(ctx conte func (s *SimulatedAgentManagerClient) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID, sessionID string) error { return nil } +func (s *SimulatedAgentManagerClient) GetExecutionIDForSession(_ context.Context, sessionID string) (string, error) { + s.mu.Lock() + defer s.mu.Unlock() + for execID, inst := range s.instances { + if inst.sessionID == sessionID { + return execID, nil + } + } + return "", fmt.Errorf("no execution found for session %s", sessionID) +} func (s *SimulatedAgentManagerClient) GetGitLog(_ context.Context, _, _ string, _ int) (*client.GitLogResult, error) { return nil, nil } diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index c303dd9e48..f8c7817be5 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -208,6 +208,9 @@ func (m *mockAgentManager) CleanupStaleExecutionBySessionID(_ context.Context, _ func (m *mockAgentManager) EnsureWorkspaceExecutionForSession(_ context.Context, _, _ string) error { return nil } +func (m *mockAgentManager) GetExecutionIDForSession(_ context.Context, _ string) (string, error) { + return "", fmt.Errorf("no execution found") +} func (m *mockAgentManager) GetGitLog(_ context.Context, _, _ string, _ int) (*client.GitLogResult, error) { return nil, nil } diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 03b490d005..92cb27220d 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -150,6 +150,11 @@ type AgentManagerClient interface { // Used for restoring workspace access on terminal-state sessions. EnsureWorkspaceExecutionForSession(ctx context.Context, taskID, sessionID string) error + // GetExecutionIDForSession returns the execution ID for a session from the + // in-memory execution store. Returns empty string and error if not found. + // Used to detect stale AgentExecutionID values in the database after restart. + GetExecutionIDForSession(ctx context.Context, sessionID string) (string, error) + // GetGitLog retrieves the git log for a session from baseCommit to HEAD. // Used for archive snapshot capture. Returns nil, nil if no execution exists. GetGitLog(ctx context.Context, sessionID, baseCommit string, limit int) (*client.GitLogResult, error) diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index 6d245b7bde..f2ed4441e8 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -502,12 +502,31 @@ func (e *Executor) buildLaunchAgentRequest(ctx context.Context, task *v1.Task, s // startAgentOnExistingWorkspace handles the case where LaunchPreparedSession is called on a session // whose workspace (agentctl) was already launched. It optionally starts just the agent subprocess. func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.Task, session *models.TaskSession, prompt string, startAgent bool, mcpMode string) (*TaskExecution, error) { + // Resolve the actual execution ID from the in-memory store. After a backend + // restart, EnsureWorkspaceExecutionForSession may have created a new execution + // with a different ID than what the database still holds. Using the stale DB + // value would cause "execution not found" errors. + executionID := session.AgentExecutionID + if liveID, err := e.agentManager.GetExecutionIDForSession(ctx, session.ID); err == nil && liveID != "" && liveID != executionID { + e.logger.Info("correcting stale execution ID from DB with live in-memory value", + zap.String("session_id", session.ID), + zap.String("stale_id", executionID), + zap.String("live_id", liveID)) + executionID = liveID + session.AgentExecutionID = liveID + if updateErr := e.repo.UpdateTaskSession(ctx, session); updateErr != nil { + e.logger.Warn("failed to persist corrected execution ID", + zap.String("session_id", session.ID), + zap.Error(updateErr)) + } + } + if !startAgent { // Workspace already launched, nothing else to do now := time.Now().UTC() return &TaskExecution{ TaskID: task.ID, - AgentExecutionID: session.AgentExecutionID, + AgentExecutionID: executionID, AgentProfileID: session.AgentProfileID, StartedAt: session.StartedAt, SessionState: v1.TaskSessionState(session.State), @@ -518,10 +537,10 @@ func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.T // Update the task description in the existing execution so StartAgentProcess picks it up if prompt != "" { - if err := e.agentManager.SetExecutionDescription(ctx, session.AgentExecutionID, prompt); err != nil { + if err := e.agentManager.SetExecutionDescription(ctx, executionID, prompt); err != nil { e.logger.Warn("failed to set execution description for existing workspace", zap.String("session_id", session.ID), - zap.String("agent_execution_id", session.AgentExecutionID), + zap.String("agent_execution_id", executionID), zap.Error(err)) // Non-fatal: agent may start without description } @@ -534,10 +553,10 @@ func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.T effectiveMcpMode = McpModeConfig } if effectiveMcpMode != "" { - if err := e.agentManager.SetMcpMode(ctx, session.AgentExecutionID, effectiveMcpMode); err != nil { + if err := e.agentManager.SetMcpMode(ctx, executionID, effectiveMcpMode); err != nil { e.logger.Error("failed to set MCP mode for existing workspace", zap.String("session_id", session.ID), - zap.String("agent_execution_id", session.AgentExecutionID), + zap.String("agent_execution_id", executionID), zap.String("mcp_mode", effectiveMcpMode), zap.Error(err)) return nil, fmt.Errorf("set MCP mode %q: %w", effectiveMcpMode, err) @@ -557,7 +576,7 @@ func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.T execution := &TaskExecution{ TaskID: task.ID, - AgentExecutionID: session.AgentExecutionID, + AgentExecutionID: executionID, AgentProfileID: session.AgentProfileID, StartedAt: now, SessionState: v1.TaskSessionStateStarting, @@ -566,12 +585,12 @@ func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.T } // Start the agent process asynchronously - e.startAgentProcessAsync(ctx, task.ID, session.ID, session.AgentExecutionID) + e.startAgentProcessAsync(ctx, task.ID, session.ID, executionID) e.logger.Info("agent starting on existing workspace", zap.String("task_id", task.ID), zap.String("session_id", session.ID), - zap.String("agent_execution_id", session.AgentExecutionID)) + zap.String("agent_execution_id", executionID)) return execution, nil } diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index 161fc5be67..6a2519ce60 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -18,11 +18,12 @@ import ( // mockAgentManager implements AgentManagerClient for testing type mockAgentManager struct { - launchAgentFunc func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) - startAgentProcessFunc func(ctx context.Context, agentExecutionID string) error - stopAgentFunc func(ctx context.Context, agentExecutionID string, force bool) error - resolveAgentProfileFunc func(ctx context.Context, profileID string) (*AgentProfileInfo, error) - setExecutionDescriptionFunc func(ctx context.Context, agentExecutionID string, description string) error + launchAgentFunc func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) + startAgentProcessFunc func(ctx context.Context, agentExecutionID string) error + stopAgentFunc func(ctx context.Context, agentExecutionID string, force bool) error + resolveAgentProfileFunc func(ctx context.Context, profileID string) (*AgentProfileInfo, error) + setExecutionDescriptionFunc func(ctx context.Context, agentExecutionID string, description string) error + getExecutionIDForSessionFunc func(ctx context.Context, sessionID string) (string, error) } func (m *mockAgentManager) LaunchAgent(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { @@ -117,6 +118,12 @@ func (m *mockAgentManager) CleanupStaleExecutionBySessionID(ctx context.Context, func (m *mockAgentManager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID, sessionID string) error { return nil } +func (m *mockAgentManager) GetExecutionIDForSession(ctx context.Context, sessionID string) (string, error) { + if m.getExecutionIDForSessionFunc != nil { + return m.getExecutionIDForSessionFunc(ctx, sessionID) + } + return "", fmt.Errorf("no execution found for session %s", sessionID) +} func (m *mockAgentManager) ResolveAgentProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) { if m.resolveAgentProfileFunc != nil { @@ -951,6 +958,72 @@ func TestLaunchPreparedSession_ExistingWorkspace_StartAgent(t *testing.T) { } } +func TestLaunchPreparedSession_StaleExecutionID_CorrectedFromLiveStore(t *testing.T) { + repo := newMockRepository() + + // Session has a stale AgentExecutionID from a previous backend run. + // After restart, EnsureWorkspaceExecutionForSession created a new execution + // with a different ID, but the database still holds the old one. + session := &models.TaskSession{ + ID: "session-123", + TaskID: "task-123", + AgentProfileID: "profile-123", + AgentExecutionID: "stale-exec-id", // stale ID from DB + State: models.TaskSessionStateCreated, + StartedAt: time.Now(), + UpdatedAt: time.Now(), + } + repo.sessions[session.ID] = session + + var startedWithID atomic.Value + agentManager := &mockAgentManager{ + startAgentProcessFunc: func(ctx context.Context, id string) error { + startedWithID.Store(id) + return nil + }, + // Simulate the live execution store having a different ID + getExecutionIDForSessionFunc: func(ctx context.Context, sessionID string) (string, error) { + if sessionID == "session-123" { + return "live-exec-id", nil + } + return "", fmt.Errorf("not found") + }, + } + + executor := newTestExecutor(t, agentManager, repo) + + task := &v1.Task{ + ID: "task-123", + WorkspaceID: "workspace-123", + Title: "Test Task", + } + + execution, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", LaunchOptions{AgentProfileID: "profile-123", Prompt: "do work", StartAgent: true}) + if err != nil { + t.Fatalf("LaunchPreparedSession failed: %v", err) + } + + // Should use the live execution ID, not the stale one + if execution.AgentExecutionID != "live-exec-id" { + t.Errorf("Expected live execution ID 'live-exec-id', got %s", execution.AgentExecutionID) + } + + // Wait for async goroutine + time.Sleep(100 * time.Millisecond) + + // StartAgentProcess should be called with the live ID + got, _ := startedWithID.Load().(string) + if got != "live-exec-id" { + t.Errorf("Expected StartAgentProcess called with 'live-exec-id', got %q", got) + } + + // Database should be updated with the corrected ID + updatedSession := repo.sessions["session-123"] + if updatedSession.AgentExecutionID != "live-exec-id" { + t.Errorf("Expected DB AgentExecutionID to be corrected to 'live-exec-id', got %s", updatedSession.AgentExecutionID) + } +} + func TestExecuteWithProfile_UsesPrepareThenLaunch(t *testing.T) { repo := newMockRepository() diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index db9cd2352c..5e55875592 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -3,6 +3,7 @@ package scheduler import ( "context" "errors" + "fmt" "path/filepath" "sync" "testing" @@ -114,6 +115,9 @@ func (m *mockAgentManager) CleanupStaleExecutionBySessionID(ctx context.Context, func (m *mockAgentManager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID, sessionID string) error { return nil } +func (m *mockAgentManager) GetExecutionIDForSession(_ context.Context, _ string) (string, error) { + return "", fmt.Errorf("no execution found") +} func (m *mockAgentManager) CancelAgent(ctx context.Context, sessionID string) error { return nil diff --git a/apps/web/e2e/tests/executor-not-found.spec.ts b/apps/web/e2e/tests/executor-not-found.spec.ts new file mode 100644 index 0000000000..b03ed33c29 --- /dev/null +++ b/apps/web/e2e/tests/executor-not-found.spec.ts @@ -0,0 +1,75 @@ +import { test, expect } from "../fixtures/test-base"; +import { KanbanPage } from "../pages/kanban-page"; +import { SessionPage } from "../pages/session-page"; +import type { Page } from "@playwright/test"; + +/** Navigate to a kanban card by title and open its session page. */ +async function openTaskSession(page: Page, title: string): Promise { + const kanban = new KanbanPage(page); + await kanban.goto(); + + const card = kanban.taskCardByTitle(title); + await expect(card).toBeVisible({ timeout: 15_000 }); + await card.click(); + await expect(page).toHaveURL(/\/s\//, { timeout: 15_000 }); + + const session = new SessionPage(page); + await session.waitForLoad(); + return session; +} + +test.describe("Executor not found after backend restart", () => { + // Backend restart tests can be flaky + test.describe.configure({ retries: 1 }); + + test("agent starts successfully when sending new prompt after backend restart", async ({ + testPage, + apiClient, + seedData, + backend, + }) => { + test.setTimeout(120_000); + + // 1. Create task and start agent with a simple scenario + await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Executor Fix Task", + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + + // 2. Navigate to session and wait for agent to finish its first turn + const session = await openTaskSession(testPage, "Executor Fix Task"); + await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ + timeout: 30_000, + }); + await expect(session.idleInput()).toBeVisible({ timeout: 15_000 }); + + // 3. Restart the backend — clears in-memory execution store, + // but DB still has the old AgentExecutionID + await backend.restart(); + + // 4. Reload the page so SSR fetches from the new backend instance + await testPage.reload(); + await session.waitForLoad(); + + // 5. Wait for auto-resume to complete (workspace restoration) + await expect(session.idleInput()).toBeVisible({ timeout: 60_000 }); + + // 6. Send a NEW prompt — this triggers LaunchPreparedSession which + // reads the stale AgentExecutionID from DB and calls + // startAgentOnExistingWorkspace. Without the fix, this fails with + // "execution not found" and the session goes to FAILED state. + await session.sendMessage("/e2e:simple-message"); + + // 7. The agent should respond successfully (not fail with executor not found) + await expect( + session.chat.getByText("simple mock response", { exact: false }).nth(1), + ).toBeVisible({ timeout: 30_000 }); + }); +});