fix(backend): recreate the worktree when resuming an unarchived task (#2508)

Co-authored-by: Kandev Agent <agent@kandev.dev>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
João Salavisa
2026-08-11 00:19:40 +02:00
committed by GitHub
parent 6a093d8eae
commit a69a026741
4 changed files with 290 additions and 19 deletions
@@ -172,16 +172,41 @@ func isAgentAlreadyRunningError(err error) bool {
return err != nil && errors.Is(err, lifecycle.ErrAgentAlreadyRunning) return err != nil && errors.Is(err, lifecycle.ErrAgentAlreadyRunning)
} }
func validateSessionWorktrees(session *models.TaskSession) error { // missingSessionWorktreePaths returns the session's recorded worktree
// directories that are no longer on disk.
func missingSessionWorktreePaths(session *models.TaskSession) []string {
var missing []string
for _, wt := range session.Worktrees { for _, wt := range session.Worktrees {
if wt.WorktreePath == "" { if wt == nil || wt.WorktreePath == "" {
continue continue
} }
if _, err := os.Stat(wt.WorktreePath); err != nil { if _, err := os.Stat(wt.WorktreePath); err != nil {
return fmt.Errorf("worktree path not found: %w", err) missing = append(missing, wt.WorktreePath)
} }
} }
return nil return missing
}
// noteMissingWorktreesBeforeResume records that a resume is about to go
// through the worktree manager's recreate path.
//
// A missing directory is NOT a resume blocker. Archive cleanup deletes the
// worktree directory while keeping the environment repository row and the git
// branch (DestroyWorktree passes removeBranch=false), so *every* resume after
// an unarchive observes exactly this shape. worktree.Manager.Create reuses the
// stored record, restores the branch — locally, or by fetching it back from
// origin — and rebuilds the directory at the same path, which is what the
// recreate path was written for. Rejecting the resume here instead made
// unarchive a dead end: the session could never be resumed and the workspace
// was never restored, even though the work itself was still intact.
func (s *Service) noteMissingWorktreesBeforeResume(sessionID string, session *models.TaskSession) {
missing := missingSessionWorktreePaths(session)
if len(missing) == 0 {
return
}
s.logger.Info("worktree directory missing on resume; it will be recreated from the stored branch",
zap.String("session_id", sessionID),
zap.Strings("paths", missing))
} }
// EnqueueTask manually adds a task to the queue // EnqueueTask manually adds a task to the queue
@@ -1584,9 +1609,7 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin
// the record may already have been cleaned up before the user clicked Resume — allow it. // the record may already have been cleaned up before the user clicked Resume — allow it.
return nil, fmt.Errorf("session is not resumable: no executor record") return nil, fmt.Errorf("session is not resumable: no executor record")
} }
if err := validateSessionWorktrees(session); err != nil { s.noteMissingWorktreesBeforeResume(sessionID, session)
return nil, err
}
// Completed sessions cannot be restarted — they require a new session. // Completed sessions cannot be restarted — they require a new session.
// Failed and cancelled sessions keep the resume token so the relaunched // Failed and cancelled sessions keep the resume token so the relaunched
@@ -1970,9 +1993,7 @@ func (s *Service) attemptColdResume(
return false, fmt.Errorf("session is not resumable: no executor record (state: %s)", session.State) return false, fmt.Errorf("session is not resumable: no executor record (state: %s)", session.State)
} }
if err := validateSessionWorktrees(session); err != nil { s.noteMissingWorktreesBeforeResume(sessionID, session)
return false, err
}
// Use context.WithoutCancel to prevent WebSocket request timeout from canceling the resume. // Use context.WithoutCancel to prevent WebSocket request timeout from canceling the resume.
// The lifecycle layer publishes events.AgentBootReady (handled by handleAgentBootReady) // The lifecycle layer publishes events.AgentBootReady (handled by handleAgentBootReady)
@@ -2520,14 +2541,11 @@ func (s *Service) validateResumeEligibility(session *models.TaskSession, resp dt
return resp return resp
} }
// Check if worktree exists (if one was used) // A missing worktree directory does not make the session unresumable.
if len(session.Worktrees) > 0 && session.Worktrees[0].WorktreePath != "" { // Archive cleanup deletes the directory and keeps the branch, so reporting
if _, err := os.Stat(session.Worktrees[0].WorktreePath); err != nil { // IsResumable=false here would hide the resume affordance for every
resp.Error = "worktree not found" // unarchived task — see noteMissingWorktreesBeforeResume. The resume
resp.IsResumable = false // itself recreates the directory through the worktree manager.
return resp
}
}
// Don't auto-resume sessions in error-recovery state. // Don't auto-resume sessions in error-recovery state.
if isErrorRecoveryState(session) { if isErrorRecoveryState(session) {
@@ -0,0 +1,171 @@
package orchestrator
import (
"context"
"os"
"path/filepath"
"sync/atomic"
"testing"
"time"
"github.com/kandev/kandev/internal/orchestrator/executor"
"github.com/kandev/kandev/internal/task/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// TestResumeTaskSession_RecreatesMissingWorktreeAfterUnarchive pins the
// unarchive → resume path. Archiving a task deletes the worktree *directory*
// but keeps the environment repository row and the git branch, so every
// resume after an unarchive sees a recorded worktree path that no longer
// exists on disk. Rejecting the resume there made unarchive a dead end — the
// session could never be resumed even though the branch was still intact.
// The resume must reach the launch instead, where the worktree manager
// recreates the directory from the stored record.
func TestResumeTaskSession_RecreatesMissingWorktreeAfterUnarchive(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
// CANCELLED with the archive reason and no executors_running row: exactly
// what archive cleanup leaves behind for an unarchived task.
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCancelled)
session, err := repo.GetTaskSession(ctx, "session1")
if err != nil {
t.Fatalf("failed to load session: %v", err)
}
session.AgentProfileID = "profile1"
session.ErrorMessage = models.SessionArchiveCancelReason
session.TaskEnvironmentID = "env1"
if err := repo.UpdateTaskSession(ctx, session); err != nil {
t.Fatalf("failed to update session: %v", err)
}
if err := repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{
ID: "env1", TaskID: "task1", ExecutorType: "worktree",
WorkspacePath: t.TempDir(), Status: models.TaskEnvironmentStatusReady,
}); err != nil {
t.Fatalf("CreateTaskEnvironment: %v", err)
}
// The directory below is never created — the archive removed it.
missingPath := filepath.Join(t.TempDir(), "archived-task", "kandev")
if err := repo.CreateTaskEnvironmentRepo(ctx, &models.TaskEnvironmentRepo{
ID: "ter1",
TaskEnvironmentID: "env1",
WorktreeID: "wid1",
RepositoryID: "repo1",
WorktreePath: missingPath,
WorktreeBranch: "feature/archived-work",
}); err != nil {
t.Fatalf("CreateTaskEnvironmentRepo: %v", err)
}
if _, statErr := os.Stat(missingPath); statErr == nil {
t.Fatalf("test setup is wrong: %s must not exist on disk", missingPath)
}
var launched atomic.Int32
agentMgr := &mockAgentManager{
repoForExecutionLookup: repo,
isAgentReadyFn: func(_ context.Context, _ string) bool { return true },
launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
launched.Add(1)
sess, getErr := repo.GetTaskSession(context.Background(), req.SessionID)
if getErr == nil && sess != nil {
sess.State = models.TaskSessionStateWaitingForInput
sess.UpdatedAt = time.Now().UTC()
_ = repo.UpdateTaskSession(context.Background(), sess)
}
return &executor.LaunchAgentResponse{AgentExecutionID: "exec-unarchived"}, nil
},
}
taskRepo := newMockTaskRepo()
taskRepo.tasks["task1"] = &v1.Task{ID: "task1", Title: "Test Task", State: v1.TaskStateInProgress}
svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr)
svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
exec, err := svc.ResumeTaskSession(ctx, "task1", "session1")
if err != nil {
t.Fatalf("resume of an unarchived session must not be blocked by the missing worktree directory: %v", err)
}
if exec == nil {
t.Fatal("ResumeTaskSession returned nil execution")
}
if launched.Load() == 0 {
t.Fatal("expected the resume to reach LaunchAgent, where the worktree is recreated")
}
}
// TestGetTaskSessionStatus_ArchivedSessionResumableWithMissingWorktree covers
// the status half of the unarchive path: an archive-cancelled session whose
// executors_running row survived cleanup must still report IsResumable, even
// though its worktree directory is gone. Reporting "worktree not found" here
// hid both the auto-resume and the manual resume button, so an unarchived task
// had no way back at all.
func TestGetTaskSessionStatus_ArchivedSessionResumableWithMissingWorktree(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCancelled)
session, err := repo.GetTaskSession(ctx, "session1")
if err != nil {
t.Fatalf("failed to load session: %v", err)
}
session.AgentProfileID = "profile1"
session.ErrorMessage = models.SessionArchiveCancelReason
session.TaskEnvironmentID = "env1"
if err := repo.UpdateTaskSession(ctx, session); err != nil {
t.Fatalf("failed to update session: %v", err)
}
if err := repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{
ID: "env1", TaskID: "task1", ExecutorType: "worktree",
WorkspacePath: t.TempDir(), Status: models.TaskEnvironmentStatusReady,
}); err != nil {
t.Fatalf("CreateTaskEnvironment: %v", err)
}
// Recorded path the archive already deleted.
if err := repo.CreateTaskEnvironmentRepo(ctx, &models.TaskEnvironmentRepo{
ID: "ter1",
TaskEnvironmentID: "env1",
WorktreeID: "wid1",
RepositoryID: "repo1",
WorktreePath: filepath.Join(t.TempDir(), "archived-task", "kandev"),
WorktreeBranch: "feature/archived-work",
}); err != nil {
t.Fatalf("CreateTaskEnvironmentRepo: %v", err)
}
now := time.Now().UTC()
if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
ID: "er1",
SessionID: "session1",
TaskID: "task1",
Status: "ready",
Resumable: true,
ResumeToken: "acp-session-archived",
CreatedAt: now,
UpdatedAt: now,
}); err != nil {
t.Fatalf("failed to upsert executor running: %v", err)
}
taskRepo := newMockTaskRepo()
taskRepo.tasks["task1"] = &v1.Task{ID: "task1", State: v1.TaskStateInProgress}
agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr)
svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
resp, err := svc.GetTaskSessionStatus(ctx, "task1", "session1")
if err != nil {
t.Fatalf("GetTaskSessionStatus returned error: %v", err)
}
if resp.Error != "" {
t.Fatalf("expected no status error for a recreatable worktree, got %q", resp.Error)
}
if !resp.IsResumable {
t.Fatal("expected IsResumable=true: the worktree is recreated by the resume itself")
}
if !resp.NeedsResume {
t.Fatal("expected NeedsResume=true so the unarchived session auto-resumes")
}
if resp.ResumeReason != resumeReasonArchiveCancelledResumable {
t.Fatalf("expected ResumeReason=%q, got %q", resumeReasonArchiveCancelledResumable, resp.ResumeReason)
}
}
@@ -1173,10 +1173,16 @@ func (m *Manager) recreate(ctx context.Context, existing *Worktree, req CreateRe
} }
} }
// Update record // Update record. DeletedAt must be cleared alongside the status: archive
// cleanup releases the reference (status=deleted + deleted_at), and
// UpdateWorktree persists deleted_at verbatim — leaving it set would keep
// the freshly recreated worktree invisible to every lookup that filters on
// `deleted_at IS NULL`, so the next launch would mint a brand-new worktree
// beside the one we just restored.
now := time.Now() now := time.Now()
existing.Path = worktreePath existing.Path = worktreePath
existing.Status = StatusActive existing.Status = StatusActive
existing.DeletedAt = nil
existing.UpdatedAt = now existing.UpdatedAt = now
if m.store != nil { if m.store != nil {
@@ -3,9 +3,12 @@ package worktree
import ( import (
"context" "context"
"errors" "errors"
"os"
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"github.com/kandev/kandev/internal/task/models"
) )
// archiveDeletesLocalBranch simulates what task archive does to a worktree's // archiveDeletesLocalBranch simulates what task archive does to a worktree's
@@ -158,3 +161,76 @@ func TestRecreate_ForkPRFetchesPullHeadRef(t *testing.T) {
t.Errorf("worktree HEAD = %q, want %q (PR head)", gotSHA, prHeadSHA) t.Errorf("worktree HEAD = %q, want %q (PR head)", gotSHA, prHeadSHA)
} }
} }
// TestCreate_RestoresReleasedWorktreeAfterArchive is the whole unarchive
// round trip at the worktree layer. Archiving a task removes the worktree
// directory and releases its reference (status=deleted + deleted_at) while
// deliberately keeping the git branch, so the next launch must rebuild the
// directory from the released record — including reactivating that record.
// Leaving deleted_at set would hide the restored worktree from every lookup
// that filters on `deleted_at IS NULL`, so the session would silently get a
// brand-new worktree instead of its own work back.
func TestCreate_RestoresReleasedWorktreeAfterArchive(t *testing.T) {
mgr, store := newReferenceCleanupTestManager(t)
ctx := context.Background()
seedReferenceCleanupSession(t, store, "task-archived", "session-archived", models.TaskSessionStateCompleted)
repoPath := initGitRepoWithRemote(t)
req := CreateRequest{
TaskID: "task-archived",
SessionID: "session-archived",
TaskTitle: "Archived work",
RepositoryID: "repository",
RepositoryPath: repoPath,
BaseBranch: "main",
TaskDirName: "task-archived",
RepoName: "repository",
}
wt, err := mgr.Create(ctx, req)
if err != nil {
t.Fatalf("create worktree: %v", err)
}
// Work the user never pushed. It only survives because archive keeps the
// branch (DestroyWorktree passes removeBranch=false).
runGit(t, wt.Path, "commit", "--allow-empty", "-m", "unpushed work")
workSHA := strings.TrimSpace(runGit(t, wt.Path, "rev-parse", "HEAD"))
// Archive: remove the directory, release the reference, keep the branch.
if err := mgr.RemoveByID(ctx, wt.ID, false); err != nil {
t.Fatalf("archive worktree: %v", err)
}
if _, statErr := os.Stat(wt.Path); !os.IsNotExist(statErr) {
t.Fatalf("archive should remove the worktree directory, stat error = %v", statErr)
}
// Unarchive + resume: the launch carries the stored worktree ID.
resumeReq := req
resumeReq.WorktreeID = wt.ID
restored, err := mgr.Create(ctx, resumeReq)
if err != nil {
t.Fatalf("resume after unarchive must recreate the worktree: %v", err)
}
if restored.Path != wt.Path {
t.Fatalf("restored path = %q, want the original %q", restored.Path, wt.Path)
}
if restored.Branch != wt.Branch {
t.Fatalf("restored branch = %q, want the original %q", restored.Branch, wt.Branch)
}
if got := strings.TrimSpace(runGit(t, restored.Path, "rev-parse", "HEAD")); got != workSHA {
t.Fatalf("restored HEAD = %q, want the pre-archive work %q", got, workSHA)
}
// The record must be visible again to the session-scoped lookup the next
// launch uses; a lingering deleted_at would strand it.
found, err := store.GetWorktreeBySessionAndRepository(ctx, "session-archived", "repository", restored.BranchSlug)
if err != nil {
t.Fatalf("look up restored worktree: %v", err)
}
if found == nil {
t.Fatal("restored worktree is still hidden from the session lookup (deleted_at was not cleared)")
}
if found.ID != wt.ID {
t.Fatalf("session lookup returned worktree %q, want the restored %q", found.ID, wt.ID)
}
assertWorktreeReferenceStatus(t, store, wt.ID, StatusActive)
}