fix: recover from stale execution ID when auto-starting agent on prepared workspace (#690)
This commit is contained in:
@@ -134,6 +134,9 @@ type mockAgentManager struct {
|
||||
passthroughStdinErr error
|
||||
markPassthroughCalls []string // session IDs
|
||||
markPassthroughErr error
|
||||
|
||||
// Optional override for GetExecutionIDForSession
|
||||
getExecutionIDForSessionFunc func(context.Context, string) (string, error)
|
||||
}
|
||||
|
||||
type stopAgentCall struct {
|
||||
@@ -235,7 +238,10 @@ 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) {
|
||||
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")
|
||||
}
|
||||
func (m *mockAgentManager) GetGitLog(_ context.Context, _, _ string, _ int, _ string) (*client.GitLogResult, error) {
|
||||
|
||||
@@ -62,6 +62,7 @@ var (
|
||||
ErrExecutionAlreadyRunning = errors.New("execution already running")
|
||||
ErrNoCloneURL = errors.New("repository has no clone URL: provider owner and name are required")
|
||||
ErrTaskArchived = errors.New("task is archived")
|
||||
ErrStaleExecution = errors.New("stale execution: no live execution in memory")
|
||||
)
|
||||
|
||||
// PromptResult contains the result of a prompt operation
|
||||
|
||||
@@ -2,6 +2,7 @@ package executor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -344,8 +345,17 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses
|
||||
|
||||
// Fast path: workspace already launched (e.g., from PrepareSession with workspace).
|
||||
// Only start the agent subprocess if requested; otherwise return early.
|
||||
// If startAgentOnExistingWorkspace returns ErrStaleExecution, the in-memory execution
|
||||
// was lost (e.g. backend restart). The session's AgentExecutionID has been cleared,
|
||||
// so we fall through to the full LaunchAgent path below.
|
||||
if session.AgentExecutionID != "" {
|
||||
return e.startAgentOnExistingWorkspace(ctx, task, session, prompt, startAgent, opts.McpMode)
|
||||
result, err := e.startAgentOnExistingWorkspace(ctx, task, session, prompt, startAgent, opts.McpMode)
|
||||
if !errors.Is(err, ErrStaleExecution) {
|
||||
return result, err
|
||||
}
|
||||
e.logger.Info("falling through to full LaunchAgent after stale execution",
|
||||
zap.String("task_id", task.ID),
|
||||
zap.String("session_id", sessionID))
|
||||
}
|
||||
|
||||
repoInfo, err := e.resolvePrimaryRepoInfo(ctx, task.ID)
|
||||
@@ -556,7 +566,24 @@ func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.T
|
||||
// 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 {
|
||||
liveID, err := e.agentManager.GetExecutionIDForSession(ctx, session.ID)
|
||||
if err != nil {
|
||||
// No execution exists in memory (e.g. backend restarted since workspace was prepared).
|
||||
// Clear AgentExecutionID and return ErrStaleExecution so the caller falls through
|
||||
// to the full LaunchAgent path, which creates a complete execution with agent
|
||||
// commands, worktree, and all required configuration.
|
||||
e.logger.Info("stale execution ID, clearing for full re-launch",
|
||||
zap.String("session_id", session.ID),
|
||||
zap.String("stale_execution_id", executionID))
|
||||
session.AgentExecutionID = ""
|
||||
if updateErr := e.repo.UpdateTaskSession(ctx, session); updateErr != nil {
|
||||
e.logger.Warn("failed to clear stale execution ID",
|
||||
zap.String("session_id", session.ID),
|
||||
zap.Error(updateErr))
|
||||
}
|
||||
return nil, ErrStaleExecution
|
||||
}
|
||||
if 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),
|
||||
|
||||
@@ -325,6 +325,13 @@ func TestLaunchPreparedSession_ExistingWorkspace_StartAgent(t *testing.T) {
|
||||
}
|
||||
return nil
|
||||
},
|
||||
// The execution must exist in the in-memory store for startAgentOnExistingWorkspace to proceed.
|
||||
getExecutionIDForSessionFunc: func(ctx context.Context, sessionID string) (string, error) {
|
||||
if sessionID == "session-123" {
|
||||
return "exec-existing", nil
|
||||
}
|
||||
return "", fmt.Errorf("not found")
|
||||
},
|
||||
}
|
||||
agentManager.setExecutionDescriptionFunc = func(ctx context.Context, id, desc string) error {
|
||||
descriptionSet = desc
|
||||
@@ -432,6 +439,72 @@ func TestLaunchPreparedSession_StaleExecutionID_CorrectedFromLiveStore(t *testin
|
||||
}
|
||||
}
|
||||
|
||||
func TestLaunchPreparedSession_StaleExecution_FallsThroughToLaunchAgent(t *testing.T) {
|
||||
repo := newMockRepository()
|
||||
|
||||
// Session has an AgentExecutionID but no live execution exists in memory
|
||||
// (e.g. backend restarted since workspace was prepared).
|
||||
session := &models.TaskSession{
|
||||
ID: "session-123",
|
||||
TaskID: "task-123",
|
||||
AgentProfileID: "profile-123",
|
||||
AgentExecutionID: "stale-exec-id",
|
||||
State: models.TaskSessionStateCreated,
|
||||
StartedAt: time.Now(),
|
||||
UpdatedAt: time.Now(),
|
||||
}
|
||||
repo.sessions[session.ID] = session
|
||||
|
||||
var launchCalled atomic.Bool
|
||||
agentManager := &mockAgentManager{
|
||||
// No live execution — simulates backend restart
|
||||
getExecutionIDForSessionFunc: func(ctx context.Context, sessionID string) (string, error) {
|
||||
return "", fmt.Errorf("no execution found for session %s", sessionID)
|
||||
},
|
||||
launchAgentFunc: func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) {
|
||||
launchCalled.Store(true)
|
||||
return &LaunchAgentResponse{
|
||||
AgentExecutionID: "new-exec-id",
|
||||
Status: v1.AgentStatusStarting,
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
|
||||
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: "review this PR",
|
||||
StartAgent: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("LaunchPreparedSession should fall through to LaunchAgent, got error: %v", err)
|
||||
}
|
||||
|
||||
// Should have used LaunchAgent (full path) instead of failing
|
||||
if !launchCalled.Load() {
|
||||
t.Error("Expected LaunchAgent to be called after stale execution fallthrough")
|
||||
}
|
||||
|
||||
// Should use the new execution ID
|
||||
if execution.AgentExecutionID != "new-exec-id" {
|
||||
t.Errorf("Expected new execution ID 'new-exec-id', got %s", execution.AgentExecutionID)
|
||||
}
|
||||
|
||||
// DB should now hold the newly created execution ID, confirming the
|
||||
// full LaunchAgent path ran and overwrote the stale value.
|
||||
updatedSession := repo.sessions["session-123"]
|
||||
if updatedSession.AgentExecutionID != "new-exec-id" {
|
||||
t.Errorf("Expected DB AgentExecutionID to be 'new-exec-id', got %q", updatedSession.AgentExecutionID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteWithProfile_UsesPrepareThenLaunch(t *testing.T) {
|
||||
repo := newMockRepository()
|
||||
|
||||
|
||||
@@ -1048,11 +1048,20 @@ func TestEnsureSessionRunning_PreparedWorkspace(t *testing.T) {
|
||||
// when StartAgentProcess is called (simulating the agent starting successfully).
|
||||
startAgentProcessCalled := false
|
||||
wrappedMgr := &sessionUpdatingAgentManager{
|
||||
mockAgentManager: &mockAgentManager{isAgentRunning: false},
|
||||
repo: repo,
|
||||
sessionID: "session1",
|
||||
taskID: "task1",
|
||||
onStartCalled: &startAgentProcessCalled,
|
||||
mockAgentManager: &mockAgentManager{
|
||||
isAgentRunning: false,
|
||||
// Return the execution ID so the existing-workspace path proceeds
|
||||
getExecutionIDForSessionFunc: func(_ context.Context, sid string) (string, error) {
|
||||
if sid == "session1" {
|
||||
return "exec-prepare-1", nil
|
||||
}
|
||||
return "", fmt.Errorf("no execution found")
|
||||
},
|
||||
},
|
||||
repo: repo,
|
||||
sessionID: "session1",
|
||||
taskID: "task1",
|
||||
onStartCalled: &startAgentProcessCalled,
|
||||
}
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
|
||||
Reference in New Issue
Block a user