diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 8d1acaf470..f3f6214ee7 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -996,7 +996,9 @@ func (s *Service) setSessionStarting(ctx context.Context, taskID string, session } allowedTerminalRecovery := !promoteTask && session.State == models.TaskSessionStateStarting && - current.State == models.TaskSessionStateFailed + (current.State == models.TaskSessionStateFailed || + (current.State == models.TaskSessionStateCancelled && + models.IsArchiveCancelReason(current.ErrorMessage))) if isTerminalSessionState(current.State) && !allowedTerminalRecovery { return &executor.SessionStateSupersededError{SessionID: session.ID, State: current.State} } diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index 9f75dbf254..91b7217df3 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go @@ -1854,6 +1854,43 @@ func TestSetSessionStartingAllowsTerminalResumeWithoutTaskPromotion(t *testing.T require.Empty(t, taskRepo.stateWrites) } +// TestSetSessionStartingAllowsArchiveCancelledResumeWithoutPromotion covers +// sessions cancelled by an archive (Service.ArchiveTask or the cascade +// archive path), which carry a distinct ErrorMessage; unlike an explicit +// user/coordinator stop, these must recover into STARTING like a Failed +// session so Resume works normally once the task is unarchived (bug: +// "Can't resume this un-archived task"). +func TestSetSessionStartingAllowsArchiveCancelledResumeWithoutPromotion(t *testing.T) { + for _, reason := range []string{models.SessionArchiveCancelReason, models.SessionArchiveTreeCancelReason} { + t.Run(reason, func(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + current, err := repo.GetTaskSession(ctx, "s1") + require.NoError(t, err) + current.State = models.TaskSessionStateCancelled + current.ErrorMessage = reason + require.NoError(t, repo.UpdateTaskSession(ctx, current)) + + next := *current + next.State = models.TaskSessionStateStarting + next.ErrorMessage = "" + next.UpdatedAt = time.Now().UTC() + + taskRepo := newMockTaskRepo() + svc := createTestService(repo, newMockStepGetter(), taskRepo) + + require.NoError(t, svc.setSessionStarting(ctx, "t1", &next, false)) + + updated, err := repo.GetTaskSession(ctx, "s1") + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateStarting, updated.State) + require.Empty(t, taskRepo.stateWrites) + }) + } +} + // Pins the call-site wiring: cancelled office turn must NOT leave the session at IDLE. func TestHandleCompleteStreamEvent_CancelledOfficeSessionLandsWaitingForInput(t *testing.T) { ctx := context.Background() diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index 207ae5ae5e..16197377a2 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -457,7 +457,9 @@ func (e *Executor) updateSessionStarting(ctx context.Context, taskID string, ses } allowedTerminalRecovery := !promoteTask && session.State == models.TaskSessionStateStarting && - current.State == models.TaskSessionStateFailed + (current.State == models.TaskSessionStateFailed || + (current.State == models.TaskSessionStateCancelled && + models.IsArchiveCancelReason(current.ErrorMessage))) if isStopTerminalSessionState(current.State) && !allowedTerminalRecovery { return &SessionStateSupersededError{SessionID: session.ID, State: current.State} } diff --git a/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go b/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go index d2ab062d81..341fa2b6ac 100644 --- a/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go @@ -333,6 +333,67 @@ func TestLaunchPreparedSession_MultiRepo_ReusesPerRepoWorktreeIDsFromEnvironment } } +// TestResumeSession_MultiRepo_PopulatesRequestRepositories is the resume-path +// counterpart of TestLaunchPreparedSession_MultiRepo_PopulatesRequestRepositories. +// buildResumeRequest loads task via the raw repository GetTask, which never +// populates Task.Repositories (that's a separate one-to-many table loaded +// only via ListTaskRepositories — see resolveAllRepoInfo). applyResumeMultiRepoConfig +// must not gate on Task.Repositories: doing so silently drops every repo but +// the primary on ANY resume of a multi-repo task, regardless of why the +// session needed resuming. +func TestResumeSession_MultiRepo_PopulatesRequestRepositories(t *testing.T) { + repo := newMockRepository() + const taskID = "task-multi-resume" + const sessionID = "session-multi-resume" + seedMultiRepoTask(t, repo, taskID) + seedWorktreeExecutor(repo) + + // Mirrors the raw repository GetTask: no Repositories slice attached. + repo.tasks[taskID] = &models.Task{ID: taskID, WorkspaceID: "ws-1", Title: "Multi Resume"} + repo.sessions[sessionID] = &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + AgentProfileID: "profile-123", + ExecutorID: models.ExecutorIDWorktree, + RepositoryID: "repo-front", + State: models.TaskSessionStateCancelled, + ErrorMessage: models.SessionArchiveTreeCancelReason, + StartedAt: time.Now(), + UpdatedAt: time.Now(), + } + + var captured *LaunchAgentRequest + agentManager := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { + captured = req + return &LaunchAgentResponse{ + AgentExecutionID: "exec-resume-multi", + WorktreePath: "/tasks/x", + Worktrees: []RepoWorktreeResult{ + {RepositoryID: "repo-front", WorktreeID: "wt-front", WorktreeBranch: "feat/x-1", WorktreePath: "/tasks/x/frontend"}, + {RepositoryID: "repo-back", WorktreeID: "wt-back", WorktreeBranch: "feat/x-2", WorktreePath: "/tasks/x/backend"}, + }, + }, nil + }, + } + exec := newTestExecutor(t, agentManager, repo) + + if _, err := exec.ResumeSession(context.Background(), repo.sessions[sessionID], false); err != nil { + t.Fatalf("ResumeSession: %v", err) + } + + if captured == nil { + t.Fatal("expected launch agent to be called") + } + if len(captured.Repositories) != 2 { + t.Fatalf("expected req.Repositories length 2 (both repos recreated on resume), got %d: %+v", + len(captured.Repositories), captured.Repositories) + } + if captured.Repositories[0].RepositoryID != "repo-front" || captured.Repositories[1].RepositoryID != "repo-back" { + t.Errorf("unexpected repo order: %+v", captured.Repositories) + } +} + func TestReuseExistingRepositoryWorktrees_EnvironmentRowsWinOverSessionRows(t *testing.T) { repo := newMockRepository() taskID := "task-env-wins" diff --git a/apps/backend/internal/orchestrator/executor/executor_resume.go b/apps/backend/internal/orchestrator/executor/executor_resume.go index 42896b39d2..fec77f3ec2 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume.go @@ -832,10 +832,13 @@ func (e *Executor) applyResumeCloneURL(req *LaunchAgentRequest, repository *mode // so the lifecycle preparer can resume/recreate each repo's worktree. The // legacy top-level fields stay populated from the primary for backwards // compat. +// +// Gates on resolveAllRepoInfo's DB-backed count, not task.Repositories: task +// here is loaded via the raw repository GetTask (validateAndLockResume), +// which never attaches the one-to-many task_repositories rows — that field +// is always empty on this path. Gating on it silently dropped every repo but +// the primary on any resume of a multi-repo task. 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 diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index 32a6a18454..ae8e5dd220 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -2554,6 +2554,66 @@ func TestPersistResumeState_SetsStartingState(t *testing.T) { }) } +// TestUpdateSessionStarting_ArchiveCancelledSessionRecovers proves +// updateSessionStarting's fallback path (no onSessionStarting callback +// wired) mirrors Service.setSessionStarting: an archive-cancelled session +// (the resume path treats CANCELLED as an expected, resumable terminal +// state) may recover into STARTING like a Failed session. +func TestUpdateSessionStarting_ArchiveCancelledSessionRecovers(t *testing.T) { + for _, reason := range []string{models.SessionArchiveCancelReason, models.SessionArchiveTreeCancelReason} { + t.Run(reason, func(t *testing.T) { + repo := newMockRepository() + executor := newTestExecutor(t, &mockAgentManager{}, repo) + repo.sessions["session-1"] = &models.TaskSession{ + ID: "session-1", + TaskID: "task-1", + State: models.TaskSessionStateCancelled, + ErrorMessage: reason, + } + next := &models.TaskSession{ + ID: "session-1", + TaskID: "task-1", + State: models.TaskSessionStateStarting, + } + + if err := executor.updateSessionStarting(context.Background(), "task-1", next, false); err != nil { + t.Fatalf("updateSessionStarting: %v", err) + } + if got := repo.sessions["session-1"].State; got != models.TaskSessionStateStarting { + t.Fatalf("session state = %q, want STARTING", got) + } + }) + } +} + +// TestUpdateSessionStarting_OrdinaryCancelledSessionStaysRejected is the +// counterpart to TestUpdateSessionStarting_ArchiveCancelledSessionRecovers: +// an ordinary/explicit cancellation (no archive-cancel reason) must stay +// rejected as superseded, never silently promoted back into STARTING. +func TestUpdateSessionStarting_OrdinaryCancelledSessionStaysRejected(t *testing.T) { + repo := newMockRepository() + executor := newTestExecutor(t, &mockAgentManager{}, repo) + repo.sessions["session-1"] = &models.TaskSession{ + ID: "session-1", + TaskID: "task-1", + State: models.TaskSessionStateCancelled, + } + next := &models.TaskSession{ + ID: "session-1", + TaskID: "task-1", + State: models.TaskSessionStateStarting, + } + + err := executor.updateSessionStarting(context.Background(), "task-1", next, false) + var superseded *SessionStateSupersededError + if !errors.As(err, &superseded) { + t.Fatalf("updateSessionStarting error = %v, want SessionStateSupersededError", err) + } + if got := repo.sessions["session-1"].State; got != models.TaskSessionStateCancelled { + t.Fatalf("session state = %q, want CANCELLED (unchanged)", got) + } +} + func TestPersistLaunchState_PrepareOnlyCannotOverwriteCoordinatorCancellation(t *testing.T) { repo := newMockRepository() executor := newTestExecutor(t, &mockAgentManager{}, repo) diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index a1da9d7925..1473845cf0 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -3930,6 +3930,67 @@ func TestResumeTaskSession_FailedKeepsResumeToken(t *testing.T) { } } +// TestResumeTaskSession_ArchiveCancelledSessionResumesSuccessfully is the +// end-to-end regression for "Can't resume this un-archived task": a session +// cancelled by an archive (Service.ArchiveTask / cascade archive) must resume +// like a Failed one once the task is unarchived. Before the fix, the launch +// succeeded but the STARTING-transition guard rejected the CANCELLED session +// as superseded, leaving the freshly-launched execution orphaned while the +// session stayed stuck at CANCELLED and the task was marked FAILED. +func TestResumeTaskSession_ArchiveCancelledSessionResumesSuccessfully(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + taskRepo := newMockTaskRepo() + + startAgentProcessCalled := false + agentMgr := &sessionUpdatingAgentManager{ + mockAgentManager: &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return &executor.LaunchAgentResponse{ + AgentExecutionID: "exec-new", + Status: v1.AgentStatusStarting, + }, nil + }, + }, + repo: repo, + sessionID: "session1", + taskID: "task1", + onStartCalled: &startAgentProcessCalled, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + + // Session was cancelled by an archive; the task has since been unarchived + // (seedTaskAndSession's task is never archived), leaving exactly the state + // an unarchive-then-resume observes. No ExecutorRunning row exists — the + // archive cleanup already tore it down, which ResumeTaskSession's initial + // guard explicitly allows for terminal-state sessions. + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCancelled) + session, _ := repo.GetTaskSession(ctx, "session1") + session.AgentProfileID = "profile-1" + session.ErrorMessage = models.SessionArchiveTreeCancelReason + _ = repo.UpdateTaskSession(ctx, session) + + if _, err := repo.GetExecutorRunningBySessionID(ctx, "session1"); !errors.Is(err, models.ErrExecutorRunningNotFound) { + t.Fatalf("precondition: expected no executors_running row for session1 (archive cleanup should have removed it), got err=%v", err) + } + + if _, err := svc.ResumeTaskSession(ctx, "task1", "session1"); err != nil { + t.Fatalf("ResumeTaskSession on archive-cancelled session returned: %v", err) + } + + resumed, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if resumed.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("session state = %s, want WAITING_FOR_INPUT (agent started successfully)", resumed.State) + } + if got, ok := taskRepo.updatedStates["task1"]; ok && got == v1.TaskStateFailed { + t.Error("task must not be marked FAILED when resume of an archive-cancelled session succeeds") + } +} + // ctxAwareTaskRepo wraps mockTaskRepo and respects ctx cancellation. Used to // prove that ResumeTaskSession's failure-recording path is insulated from a // pre-cancelled caller ctx (the WS-disconnect scenario). diff --git a/apps/backend/internal/task/models/resume_safety.go b/apps/backend/internal/task/models/resume_safety.go index 734fe8c34f..ce3a8e67a3 100644 --- a/apps/backend/internal/task/models/resume_safety.go +++ b/apps/backend/internal/task/models/resume_safety.go @@ -69,3 +69,25 @@ func RowMustBePreserved(running *ExecutorRunning, sessionState TaskSessionState) } return IsResumableSessionState(sessionState) } + +// SessionArchiveCancelReason and SessionArchiveTreeCancelReason are the +// TaskSession.ErrorMessage values written when a session is auto-cancelled +// by archiving — Service.ArchiveTask's CancelActiveTaskSessionsByTaskID call +// (single-task archive) and HandoffService.cancelActiveRuns (cascade archive) +// respectively. They exist so the resume path can tell "cancelled because the +// task was archived" apart from an explicit user/coordinator stop. +const ( + SessionArchiveCancelReason = "task archived" + SessionArchiveTreeCancelReason = "task tree archived" +) + +// IsArchiveCancelReason reports whether reason is a session-cancellation +// message written by an archive path rather than an explicit user or +// coordinator stop. A session cancelled by archiving is expected to resume +// normally once the task is unarchived, so callers gating a STARTING +// transition on "was this session explicitly, intentionally cancelled" must +// treat an archive-cancelled session like a Failed one instead of rejecting +// it outright — see setSessionStarting / updateSessionStarting. +func IsArchiveCancelReason(reason string) bool { + return reason == SessionArchiveCancelReason || reason == SessionArchiveTreeCancelReason +} diff --git a/apps/backend/internal/task/service/handoff_cascade.go b/apps/backend/internal/task/service/handoff_cascade.go index dcf8a85a1d..9b92198624 100644 --- a/apps/backend/internal/task/service/handoff_cascade.go +++ b/apps/backend/internal/task/service/handoff_cascade.go @@ -211,7 +211,7 @@ func (s *HandoffService) ArchiveTaskTree(ctx context.Context, rootID string, cas // Cancel active runs first. Failures are logged and skipped — a // stuck cancel must not block the archive cascade; the orchestrator // will reconcile any orphan execution on its next tick. - s.cancelActiveRuns(ctx, all, "task tree archived") + s.cancelActiveRuns(ctx, all, models.SessionArchiveTreeCancelReason) // Archive deepest first so parent_id pointers stay valid through // the walk; not strictly required by the schema (no FK on parent_id) diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index 311fa9d363..65ad409f7e 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -1254,7 +1254,7 @@ func (s *Service) finalizeCancelledSessions(ctx context.Context, taskID string, var cancelledSessions []*models.TaskSession var cancelErr error for attempt := 1; attempt <= maxCancelAttempts; attempt++ { - cancelledSessions, cancelErr = s.sessions.CancelActiveTaskSessionsByTaskID(ctx, taskID, "task archived") + cancelledSessions, cancelErr = s.sessions.CancelActiveTaskSessionsByTaskID(ctx, taskID, models.SessionArchiveCancelReason) if cancelErr == nil { break } @@ -1286,7 +1286,7 @@ func (s *Service) finalizeCancelledSessions(ctx context.Context, taskID string, // can no longer consume a shared batch-wide budget and starve the // events for sessions later in the loop. detachedCtx := context.WithoutCancel(ctx) - s.publishSessionsCancelled(detachedCtx, taskID, activeSessions, cancelledSessions, "task archived") + s.publishSessionsCancelled(detachedCtx, taskID, activeSessions, cancelledSessions, models.SessionArchiveCancelReason) } func (s *Service) registerTaskRuntimeStopOwners(stopTargets []taskStopTarget, force bool) {