From 21db57bdfdb8a083d9dd3052916d0a115ce753dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 28 May 2026 21:48:12 +0100 Subject: [PATCH] fix(orchestrator): repair worktree state and prompt persistence on resume (#1121) --- .../server/adapter/transport/acp/adapter.go | 9 +- .../adapter/transport/acp/permission_race.go | 41 ++++ .../transport/acp/permission_race_test.go | 124 ++++++++++ .../executor/executor_environment_test.go | 2 +- .../orchestrator/executor/executor_execute.go | 6 + .../orchestrator/executor/executor_resume.go | 225 ++++++++++++------ .../executor/executor_resume_test.go | 197 ++++++++++++++- apps/backend/internal/orchestrator/service.go | 3 + .../internal/orchestrator/task_operations.go | 48 ++++ .../orchestrator/task_operations_test.go | 107 +++++++++ .../system/database/maintenance_test.go | 13 +- 11 files changed, 690 insertions(+), 85 deletions(-) create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race.go create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race_test.go diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go index cbbb337127..d3be8d1b04 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go @@ -2049,12 +2049,9 @@ func (a *Adapter) handlePermissionRequest(ctx context.Context, req *PermissionRe } // Only emit a synthetic tool_call event if no ToolCall notification preceded this. - // When a ToolCall notification exists (tracked in activeToolCalls), the message - // was already created by convertToolCallUpdate → handleToolCallEvent. - // Emitting a second tool_call for the same ID creates duplicate messages in the UI. - a.mu.RLock() - _, alreadyTracked := a.activeToolCalls[req.ToolCallID] - a.mu.RUnlock() + // waitForActiveToolCall bounds the race window between a SessionUpdate.ToolCall + // notification and a same-id request_permission dispatched on separate goroutines. + alreadyTracked := a.waitForActiveToolCall(ctx, req.ToolCallID, syntheticToolCallRaceWindow) if !alreadyTracked { toolCallEvent := AgentEvent{ diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race.go new file mode 100644 index 0000000000..7328a9f48d --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race.go @@ -0,0 +1,41 @@ +package acp + +import ( + "context" + "time" +) + +// syntheticToolCallRaceWindow bounds how long handlePermissionRequest waits +// for a concurrently-dispatched ToolCall notification to populate +// activeToolCalls before falling back to emitting a synthetic tool_call. Sized +// to cover the goroutine-scheduling gap between the SDK delivering a +// SessionUpdate.ToolCall and a request_permission for the same toolCallID, +// without adding noticeable latency to permission prompts. +const syntheticToolCallRaceWindow = 100 * time.Millisecond + +// waitForActiveToolCall polls activeToolCalls for the given id, sleeping in +// small increments, and returns true if the entry appears within timeout. +// Returns false when the wait expires without the entry materializing — the +// expected outcome when the agent really did skip the tool_call notification. +// Honors ctx so a cancelled session aborts the poll loop immediately rather +// than blocking the handler for the full timeout window. +func (a *Adapter) waitForActiveToolCall(ctx context.Context, toolCallID string, timeout time.Duration) bool { + const pollInterval = 10 * time.Millisecond + deadline := time.Now().Add(timeout) + for { + a.mu.RLock() + _, tracked := a.activeToolCalls[toolCallID] + a.mu.RUnlock() + if tracked { + return true + } + if time.Now().After(deadline) { + return false + } + select { + case <-ctx.Done(): + return false + case <-time.After(pollInterval): + } + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race_test.go new file mode 100644 index 0000000000..ea35d078ff --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/permission_race_test.go @@ -0,0 +1,124 @@ +package acp + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +// TestWaitForActiveToolCall_ReturnsOnContextCancel ensures a cancelled session +// breaks out of the poll loop immediately instead of blocking the handler +// until the timeout window expires. +func TestWaitForActiveToolCall_ReturnsOnContextCancel(t *testing.T) { + a := newTestAdapter() + ctx, cancel := context.WithCancel(t.Context()) + + go func() { + time.Sleep(5 * time.Millisecond) + cancel() + }() + + start := time.Now() + tracked := a.waitForActiveToolCall(ctx, "tool-1", time.Second) + elapsed := time.Since(start) + + if tracked { + t.Fatalf("expected waitForActiveToolCall to return false on cancellation, got true") + } + if elapsed >= 200*time.Millisecond { + t.Errorf("waitForActiveToolCall did not honor cancel: elapsed=%v", elapsed) + } +} + +// TestHandlePermissionRequest_NoDuplicateWhenToolCallNotificationLanded verifies +// the activeToolCalls check suppresses the synthetic tool_call emit when a +// SessionUpdate.ToolCall has already populated the map. Pre-existing guard; +// covered here so the race fix below has a baseline counterpart. +func TestHandlePermissionRequest_NoDuplicateWhenToolCallNotificationLanded(t *testing.T) { + a := newTestAdapter() + a.activeToolCalls["tool-1"] = &streams.NormalizedPayload{} + + _, _ = a.handlePermissionRequest(t.Context(), &PermissionRequest{ + SessionID: "sess-1", + ToolCallID: "tool-1", + Title: "Kandev: List Workspaces", + ActionType: "other", + }) + + for _, ev := range drainEvents(a) { + if ev.Type == streams.EventTypeToolCall { + t.Fatalf("did not expect synthetic tool_call when notification already tracked, got %#v", ev) + } + } +} + +// TestHandlePermissionRequest_WaitsForRacingToolCallNotification covers the +// race window: the SDK delivers a ToolCall SessionUpdate and a +// request_permission for the same toolCallID on separate goroutines, and the +// permission handler may run before the notification handler has populated +// activeToolCalls. Without the bounded wait, the handler would emit a +// synthetic tool_call that becomes a duplicate message in the chat. +func TestHandlePermissionRequest_WaitsForRacingToolCallNotification(t *testing.T) { + a := newTestAdapter() + + go func() { + time.Sleep(20 * time.Millisecond) + a.mu.Lock() + a.activeToolCalls["tool-1"] = &streams.NormalizedPayload{} + a.mu.Unlock() + }() + + _, _ = a.handlePermissionRequest(t.Context(), &PermissionRequest{ + SessionID: "sess-1", + ToolCallID: "tool-1", + Title: "Kandev: List Workspaces", + ActionType: "other", + }) + + for _, ev := range drainEvents(a) { + if ev.Type == streams.EventTypeToolCall { + t.Fatalf("waited for racing tool_call notification but still emitted synthetic %#v", ev) + } + } +} + +// TestHandlePermissionRequest_EmitsSyntheticAfterTimeout exercises the +// fallback path: when no ToolCall notification ever arrives, the bounded wait +// must give up and emit the synthetic tool_call so the UI still gets a row to +// attach the approval to. Agents that skip the ToolCall notification and go +// straight to request_permission rely on this path. +func TestHandlePermissionRequest_EmitsSyntheticAfterTimeout(t *testing.T) { + a := newTestAdapter() + + start := time.Now() + _, _ = a.handlePermissionRequest(t.Context(), &PermissionRequest{ + SessionID: "sess-1", + ToolCallID: "tool-1", + Title: "Kandev: List Workspaces", + ActionType: "other", + }) + elapsed := time.Since(start) + + if elapsed < syntheticToolCallRaceWindow { + t.Errorf("handler returned before the race window expired: elapsed=%v window=%v", elapsed, syntheticToolCallRaceWindow) + } + + var synthetic *AgentEvent + for _, ev := range drainEvents(a) { + if ev.Type == streams.EventTypeToolCall && ev.ToolCallID == "tool-1" { + ev := ev + synthetic = &ev + } + } + if synthetic == nil { + t.Fatalf("expected synthetic tool_call after timeout, got none") + } + if synthetic.ToolStatus != "pending_permission" { + t.Errorf("synthetic ToolStatus = %q, want %q", synthetic.ToolStatus, "pending_permission") + } + if synthetic.ToolTitle != "Kandev: List Workspaces" { + t.Errorf("synthetic ToolTitle = %q, want %q", synthetic.ToolTitle, "Kandev: List Workspaces") + } +} diff --git a/apps/backend/internal/orchestrator/executor/executor_environment_test.go b/apps/backend/internal/orchestrator/executor/executor_environment_test.go index af1348285b..1d1e0bd101 100644 --- a/apps/backend/internal/orchestrator/executor/executor_environment_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_environment_test.go @@ -269,7 +269,7 @@ func TestBuildResumeRequest_ReusesTaskEnvironmentRuntimeMetadata(t *testing.T) { }, } - req, _, _, running, err := exec.buildResumeRequest(context.Background(), task, session, true) + req, _, _, _, running, err := exec.buildResumeRequest(context.Background(), task, session, true) if err != nil { t.Fatalf("buildResumeRequest returned error: %v", err) } diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index 926201b3b2..7d1c0d7621 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -1071,6 +1071,12 @@ func (e *Executor) persistTaskEnvironment( if sandboxID := extractSandboxID(resp.Metadata); sandboxID != "" { existingEnv.SandboxID = sandboxID } + // Refresh TaskDirName when the request carries a new value — covers + // resume-after-failure where the original env row was stamped with an + // empty task_dir_name and the resume regenerates it. + if req.TaskDirName != "" { + existingEnv.TaskDirName = req.TaskDirName + } if err := e.repo.UpdateTaskEnvironment(ctx, existingEnv); err != nil { e.logger.Warn("failed to update task environment", zap.String("task_id", taskID), diff --git a/apps/backend/internal/orchestrator/executor/executor_resume.go b/apps/backend/internal/orchestrator/executor/executor_resume.go index a97497eb3f..029f6cd9bd 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume.go @@ -9,6 +9,7 @@ import ( "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/common/gitref" "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/worktree" v1 "github.com/kandev/kandev/pkg/api/v1" "go.uber.org/zap" ) @@ -325,7 +326,7 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio } defer unlock() - req, repositoryID, _, _, err := e.buildResumeRequest(ctx, task, session, startAgent) + req, repositoryID, execCfg, existingEnv, _, err := e.buildResumeRequest(ctx, task, session, startAgent) if err != nil { return nil, err } @@ -387,6 +388,16 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio e.persistResumeState(ctx, task.ID, session, startAgent) e.persistWorktreeAssociation(ctx, task.ID, session, repositoryID, resp) + // Refresh task_environments after a successful resume so the row reflects + // the new worktree, container, and ready status. Without this, sessions + // that failed on initial launch (empty task_dir_name / worktree_path, + // status=stopped) stay stuck on those stale values even though the resume + // just prepared a fresh worktree — the frontend keeps showing + // "Executor environment is unavailable (stopped)" and disables chat input. + // existingEnv is captured from buildResumeRequest above (resolveResumeTaskEnvironment + // already aborts on real DB errors); when nil, persistTaskEnvironment will + // re-fetch under the per-task lock and create a fresh row if needed. + e.persistTaskEnvironment(ctx, task.ID, session, existingEnv, req, resp, execCfg) worktreePath := resp.WorktreePath worktreeBranch := resp.WorktreeBranch @@ -493,7 +504,7 @@ func (e *Executor) validateAndLockResume(ctx context.Context, session *models.Ta // buildResumeRequest constructs the LaunchAgentRequest for a session resume, resolving executor config, // repository details, worktree settings, and ACP resume token. // Returns the request, repository ID, executor config, existing ExecutorRunning record (may be nil), and error. -func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, session *models.TaskSession, startAgent bool) (*LaunchAgentRequest, string, executorConfig, *models.ExecutorRunning, error) { +func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, session *models.TaskSession, startAgent bool) (*LaunchAgentRequest, string, executorConfig, *models.TaskEnvironment, *models.ExecutorRunning, error) { req := &LaunchAgentRequest{ TaskID: task.ID, SessionID: session.ID, @@ -523,15 +534,15 @@ func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, sessio existingEnv, err := e.resolveResumeTaskEnvironment(ctx, task.ID, session) if err != nil { - return nil, "", execConfig, nil, err + return nil, "", execConfig, nil, nil, err } if session.TaskEnvironmentID != "" { req.TaskEnvironmentID = session.TaskEnvironmentID } - repositoryID, err := e.applyResumeRepoConfig(ctx, task, session, req) + repositoryID, err := e.applyResumeRepoConfig(ctx, task, session, req, existingEnv) if err != nil { - return nil, "", execConfig, nil, err + return nil, "", execConfig, nil, nil, err } e.reuseExistingEnvironment(ctx, req, existingEnv) @@ -543,7 +554,7 @@ func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, sessio existingRunning := e.applyRunningRecordToResumeRequest(ctx, req, task, session, startAgent) - return req, repositoryID, execConfig, existingRunning, nil + return req, repositoryID, execConfig, existingEnv, existingRunning, nil } func (e *Executor) resolveResumeTaskEnvironment(ctx context.Context, taskID string, session *models.TaskSession) (*models.TaskEnvironment, error) { @@ -642,21 +653,16 @@ func (e *Executor) applyRunningRecordToResumeRequest(ctx context.Context, req *L } // applyResumeRepoConfig resolves repository details and applies them to req. -// Returns the resolved repositoryID. -func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, session *models.TaskSession, req *LaunchAgentRequest) (string, error) { - repositoryID := session.RepositoryID - if repositoryID == "" && len(task.Repositories) > 0 { - repositoryID = task.Repositories[0].RepositoryID - } - - baseBranch := session.BaseBranch - if baseBranch == "" && len(task.Repositories) > 0 && task.Repositories[0].BaseBranch != "" { - baseBranch = task.Repositories[0].BaseBranch - } +// Returns the resolved repositoryID. existingEnv is the task_environments row +// resolved upstream in buildResumeRequest; it is passed in so we can derive +// TaskDirName without an extra DB round-trip and so both the single-repo and +// multi-repo branches agree on the same fallback name (the fallback uses a +// random suffix, so two independent calls would produce different names). +func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, session *models.TaskSession, req *LaunchAgentRequest, existingEnv *models.TaskEnvironment) (string, error) { + repositoryID, baseBranch := resolveResumeRepoIDAndBranch(task, session) if baseBranch != "" { req.Branch = baseBranch } - if repositoryID == "" { return "", nil } @@ -671,6 +677,43 @@ func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, ses } repositoryPath := repository.LocalPath + applyResumeRepoBasics(req, repository, repositoryPath) + + if err := e.applyResumeCloneURL(req, repository, baseBranch); err != nil { + return "", err + } + + if shouldUseWorktree(req.ExecutorType) && repositoryPath != "" { + e.applyResumeWorktreeConfig(ctx, task, req, repository, repositoryID, repositoryPath, baseBranch, existingEnv) + } + + if err := e.applyResumeMultiRepoConfig(ctx, task, req, existingEnv); err != nil { + return repositoryID, err + } + + return repositoryID, nil +} + +// resolveResumeRepoIDAndBranch picks the primary repositoryID and baseBranch +// for a resume, preferring the session's persisted values and falling back to +// the task's primary repository when the session row was created before those +// fields existed. +func resolveResumeRepoIDAndBranch(task *v1.Task, session *models.TaskSession) (string, string) { + repositoryID := session.RepositoryID + if repositoryID == "" && len(task.Repositories) > 0 { + repositoryID = task.Repositories[0].RepositoryID + } + baseBranch := session.BaseBranch + if baseBranch == "" && len(task.Repositories) > 0 && task.Repositories[0].BaseBranch != "" { + baseBranch = task.Repositories[0].BaseBranch + } + return repositoryID, baseBranch +} + +// applyResumeRepoBasics copies the repository's local path and setup script +// onto the request. Pulled out of applyResumeRepoConfig so the parent's +// cyclomatic complexity stays inside the lint budget. +func applyResumeRepoBasics(req *LaunchAgentRequest, repository *models.Repository, repositoryPath string) { if repositoryPath != "" { req.RepositoryURL = repositoryPath } @@ -680,68 +723,100 @@ func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, ses } req.Metadata[lifecycle.MetadataKeyRepoSetupScript] = repository.SetupScript } +} - if e.capabilities != nil && e.capabilities.RequiresCloneURL(req.ExecutorType) { - cloneURL := repositoryCloneURL(repository) - if cloneURL == "" { - return "", ErrNoCloneURL - } - req.RepositoryURL = cloneURL - if req.Metadata == nil { - req.Metadata = make(map[string]interface{}) - } - req.Metadata["repository_clone_url"] = cloneURL - - // Clone-based remote executors (Sprites, Docker, …) need BaseBranch - // in launch metadata so the prepare script's `git clone --branch ` - // resolves. The local executor must NOT receive BaseBranch on resume: - // LocalPreparer would force a `git fetch && git checkout` against - // the user's actual workspace and clobber the "use current state" UX. - // Worktree executors set their own BaseBranch in the block below. - if baseBranch != "" { - req.BaseBranch = baseBranch - } +// applyResumeCloneURL handles clone-based remote executors: it stamps the +// clone URL on the request and propagates BaseBranch into launch metadata so +// the prepare script's `git clone --branch ` resolves. Local executors skip +// this path so LocalPreparer doesn't clobber the "use current state" UX. +func (e *Executor) applyResumeCloneURL(req *LaunchAgentRequest, repository *models.Repository, baseBranch string) error { + if e.capabilities == nil || !e.capabilities.RequiresCloneURL(req.ExecutorType) { + return nil } - - if shouldUseWorktree(req.ExecutorType) && repositoryPath != "" { - req.UseWorktree = true - req.RepositoryPath = repositoryPath - req.RepositoryID = repositoryID - if baseBranch != "" { - req.BaseBranch = baseBranch - } else { - req.BaseBranch = defaultBaseBranch - } - // Carry forward CheckoutBranch from task repository (e.g. PR head branch) - primaryTaskRepo, _ := e.repo.GetPrimaryTaskRepository(ctx, task.ID) - if primaryTaskRepo != nil && primaryTaskRepo.CheckoutBranch != "" { - req.CheckoutBranch = primaryTaskRepo.CheckoutBranch - req.PRNumber = prNumberFromMetadata(primaryTaskRepo.Metadata) - } - req.WorktreeBranchPrefix = repository.WorktreeBranchPrefix - req.PullBeforeWorktree = repository.PullBeforeWorktree + cloneURL := repositoryCloneURL(repository) + if cloneURL == "" { + return ErrNoCloneURL } - - // Multi-repo: when the task has more than one repository, populate - // req.Repositories so the lifecycle preparer can resume/recreate each - // repo's worktree. The legacy top-level fields above stay populated - // from the primary for backwards compat. - if len(task.Repositories) > 1 { - allRepos, allErr := e.resolveAllRepoInfo(ctx, task.ID) - if allErr != nil { - return repositoryID, allErr - } - if len(allRepos) > 1 { - req.Repositories = buildRepoSpecs(allRepos) - // Stamp the per-repo TaskDirName so the preparer reuses the same - // task root that the original launch created. - if env, _ := e.repo.GetTaskEnvironmentByTaskID(ctx, task.ID); env != nil && env.TaskDirName != "" { - req.TaskDirName = env.TaskDirName - } - } + req.RepositoryURL = cloneURL + if req.Metadata == nil { + req.Metadata = make(map[string]interface{}) } + req.Metadata["repository_clone_url"] = cloneURL + if baseBranch != "" { + req.BaseBranch = baseBranch + } + return nil +} - return repositoryID, nil +// applyResumeMultiRepoConfig populates req.Repositories for multi-repo tasks +// so the lifecycle preparer can resume/recreate each repo's worktree. The +// legacy top-level fields stay populated from the primary for backwards +// compat. +func (e *Executor) applyResumeMultiRepoConfig(ctx context.Context, task *v1.Task, req *LaunchAgentRequest, existingEnv *models.TaskEnvironment) error { + if len(task.Repositories) <= 1 { + return nil + } + allRepos, err := e.resolveAllRepoInfo(ctx, task.ID) + if err != nil { + return err + } + if len(allRepos) > 1 { + req.Repositories = buildRepoSpecs(allRepos) + req.TaskDirName = resolveResumeTaskDirName(existingEnv, task) + } + return nil +} + +// applyResumeWorktreeConfig stamps the worktree-related fields on req for a +// single-repo worktree resume. Extracted from applyResumeRepoConfig to keep +// cognitive complexity within the lint budget. +func (e *Executor) applyResumeWorktreeConfig( + ctx context.Context, + task *v1.Task, + req *LaunchAgentRequest, + repository *models.Repository, + repositoryID, repositoryPath, baseBranch string, + existingEnv *models.TaskEnvironment, +) { + req.UseWorktree = true + req.RepositoryPath = repositoryPath + req.RepositoryID = repositoryID + if baseBranch != "" { + req.BaseBranch = baseBranch + } else { + req.BaseBranch = defaultBaseBranch + } + // Carry forward CheckoutBranch from task repository (e.g. PR head branch) + primaryTaskRepo, _ := e.repo.GetPrimaryTaskRepository(ctx, task.ID) + if primaryTaskRepo != nil && primaryTaskRepo.CheckoutBranch != "" { + req.CheckoutBranch = primaryTaskRepo.CheckoutBranch + req.PRNumber = prNumberFromMetadata(primaryTaskRepo.Metadata) + } + req.WorktreeBranchPrefix = repository.WorktreeBranchPrefix + req.PullBeforeWorktree = repository.PullBeforeWorktree + // Worktree manager requires TaskDirName and RepoName. Mirror the + // initial-launch path (applyRepositoryConfig) so resumes of single-repo + // tasks don't fail with ErrTaskDirRequired. Prefer a persisted + // TaskDirName so we reuse the same on-disk task root; fall back to a + // freshly generated one when the original launch failed before the + // environment was stamped. + if repository.Name != "" { + req.RepoName = repository.Name + } + req.TaskDirName = resolveResumeTaskDirName(existingEnv, task) +} + +// resolveResumeTaskDirName returns the per-task directory name to use when +// resuming. It prefers the value persisted on task_environments (so we reuse +// the same ~/.kandev/tasks/{name}/ root the initial launch created) and falls +// back to a fresh semantic name when the original launch failed before any +// environment was stamped. That fallback is what lets a previously failed +// session recover instead of looping on ErrTaskDirRequired. +func resolveResumeTaskDirName(existingEnv *models.TaskEnvironment, task *v1.Task) string { + if existingEnv != nil && existingEnv.TaskDirName != "" { + return existingEnv.TaskDirName + } + return worktree.SemanticWorktreeName(task.Title, worktree.SmallSuffix(3)) } // persistResumeState updates the session row after a successful resume launch. diff --git a/apps/backend/internal/orchestrator/executor/executor_resume_test.go b/apps/backend/internal/orchestrator/executor/executor_resume_test.go index 4c5ebe8b61..9c89a5539d 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "testing" "time" @@ -496,7 +497,7 @@ func TestApplyResumeRepoConfig_BaseBranchByExecutorType(t *testing.T) { ExecutorType: tc.executorType, } - if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req); err != nil { + if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, nil); err != nil { t.Fatalf("applyResumeRepoConfig: %v", err) } @@ -510,6 +511,200 @@ func TestApplyResumeRepoConfig_BaseBranchByExecutorType(t *testing.T) { } } +// TestApplyResumeRepoConfig_WorktreeStampsTaskDir locks in the fix for +// resumes of single-repo worktree tasks: the lifecycle preparer hands the +// request to worktree.Manager.Create, which rejects requests missing +// TaskDirName or RepoName with ErrTaskDirRequired. The initial-launch path +// (applyRepositoryConfig) sets both; the resume path must do the same, and +// must also be able to regenerate TaskDirName when the original launch +// failed before any task_environments row was stamped. +func TestApplyResumeRepoConfig_WorktreeStampsTaskDir(t *testing.T) { + t.Run("reuses persisted TaskDirName when present", func(t *testing.T) { + repo := newMockRepository() + repo.repositories["repo-1"] = &models.Repository{ + ID: "repo-1", + Name: "my-repo", + LocalPath: "/tmp/repo", + } + existingEnv := &models.TaskEnvironment{ + ID: "env-1", + TaskID: "task-1", + TaskDirName: "previously-stamped_abc", + } + repo.taskEnvironments["env-1"] = existingEnv + repo.tasks["task-1"] = &models.Task{ID: "task-1"} + task := &v1.Task{ID: "task-1", Title: "Fix login bug"} + session := &models.TaskSession{ + ID: "sess-1", + TaskID: "task-1", + RepositoryID: "repo-1", + BaseBranch: "main", + } + exec := newTestExecutor(t, &mockAgentManager{}, repo) + req := &LaunchAgentRequest{TaskID: "task-1", SessionID: "sess-1", ExecutorType: "worktree"} + + if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, existingEnv); err != nil { + t.Fatalf("applyResumeRepoConfig: %v", err) + } + + if req.TaskDirName != "previously-stamped_abc" { + t.Errorf("TaskDirName = %q, want %q", req.TaskDirName, "previously-stamped_abc") + } + if req.RepoName != "my-repo" { + t.Errorf("RepoName = %q, want %q", req.RepoName, "my-repo") + } + }) + + t.Run("regenerates TaskDirName when persisted value is empty", func(t *testing.T) { + // This is the failure mode from the bug report: the initial launch + // failed before stamping task_dir_name, so the env row exists with + // an empty TaskDirName. The resume must still be able to proceed. + repo := newMockRepository() + repo.repositories["repo-1"] = &models.Repository{ + ID: "repo-1", + Name: "my-repo", + LocalPath: "/tmp/repo", + } + repo.tasks["task-1"] = &models.Task{ID: "task-1"} + task := &v1.Task{ID: "task-1", Title: "Fix login bug"} + session := &models.TaskSession{ + ID: "sess-1", + TaskID: "task-1", + RepositoryID: "repo-1", + BaseBranch: "main", + } + exec := newTestExecutor(t, &mockAgentManager{}, repo) + req := &LaunchAgentRequest{TaskID: "task-1", SessionID: "sess-1", ExecutorType: "worktree"} + existingEnv := &models.TaskEnvironment{ + ID: "env-1", + TaskID: "task-1", + TaskDirName: "", + } + + if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, existingEnv); err != nil { + t.Fatalf("applyResumeRepoConfig: %v", err) + } + + if req.TaskDirName == "" { + t.Error("TaskDirName must not be empty; worktree.Manager.Create would reject the request") + } + // Semantic name format: {sanitized-title}_{suffix}, suffix is 3 chars. + if !strings.HasPrefix(req.TaskDirName, "fix-login-bug_") { + t.Errorf("TaskDirName = %q, want prefix %q", req.TaskDirName, "fix-login-bug_") + } + if req.RepoName != "my-repo" { + t.Errorf("RepoName = %q, want %q", req.RepoName, "my-repo") + } + }) + + t.Run("regenerates TaskDirName when no env row exists", func(t *testing.T) { + repo := newMockRepository() + repo.repositories["repo-1"] = &models.Repository{ + ID: "repo-1", + Name: "my-repo", + LocalPath: "/tmp/repo", + } + repo.tasks["task-1"] = &models.Task{ID: "task-1"} + task := &v1.Task{ID: "task-1", Title: "Fix login bug"} + session := &models.TaskSession{ + ID: "sess-1", + TaskID: "task-1", + RepositoryID: "repo-1", + BaseBranch: "main", + } + exec := newTestExecutor(t, &mockAgentManager{}, repo) + req := &LaunchAgentRequest{TaskID: "task-1", SessionID: "sess-1", ExecutorType: "worktree"} + + if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, nil); err != nil { + t.Fatalf("applyResumeRepoConfig: %v", err) + } + + if req.TaskDirName == "" { + t.Error("TaskDirName must not be empty even without an env row") + } + if req.RepoName != "my-repo" { + t.Errorf("RepoName = %q, want %q", req.RepoName, "my-repo") + } + }) +} + +// TestResumeSession_RefreshesStaleEnvironmentRow locks in the fix for the +// post-restart bug where a resume of a session that had previously failed mid- +// launch left task_environments stuck at status=stopped with empty +// task_dir_name / worktree_path. The frontend polls that row to decide +// whether the chat input is enabled, so the stale state stranded the UI on +// "Executor environment is unavailable (stopped)" even after the resume +// successfully prepared a fresh worktree. +func TestResumeSession_RefreshesStaleEnvironmentRow(t *testing.T) { + repo := newMockRepository() + taskID := "task-resume-env" + sessionID := "sess-resume-env" + + repo.repositories["repo-1"] = &models.Repository{ + ID: "repo-1", Name: "my-repo", LocalPath: "/repos/my-repo", + } + // Worktree executor: required for the resume to hit the worktree branch in + // applyResumeRepoConfig that stamps TaskDirName onto req. + repo.executors["exec-worktree"] = &models.Executor{ + ID: "exec-worktree", Type: models.ExecutorTypeWorktree, + } + repo.tasks[taskID] = &models.Task{ID: taskID, Title: "Resume after failure"} + repo.sessions[sessionID] = &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + AgentProfileID: "profile-1", + State: models.TaskSessionStateFailed, + ExecutorID: "exec-worktree", + RepositoryID: "repo-1", + BaseBranch: "main", + TaskEnvironmentID: "env-stale", + } + // Stale env row from the previous failed launch: status=stopped, no + // worktree_path, no task_dir_name. This is the exact shape produced when + // LaunchAgent fails before persistTaskEnvironment writes the worktree + // fields back. + repo.taskEnvironments["env-stale"] = &models.TaskEnvironment{ + ID: "env-stale", + TaskID: taskID, + Status: models.TaskEnvironmentStatusStopped, + WorktreePath: "", + TaskDirName: "", + } + + const newWorktreePath = "/home/u/.kandev/tasks/resume-after-failure_abc/my-repo" + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *LaunchAgentRequest) (*LaunchAgentResponse, error) { + return &LaunchAgentResponse{ + AgentExecutionID: "exec-new", + WorktreeID: "wt-new", + WorktreePath: newWorktreePath, + WorktreeBranch: "kandev/resume-after-failure", + Status: v1.AgentStatusStarting, + }, nil + }, + } + exec := newTestExecutor(t, agentMgr, repo) + + if _, err := exec.ResumeSession(context.Background(), repo.sessions[sessionID], true); err != nil { + t.Fatalf("ResumeSession: %v", err) + } + + env := repo.taskEnvironments["env-stale"] + if env.Status != models.TaskEnvironmentStatusReady { + t.Errorf("env.Status = %q, want %q — without the refresh the frontend keeps showing the executor as unavailable", + env.Status, models.TaskEnvironmentStatusReady) + } + if env.WorktreePath != newWorktreePath { + t.Errorf("env.WorktreePath = %q, want %q", env.WorktreePath, newWorktreePath) + } + if env.WorktreeID != "wt-new" { + t.Errorf("env.WorktreeID = %q, want %q", env.WorktreeID, "wt-new") + } + if env.TaskDirName == "" { + t.Error("env.TaskDirName must not be empty after resume; the worktree manager needs it for the on-disk task root") + } +} + func TestIsTerminalSessionState(t *testing.T) { cases := []struct { state models.TaskSessionState diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 3f7b9f9f94..2cbb0b52be 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -181,6 +181,9 @@ type sessionExecutorStore interface { ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) ListNonTerminalSessionsByAgentInstance(ctx context.Context, agentInstanceID string) ([]*models.TaskSession, error) DeleteTaskSession(ctx context.Context, id string) error + // Messages — used by resume to backfill the initial user prompt when a + // prior launch failed before recordInitialMessage ran. + ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) // Workspace GetWorkspace(ctx context.Context, id string) (*models.Workspace, error) // Task environment diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 2df622c8e8..b9638aa61a 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -819,6 +819,34 @@ func (s *Service) applyWorkflowAndPlanMode(ctx context.Context, prompt string, t return effectivePrompt, planMode || stepHasPlanMode } +// backfillInitialUserMessageIfMissing records the task's description as the +// first user message when the session has no messages at all. This covers the +// edge case where the initial launch failed before recordInitialMessage was +// called (postLaunchStart only runs after a successful LaunchAgent), leaving +// the chat empty even though the task carries the prompt the user originally +// typed. +// +// The check requires *zero* messages, not just zero user messages: if any +// agent output already exists from a partial prior run, the backfilled +// message would land at the bottom of the chat (CreateMessage stamps +// CreatedAt=now), which is worse than leaving the chat alone. +func (s *Service) backfillInitialUserMessageIfMissing(ctx context.Context, taskID, sessionID, prompt string) { + if prompt == "" || s.messageCreator == nil { + return + } + msgs, err := s.repo.ListMessages(ctx, sessionID) + if err != nil { + s.logger.Warn("backfill initial message: list messages failed", + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + if len(msgs) > 0 { + return + } + s.recordInitialMessage(ctx, taskID, sessionID, prompt, false, false, nil) +} + // recordInitialMessage creates the initial user message and updates session state after launch. // autoStart marks the message as having been created by an automated trigger // (workflow auto-start, PR/issue watch, Jira/Linear integration) so cleanup @@ -954,6 +982,26 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin // Preserve persisted task/session state; resume should not mutate state/columns. execution.SessionState = v1.TaskSessionState(session.State) + // Backfill the initial user message when a prior failed launch never got + // to recordInitialMessage. Without this, the resume can succeed and the + // agent starts replying, but the chat shows agent output with no user + // prompt above it. + // + // We use task.Description (the raw user input) rather than the + // workflow-effective prompt produced by applyWorkflowAndPlanMode. The + // effective prompt may carry a plan-mode prefix or be templated through a + // workflow step, but reconstructing the exact prompt the original launch + // sent to the agent is brittle (workflow state may have advanced since). + // Surfacing the raw description is intentionally conservative: it shows + // what the user actually typed, which is what they expect to see in chat. + if task, taskErr := s.repo.GetTask(resumeCtx, taskID); taskErr != nil { + s.logger.Warn("resume: failed to load task for initial message backfill", + zap.String("task_id", taskID), + zap.Error(taskErr)) + } else if task != nil { + s.backfillInitialUserMessageIfMissing(resumeCtx, taskID, sessionID, task.Description) + } + s.logger.Debug("task session resumed and ready for input", zap.String("task_id", taskID), zap.String("session_id", sessionID)) diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 05f3c60da2..c31c18aaee 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -473,6 +473,113 @@ func (m *mockMessageCreator) AppendThinkingMessage(context.Context, string, stri } func (m *mockMessageCreator) InvalidateModelCache(string) {} +// --- backfillInitialUserMessageIfMissing --- + +func TestBackfillInitialUserMessageIfMissing_RecordsWhenSessionEmpty(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + + // Session has zero messages — backfill should record the prompt. + svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "original prompt") + + if len(mc.userMessages) != 1 { + t.Fatalf("expected 1 user message recorded, got %d", len(mc.userMessages)) + } + if mc.userMessages[0].content != "original prompt" { + t.Errorf("content = %q, want %q", mc.userMessages[0].content, "original prompt") + } +} + +func TestBackfillInitialUserMessageIfMissing_SkipsWhenUserMessageExists(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + + // Seed an existing user message — the backfill must be a no-op so a + // successful prior launch isn't duplicated on a subsequent resume. + if err := repo.CreateTurn(ctx, &models.Turn{ID: "turn1", TaskSessionID: "session1", TaskID: "task1", StartedAt: time.Now().UTC()}); err != nil { + t.Fatalf("create turn: %v", err) + } + if err := repo.CreateMessage(ctx, &models.Message{ + ID: "msg1", + TaskSessionID: "session1", + TaskID: "task1", + TurnID: "turn1", + AuthorType: models.MessageAuthorUser, + Content: "user already sent this", + CreatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("create message: %v", err) + } + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + + svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "would be a duplicate") + + if len(mc.userMessages) != 0 { + t.Fatalf("expected no user message recorded (one already exists), got %d", len(mc.userMessages)) + } +} + +// TestBackfillInitialUserMessageIfMissing_SkipsWhenAgentMessageExists covers +// the regression where a partial prior run produced agent output but never +// recorded the initial user message. Recording the user message now with +// CreatedAt=time.Now() would place it at the bottom of the chat (after the +// agent messages), which is worse than leaving the chat alone. +func TestBackfillInitialUserMessageIfMissing_SkipsWhenAgentMessageExists(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + + if err := repo.CreateTurn(ctx, &models.Turn{ID: "turn1", TaskSessionID: "session1", TaskID: "task1", StartedAt: time.Now().UTC()}); err != nil { + t.Fatalf("create turn: %v", err) + } + if err := repo.CreateMessage(ctx, &models.Message{ + ID: "agent-msg-1", + TaskSessionID: "session1", + TaskID: "task1", + TurnID: "turn1", + AuthorType: models.MessageAuthorAgent, + Content: "agent partial output from a prior run", + CreatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("create agent message: %v", err) + } + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + + svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "the original prompt") + + if len(mc.userMessages) != 0 { + t.Fatalf("expected no backfill when agent messages exist, got %d", len(mc.userMessages)) + } +} + +func TestBackfillInitialUserMessageIfMissing_SkipsEmptyPrompt(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + + svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "") + + if len(mc.userMessages) != 0 { + t.Fatalf("expected no user message for empty prompt, got %d", len(mc.userMessages)) + } +} + func TestRecordInitialMessage_DoesNotChangeSessionState(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) diff --git a/apps/backend/internal/system/database/maintenance_test.go b/apps/backend/internal/system/database/maintenance_test.go index 4599ca1a62..3d666fd4a3 100644 --- a/apps/backend/internal/system/database/maintenance_test.go +++ b/apps/backend/internal/system/database/maintenance_test.go @@ -107,7 +107,7 @@ func TestOptimize_StartsJobAndSucceeds(t *testing.T) { } func TestHandleVacuum_Returns202WithJobID(t *testing.T) { - svc, _, _, _ := newTestService(t) + svc, tracker, _, _ := newTestService(t) gin.SetMode(gin.TestMode) r := gin.New() r.POST("/vacuum", HandleVacuum(svc)) @@ -119,10 +119,15 @@ func TestHandleVacuum_Returns202WithJobID(t *testing.T) { if !contains(w.Body.String(), `"job_id"`) { t.Errorf("body missing job_id: %s", w.Body.String()) } + // Make sure the spawned job finishes before the test exits so the temp dir + // cleanup does not race with VACUUM still writing the SQLite journal. + for _, j := range tracker.List() { + waitForState(t, tracker, j.ID, jobs.StateSucceeded) + } } func TestHandleOptimize_Returns202WithJobID(t *testing.T) { - svc, _, _, _ := newTestService(t) + svc, tracker, _, _ := newTestService(t) gin.SetMode(gin.TestMode) r := gin.New() r.POST("/optimize", HandleOptimize(svc)) @@ -134,4 +139,8 @@ func TestHandleOptimize_Returns202WithJobID(t *testing.T) { if !contains(w.Body.String(), `"job_id"`) { t.Errorf("body missing job_id: %s", w.Body.String()) } + // Same cleanup-race guard as TestHandleVacuum_Returns202WithJobID. + for _, j := range tracker.List() { + waitForState(t, tracker, j.ID, jobs.StateSucceeded) + } }