fix: resume an unarchived task's archive-cancelled and multi-repo sessions (#1905)
This commit is contained in:
@@ -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}
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user