fix: resolve stale execution ID after backend restart (#473)
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<SessionPage> {
|
||||
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 });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user