diff --git a/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go b/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go index 65501c4925..cbcbd54013 100644 --- a/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go @@ -528,6 +528,18 @@ func wireBootReadySimulator(svc *Service, agentMgr *mockAgentManager, newExecID } } agentMgr.launchAgentFunc = func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + // Record the initial ACP prompt baked into the launch request (this is + // how StartCreatedSession/LaunchPreparedSession delivers a merged + // hand-off prompt for a fresh session — unlike PromptTask's lazy-resume + // path for an already-launched session, there is no separate + // PromptAgent call to capture) so assertHandoffDeliveredOrQueued can + // find it via the same capturedPromptsForExecution helper. + agentMgr.mu.Lock() + agentMgr.capturedPromptCalls = append(agentMgr.capturedPromptCalls, promptCall{ + ExecutionID: newExecID, + Prompt: req.TaskDescription, + }) + agentMgr.mu.Unlock() // Simulate the lifecycle manager's persistExecutorRunning: in production // the row is upserted in lockstep with executionStore.Add; here we mirror // that timing so the orchestrator's GetExecutionIDForSession lookup @@ -619,29 +631,56 @@ func (sc *pendingMoveScenario) assertOneTransitionToInProgress(t *testing.T, ste t.Error("review session must no longer be primary (the impl session takes over)") } - impl, err := sc.repo.GetTaskSession(sc.ctx, sc.implSessionID) + // The original impl session was seeded COMPLETED (it was previously + // launched and completed a real turn — mirroring production). Terminal + // sessions are never revived for workflow re-entry (see + // findReusableSessionForProfile), so it must stay exactly as seeded: + // terminal, non-primary, historically intact. + oldImpl, err := sc.repo.GetTaskSession(sc.ctx, sc.implSessionID) if err != nil { - t.Fatalf("load impl session: %v", err) + t.Fatalf("load original impl session: %v", err) + } + if oldImpl.State != models.TaskSessionStateCompleted { + t.Errorf("original impl session state = %q, want it to remain COMPLETED (never revived)", oldImpl.State) + } + if oldImpl.IsPrimary { + t.Error("original impl session must remain non-primary (never revived)") + } + + // Re-entry into the Impl profile must create a FRESH session rather than + // resurrecting session-impl's stale ACP conversation. + sessions, err := sc.repo.ListTaskSessions(sc.ctx, "task-1") + if err != nil { + t.Fatalf("list sessions: %v", err) + } + var freshImpl *models.TaskSession + for _, s := range sessions { + if s.AgentProfileID == profileImpl && s.ID != sc.implSessionID { + freshImpl = s + } + } + if freshImpl == nil { + t.Fatal("expected a fresh impl-profile session distinct from the original COMPLETED session-impl") } - if !impl.IsPrimary { - t.Error("impl session must be primary after the deferred move applies") + if !freshImpl.IsPrimary { + t.Error("fresh impl session must be primary after the deferred move applies") } - if impl.State == models.TaskSessionStateCompleted { - t.Errorf("impl session state = %q, expected non-terminal (revived for a new turn)", impl.State) + if isTerminalSessionState(freshImpl.State) { + t.Errorf("fresh impl session state = %q, expected non-terminal", freshImpl.State) } - sc.assertHandoffDeliveredOrQueued(t) + sc.assertHandoffDeliveredOrQueued(t, freshImpl.ID) } -// assertHandoffDeliveredOrQueued checks the hand-off prompt landed on the impl -// session — either delivered to its agent (PromptAgent capture) or sitting in -// the queue waiting for delivery. Both are acceptable; the failure mode the -// regression catches is "lost" (neither delivered nor queued) or "delivered -// to the wrong session". -func (sc *pendingMoveScenario) assertHandoffDeliveredOrQueued(t *testing.T) { +// assertHandoffDeliveredOrQueued checks the hand-off prompt landed on the +// given session — either delivered to its agent (PromptAgent capture) or +// sitting in the queue waiting for delivery. Both are acceptable; the +// failure mode the regression catches is "lost" (neither delivered nor +// queued) or "delivered to the wrong session". +func (sc *pendingMoveScenario) assertHandoffDeliveredOrQueued(t *testing.T, targetSessionID string) { t.Helper() implPrompts := capturedPromptsForExecution(sc.agentMgr, sc.implRelaunchExec) - implQueued := sc.svc.messageQueue.GetStatus(sc.ctx, sc.implSessionID) + implQueued := sc.svc.messageQueue.GetStatus(sc.ctx, targetSessionID) if len(implPrompts) == 0 && implQueued.Count == 0 { t.Error("hand-off prompt was neither delivered to the impl session nor queued for it") diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index ac12a1f1cd..b2f7d07fcd 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -29,7 +29,10 @@ import ( type turnCompletionCause string -var errDeferredMoveAlreadyApplied = errors.New("deferred move already applied") +var ( + errDeferredMoveAlreadyApplied = errors.New("deferred move already applied") + errReusableSessionNoLongerActive = errors.New("reusable session is no longer active") +) type taskMetadataKeyRemover interface { RemoveTaskMetadataKey(context.Context, string, string) (bool, error) @@ -1479,10 +1482,14 @@ func (s *Service) tagSessionAsWorkflowSwitched(ctx context.Context, sessionID st } // switchSessionForStep activates a session for the new agent profile. -// If an existing session on this task already uses the target profile it is -// reused (re-promoted to primary, brought out of COMPLETED if it had been -// switched away from previously). Otherwise a new session is prepared. -// In both cases the previous session is stopped and marked COMPLETED. +// If a nonterminal session on this task already uses the target profile, it +// is reused (re-promoted to primary). Otherwise a new session is prepared — +// including when the only matching session is terminal (COMPLETED, FAILED, +// or CANCELLED): workflow re-entry never resumes a terminal session's ACP +// conversation, because prior-completion state in that conversation can +// mislead the agent into replaying stale routing intent (see +// findReusableSessionForProfile). In both cases the previous session is +// stopped and marked COMPLETED. func (s *Service) switchSessionForStep(ctx context.Context, taskID string, currentSession *models.TaskSession, newAgentProfileID string) (*models.TaskSession, error) { s.logger.Info("switching session for workflow step agent profile change", zap.String("task_id", taskID), @@ -1504,16 +1511,35 @@ func (s *Service) switchSessionForStep(ctx context.Context, taskID string, curre zap.Error(lookupErr)) } if existing != nil { - return s.reuseSessionForStep(ctx, taskID, currentSession, existing) + reused, err := s.reuseSessionForStep(ctx, taskID, currentSession, existing) + if err == nil { + return reused, nil + } + if !errors.Is(err, errReusableSessionNoLongerActive) { + return nil, err + } + s.logger.Info("reusable session became terminal before workflow promotion; creating fresh session", + zap.String("task_id", taskID), + zap.String("session_id", existing.ID), + zap.String("agent_profile_id", newAgentProfileID)) } return s.createNewSessionForStep(ctx, taskID, currentSession, newAgentProfileID) } -// findReusableSessionForProfile returns the most-recently-updated session on -// this task that uses the target profile (and is not the session being -// switched away from), or nil if none exists. Failed/cancelled sessions are -// excluded — those are dead and shouldn't be revived implicitly. +// findReusableSessionForProfile returns the most-recently-updated +// *nonterminal* session on this task that uses the target profile (and is +// not the session being switched away from), or nil if none exists. +// +// Terminal sessions (COMPLETED, FAILED, CANCELLED) are always excluded — +// they are historical endpoints, not workflow-reusable. A prior incident +// showed why: reviving a COMPLETED session lazily resumed its persisted ACP +// conversation, which still contained the agent's earlier completion state. +// Seeing the task routed back to that step, the agent reasonably inferred +// its prior completion had been cancelled and moved the task backward, +// re-arming the same cycle on the next re-entry. Terminal-profile re-entry +// always goes through createNewSessionForStep instead, which gets a fresh +// ACP conversation and the canonical current task/workflow context. func (s *Service) findReusableSessionForProfile(ctx context.Context, taskID, profileID, excludeSessionID string) (*models.TaskSession, error) { if profileID == "" { return nil, nil @@ -1530,11 +1556,7 @@ func (s *Service) findReusableSessionForProfile(ctx context.Context, taskID, pro if sess.AgentProfileID != profileID { continue } - // Skip user-cancelled sessions — those are explicit stops and - // shouldn't be auto-revived. FAILED sessions are reused (the failure - // may have been transient; either way the user expects "one session - // per profile per task" so we revive rather than orphan a duplicate). - if sess.State == models.TaskSessionStateCancelled { + if isTerminalSessionState(sess.State) { continue } if best == nil || sess.UpdatedAt.After(best.UpdatedAt) { @@ -1544,19 +1566,15 @@ func (s *Service) findReusableSessionForProfile(ctx context.Context, taskID, pro return best, nil } -// reuseSessionForStep promotes an existing session to primary, brings it out -// of COMPLETED/FAILED if needed, and stops + completes the previous session. -// The agent for the reused session is not relaunched here — when a prompt -// arrives, the autoStart/PromptTask paths handle the launch. -// -// Previously-launched sessions (executors_running record exists, has resume -// token) are flipped to WAITING_FOR_INPUT so PromptTask's ensureSessionRunning -// lazy-resumes them via ResumeSession. -// -// Never-launched sessions (e.g. PrepareSession created the row but the -// workflow switched away before the agent started) have no executors_running -// record. They go to CREATED so autoStartStepPrompt routes through -// StartCreatedSession → LaunchPreparedSession (a full fresh launch). +// reuseSessionForStep promotes an existing nonterminal session to primary +// and stops + completes the previous session. The agent for the reused +// session is not relaunched here — when a prompt arrives, the +// autoStart/PromptTask paths handle the launch (including lazy-resume via +// ResumeSession for a session that was previously launched and is currently +// WAITING_FOR_INPUT). Before it promotes the candidate, it atomically checks +// that the persisted row is still nonterminal. This closes the lookup-to- +// promotion race where an agent completion could otherwise make a stale ACP +// conversation primary again. func (s *Service) reuseSessionForStep(ctx context.Context, taskID string, currentSession, existing *models.TaskSession) (*models.TaskSession, error) { s.logger.Info("reusing existing session for profile", zap.String("task_id", taskID), @@ -1565,16 +1583,14 @@ func (s *Service) reuseSessionForStep(ctx context.Context, taskID string, curren zap.String("reused_profile", existing.AgentProfileID), zap.String("reused_state", string(existing.State))) - if existing.State == models.TaskSessionStateCompleted || existing.State == models.TaskSessionStateFailed { - s.reviveReusedSession(ctx, existing) - } - - s.tagSessionAsWorkflowSwitched(ctx, existing.ID) - - if err := s.SetPrimarySession(ctx, existing.ID); err != nil { + promoted, err := s.setNonterminalSessionPrimary(ctx, existing.ID) + if err != nil { s.logger.Warn("failed to set reused session as primary", zap.String("session_id", existing.ID), zap.Error(err)) + } else if !promoted { + return nil, errReusableSessionNoLongerActive } + s.tagSessionAsWorkflowSwitched(ctx, existing.ID) // Transfer any queued message and pending move from the session being // switched away from to the reused session — without this, a hand-off @@ -1595,33 +1611,10 @@ func (s *Service) reuseSessionForStep(ctx context.Context, taskID string, curren return existing, nil } -// reviveReusedSession flips a terminal (COMPLETED/FAILED) session back to a -// state where the downstream autoStart/PromptTask paths can launch its agent. -// The target state depends on whether the session was ever launched: -// - Has executors_running record → WAITING_FOR_INPUT, lazy-resume from token -// - No record → CREATED, fresh launch via StartCreatedSession -// -// The previous error message (from a prior FAILED state) is cleared so the -// frontend stops surfacing stale red banners on a now-active session. -func (s *Service) reviveReusedSession(ctx context.Context, session *models.TaskSession) { - wasLaunched := false - if running, err := s.repo.GetExecutorRunningBySessionID(ctx, session.ID); err == nil && running != nil { - wasLaunched = true - } - if wasLaunched { - session.State = models.TaskSessionStateWaitingForInput - } else { - session.State = models.TaskSessionStateCreated - } - session.CompletedAt = nil - session.ErrorMessage = "" - session.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateTaskSession(ctx, session); err != nil { - s.logger.Warn("failed to revive reused session out of COMPLETED", - zap.String("session_id", session.ID), - zap.String("target_state", string(session.State)), - zap.Error(err)) - } +// setNonterminalSessionPrimary promotes a workflow-reused session only when +// its persisted state is still nonterminal. +func (s *Service) setNonterminalSessionPrimary(ctx context.Context, sessionID string) (bool, error) { + return s.repo.SetSessionPrimaryIfNonterminal(ctx, sessionID) } // createNewSessionForStep is the original switch-and-create-fresh-session path, diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go index 4cb75d3415..534d7b15d6 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_profile_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "strings" + "sync" "testing" "time" @@ -20,6 +21,34 @@ import ( wfmodels "github.com/kandev/kandev/internal/workflow/models" ) +// terminalizeCandidateBeforePromotionRepo pauses a profile-switch promotion +// after lookup so the test can terminalize the selected row before the +// promotion write begins. +type terminalizeCandidateBeforePromotionRepo struct { + sessionExecutorStore + promotionReached chan struct{} + allowPromotion chan struct{} + once sync.Once +} + +func (r *terminalizeCandidateBeforePromotionRepo) waitForPromotion() { + r.once.Do(func() { close(r.promotionReached) }) + <-r.allowPromotion +} + +func (r *terminalizeCandidateBeforePromotionRepo) SetSessionPrimary(ctx context.Context, sessionID string) error { + r.waitForPromotion() + return r.sessionExecutorStore.SetSessionPrimary(ctx, sessionID) +} + +func (r *terminalizeCandidateBeforePromotionRepo) SetSessionPrimaryIfNonterminal( + ctx context.Context, + sessionID string, +) (bool, error) { + r.waitForPromotion() + return r.sessionExecutorStore.SetSessionPrimaryIfNonterminal(ctx, sessionID) +} + func seedAutopilotTaskAndSession(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID string, sessionState models.TaskSessionState) { t.Helper() ctx := context.Background() @@ -587,11 +616,14 @@ func TestSwitchSessionForStep(t *testing.T) { }) } -// TestSwitchSessionForStep_ReusesExistingProfileSession verifies the core -// requirement: when switching to a profile that already has a session on this -// task, switchSessionForStep reuses it instead of creating a third session. -// Covers the A→B→A round trip (and beyond) at the unit-test level. -func TestSwitchSessionForStep_ReusesExistingProfileSession(t *testing.T) { +// TestSwitchSessionForStep_ReusesNonterminalSession verifies the core +// requirement: when switching to a profile that already has a *nonterminal* +// session on this task, switchSessionForStep reuses it instead of creating a +// third session. Covers the A→B→A round trip (and beyond) at the unit-test +// level for the case where profile-A's prior session is still legitimately +// active (WAITING_FOR_INPUT), e.g. it was previously launched and has a +// resume token it should keep using. +func TestSwitchSessionForStep_ReusesNonterminalSession(t *testing.T) { ctx := context.Background() now := time.Now().UTC() @@ -608,23 +640,28 @@ func TestSwitchSessionForStep_ReusesExistingProfileSession(t *testing.T) { } _ = repo.CreateTask(ctx, task) - // Prior session for profile-A — was active before, then completed when - // the workflow switched away from this profile last time. - completedAt := now.Add(-2 * time.Minute) + // Prior session for profile-A — still nonterminal (waiting for the next + // prompt) from the last time this profile was active on this task. prior := &models.TaskSession{ ID: "session-a", TaskID: "t1", AgentProfileID: "profile-a", ExecutorID: "exec-local", ExecutorProfileID: "ep1", - State: models.TaskSessionStateCompleted, + AgentExecutionID: "ae-a", + State: models.TaskSessionStateWaitingForInput, IsPrimary: false, Metadata: map[string]interface{}{"existing": "preserved"}, - CompletedAt: &completedAt, StartedAt: now.Add(-3 * time.Minute), - UpdatedAt: completedAt, + UpdatedAt: now.Add(-2 * time.Minute), } _ = repo.CreateTaskSession(ctx, prior) + _ = repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "er-a", SessionID: "session-a", TaskID: "t1", + ResumeToken: "acp-session-a", + Resumable: true, + CreatedAt: now.Add(-2 * time.Minute), UpdatedAt: now.Add(-2 * time.Minute), + }) // Currently-active session for profile-B — about to be switched away from. current := &models.TaskSession{ @@ -681,18 +718,12 @@ func TestSwitchSessionForStep_ReusesExistingProfileSession(t *testing.T) { t.Errorf("expected 2 sessions after reuse, got %d", len(sessions)) } - // The reused session must be back to a non-terminal state (so it can - // receive the next prompt) and be primary. Specifically, since the prior - // session has no executors_running record (never launched), it should - // flip to CREATED so autoStartStepPrompt routes through StartCreatedSession - // for a fresh launch (instead of hitting "no executor record" in - // ensureSessionRunning). + // The reused session must keep its nonterminal state (still WAITING, so + // PromptTask's ensureSessionRunning lazy-resumes it via ResumeSession) and + // become primary. reused, _ := repo.GetTaskSession(ctx, "session-a") - if reused.State != models.TaskSessionStateCreated { - t.Errorf("never-launched reused session must be CREATED (so StartCreatedSession launches it fresh), got %s", reused.State) - } - if reused.CompletedAt != nil { - t.Error("reused session must have CompletedAt cleared") + if reused.State != models.TaskSessionStateWaitingForInput { + t.Errorf("nonterminal reused session must stay WAITING_FOR_INPUT, got %s", reused.State) } if !reused.IsPrimary { t.Error("reused session must be primary") @@ -714,12 +745,94 @@ func TestSwitchSessionForStep_ReusesExistingProfileSession(t *testing.T) { } } -// TestSwitchSessionForStep_ReusesPreviouslyLaunchedSession covers the other -// branch of the revive: when the reused session has an executors_running -// record (it was previously launched and has a resume token), it flips to -// WAITING_FOR_INPUT so PromptTask's ensureSessionRunning lazy-resumes the -// agent via ResumeSession (preserving its prior conversation context). -func TestSwitchSessionForStep_ReusesPreviouslyLaunchedSession(t *testing.T) { +func TestSwitchSessionForStep_CreatesFreshSessionWhenCandidateTerminalizesBeforePromotion(t *testing.T) { + ctx := context.Background() + now := time.Now().UTC() + repo := setupTestRepo(t) + + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t1", WorkflowID: "wf1", WorkflowStepID: "step1", Title: "Test", Description: "Test", + State: v1.TaskStateInProgress, CreatedAt: now, UpdatedAt: now, + })) + + candidate := &models.TaskSession{ + ID: "session-a", TaskID: "t1", AgentProfileID: "profile-a", ExecutorID: "exec-local", + ExecutorProfileID: "ep1", State: models.TaskSessionStateWaitingForInput, + StartedAt: now.Add(-time.Minute), UpdatedAt: now.Add(-time.Minute), + } + current := &models.TaskSession{ + ID: "session-b", TaskID: "t1", AgentProfileID: "profile-b", ExecutorID: "exec-local", + ExecutorProfileID: "ep1", State: models.TaskSessionStateRunning, IsPrimary: true, + StartedAt: now, UpdatedAt: now, + } + requireNoError(t, repo.CreateTaskSession(ctx, candidate)) + requireNoError(t, repo.CreateTaskSession(ctx, current)) + + taskRepo := newMockTaskRepo() + taskRepo.tasks["t1"] = &v1.Task{ID: "t1", WorkspaceID: "ws1", WorkflowID: "wf1", Title: "Test", Description: "Test", State: v1.TaskStateInProgress} + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + log := testLogger() + exec := executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}) + svc := &Service{ + logger: log, workflowStepGetter: newMockStepGetter(), taskRepo: taskRepo, agentManager: agentMgr, + messageQueue: messagequeue.NewServiceMemory(log), executor: exec, + scheduler: scheduler.NewScheduler(queue.NewTaskQueue(100), exec, taskRepo, log, scheduler.SchedulerConfig{}), + } + barrierRepo := &terminalizeCandidateBeforePromotionRepo{ + sessionExecutorStore: repo, + promotionReached: make(chan struct{}), + allowPromotion: make(chan struct{}), + } + svc.repo = barrierRepo + + resultCh := make(chan *models.TaskSession, 1) + errCh := make(chan error, 1) + go func() { + result, err := svc.switchSessionForStep(ctx, "t1", current, "profile-a") + resultCh <- result + errCh <- err + }() + + <-barrierRepo.promotionReached + requireNoError(t, repo.UpdateTaskSessionState(ctx, candidate.ID, models.TaskSessionStateCompleted, "finished concurrently")) + close(barrierRepo.allowPromotion) + + if err := <-errCh; err != nil { + t.Fatalf("switch session: %v", err) + } + result := <-resultCh + if result == nil || result.ID == candidate.ID { + t.Fatalf("expected a fresh session after candidate terminalized, got %+v", result) + } + + storedCandidate, err := repo.GetTaskSession(ctx, candidate.ID) + requireNoError(t, err) + if storedCandidate.State != models.TaskSessionStateCompleted { + t.Errorf("candidate state = %s, want COMPLETED", storedCandidate.State) + } + if storedCandidate.IsPrimary { + t.Error("terminalized candidate must not become primary") + } + fresh, err := repo.GetTaskSession(ctx, result.ID) + requireNoError(t, err) + if fresh.State != models.TaskSessionStateCreated || !fresh.IsPrimary { + t.Errorf("fresh session = state %s, primary %t; want CREATED primary", fresh.State, fresh.IsPrimary) + } +} + +// TestSwitchSessionForStep_CompletedSessionNotReused locks in the corrective +// fix: a COMPLETED session for the target profile must NOT be revived and +// resumed. Reviving it would lazily resume its persisted ACP conversation, +// which still contains the agent's earlier completion state — the live +// incident this test guards against had the agent see the task routed back +// to a step it had already completed and, reading its own prior "done" +// context, infer the completion had been undone and move the task backward, +// re-arming the same cycle on every re-entry. A fresh session must be +// created instead, and the old COMPLETED session must be left immutable and +// non-primary. +func TestSwitchSessionForStep_CompletedSessionNotReused(t *testing.T) { ctx := context.Background() now := time.Now().UTC() @@ -736,10 +849,9 @@ func TestSwitchSessionForStep_ReusesPreviouslyLaunchedSession(t *testing.T) { } _ = repo.CreateTask(ctx, task) - // Prior session for profile-A — was previously active and has the - // signals of a real launch: an executors_running record with a resume - // token. This should route through the WAITING_FOR_INPUT branch of - // reviveReusedSession. + // Prior QA session for profile-A: COMPLETED, and — like the live + // incident — it was previously launched and still carries a persisted + // ACP resume token via its executors_running record. completedAt := now.Add(-2 * time.Minute) prior := &models.TaskSession{ ID: "session-a", @@ -749,6 +861,8 @@ func TestSwitchSessionForStep_ReusesPreviouslyLaunchedSession(t *testing.T) { ExecutorProfileID: "ep1", AgentExecutionID: "ae-a-1", State: models.TaskSessionStateCompleted, + IsPrimary: false, + Metadata: map[string]interface{}{"existing": "preserved"}, CompletedAt: &completedAt, StartedAt: now.Add(-3 * time.Minute), UpdatedAt: completedAt, @@ -761,6 +875,7 @@ func TestSwitchSessionForStep_ReusesPreviouslyLaunchedSession(t *testing.T) { CreatedAt: completedAt, UpdatedAt: completedAt, }) + // Currently-active session for profile-B — about to be switched away from. current := &models.TaskSession{ ID: "session-b", TaskID: "t1", @@ -796,28 +911,94 @@ func TestSwitchSessionForStep_ReusesPreviouslyLaunchedSession(t *testing.T) { scheduler: sched, } - revived, err := svc.switchSessionForStep(ctx, "t1", current, "profile-a") + fresh, err := svc.switchSessionForStep(ctx, "t1", current, "profile-a") if err != nil { t.Fatalf("unexpected error: %v", err) } - if revived == nil || revived.ID != "session-a" { - t.Fatalf("expected reused session-a, got %+v", revived) + + // Critical: a brand-new session must be created — the COMPLETED session + // is never selected for reuse. + if fresh == nil || fresh.ID == "session-a" { + t.Fatalf("expected a fresh session distinct from the COMPLETED session-a, got %+v", fresh) + } + if fresh.AgentProfileID != "profile-a" { + t.Errorf("fresh session profile = %q, want profile-a", fresh.AgentProfileID) + } + // AC2: the fresh session starts through the CREATED/StartCreatedSession + // path, so it gets a new ACP conversation on first prompt. + if fresh.State != models.TaskSessionStateCreated { + t.Errorf("fresh session state = %s, want CREATED", fresh.State) + } + freshFromDB, err := repo.GetTaskSession(ctx, fresh.ID) + if err != nil { + t.Fatalf("failed to get fresh session: %v", err) + } + if !freshFromDB.IsPrimary { + t.Error("fresh session must be primary") } - reused, _ := repo.GetTaskSession(ctx, "session-a") - if reused.State != models.TaskSessionStateWaitingForInput { - t.Errorf("previously-launched reused session must be WAITING_FOR_INPUT (so PromptTask lazy-resumes via ResumeSession), got %s", reused.State) + // AC3: the fresh session must not inherit the old session's resume token. + freshRunning, err := repo.GetExecutorRunningBySessionID(ctx, fresh.ID) + if err == nil || freshRunning != nil { + t.Errorf("fresh session must have no executors_running record (no inherited resume token), got running=%+v err=%v", freshRunning, err) } - if got := reused.Metadata[models.SessionMetaKeyCreatedBy]; got != models.SessionCreatedByWorkflowSwitch { - t.Errorf("reused session created_by metadata = %v, want %q", got, models.SessionCreatedByWorkflowSwitch) + + // Total session count is now 3: the old COMPLETED session-a, the parked + // session-b, and the fresh session. + sessions, err := repo.ListTaskSessions(ctx, "t1") + if err != nil { + t.Fatalf("failed to list sessions: %v", err) + } + if len(sessions) != 3 { + t.Errorf("expected 3 sessions (old completed + parked + fresh), got %d", len(sessions)) + } + + // AC3: the old COMPLETED session must remain terminal, non-primary, and + // historically intact — untouched by the switch. + oldSession, err := repo.GetTaskSession(ctx, "session-a") + if err != nil { + t.Fatalf("failed to get old session: %v", err) + } + if oldSession.State != models.TaskSessionStateCompleted { + t.Errorf("old session state = %s, want it to remain COMPLETED", oldSession.State) + } + if oldSession.IsPrimary { + t.Error("old COMPLETED session must remain non-primary") + } + if oldSession.CompletedAt == nil || !oldSession.CompletedAt.Equal(completedAt) { + t.Errorf("old session CompletedAt = %v, want unchanged %v", oldSession.CompletedAt, completedAt) + } + if got := oldSession.Metadata["existing"]; got != "preserved" { + t.Errorf("old session metadata must be untouched, got %v", got) + } + + // The old session's resume token itself must remain on file (history is + // preserved) even though it is never handed to the fresh execution. + oldRunning, err := repo.GetExecutorRunningBySessionID(ctx, "session-a") + if err != nil { + t.Fatalf("failed to look up executor running for old session: %v", err) + } + if oldRunning == nil || oldRunning.ResumeToken != "acp-session-a" { + t.Errorf("old session's own resume token must remain intact, got %+v", oldRunning) + } + + // The previous current session-b must now be COMPLETED, not primary. + parked, _ := repo.GetTaskSession(ctx, "session-b") + if parked.State != models.TaskSessionStateCompleted { + t.Errorf("previous current session must be COMPLETED, got %s", parked.State) + } + if parked.IsPrimary { + t.Error("previous current session must no longer be primary") } } -// TestSwitchSessionForStep_ReusesFailedSession exercises the requirement that -// FAILED sessions are reused too. Without this, a previously-failed session -// would be skipped and a fresh one created, leaving the FAILED one as a -// duplicate tab in the UI showing its stale error banner. -func TestSwitchSessionForStep_ReusesFailedSession(t *testing.T) { +// TestSwitchSessionForStep_FailedSessionNotReused mirrors +// TestSwitchSessionForStep_CompletedSessionNotReused for FAILED sessions. +// FAILED is terminal too, and a failed ACP conversation can carry equally +// stale or partial routing intent, so it must not be implicitly resumed +// either — a fresh session is created and the FAILED session is left as +// historical record (including its error message). +func TestSwitchSessionForStep_FailedSessionNotReused(t *testing.T) { ctx := context.Background() now := time.Now().UTC() @@ -891,29 +1072,37 @@ func TestSwitchSessionForStep_ReusesFailedSession(t *testing.T) { scheduler: sched, } - revived, err := svc.switchSessionForStep(ctx, "t1", current, "profile-a") + fresh, err := svc.switchSessionForStep(ctx, "t1", current, "profile-a") if err != nil { t.Fatalf("unexpected error: %v", err) } - if revived == nil || revived.ID != "session-a" { - t.Fatalf("expected reused FAILED session-a, got %+v", revived) + if fresh == nil || fresh.ID == "session-a" { + t.Fatalf("expected a fresh session distinct from the FAILED session-a, got %+v", fresh) + } + if fresh.State != models.TaskSessionStateCreated { + t.Errorf("fresh session state = %s, want CREATED", fresh.State) } - // No duplicate session must be created. + // No session should be reused; a third session now exists. sessions, _ := repo.ListTaskSessions(ctx, "t1") - if len(sessions) != 2 { - t.Errorf("expected 2 sessions, got %d (FAILED session was not reused)", len(sessions)) + if len(sessions) != 3 { + t.Errorf("expected 3 sessions (old failed + parked + fresh), got %d", len(sessions)) } - reused, _ := repo.GetTaskSession(ctx, "session-a") - if reused.State != models.TaskSessionStateWaitingForInput { - t.Errorf("FAILED reused session must flip to WAITING_FOR_INPUT (lazy-resume via token), got %s", reused.State) + // The FAILED session must be left exactly as it was — including the + // error message, which a revive-in-place would have cleared. + oldSession, _ := repo.GetTaskSession(ctx, "session-a") + if oldSession.State != models.TaskSessionStateFailed { + t.Errorf("old session state = %s, want it to remain FAILED", oldSession.State) + } + if oldSession.ErrorMessage != "execution already running" { + t.Errorf("old FAILED session ErrorMessage must be preserved, got %q", oldSession.ErrorMessage) } - if reused.ErrorMessage != "" { - t.Errorf("FAILED reused session must have ErrorMessage cleared, got %q", reused.ErrorMessage) + if oldSession.CompletedAt == nil || !oldSession.CompletedAt.Equal(failedAt) { + t.Errorf("old session CompletedAt = %v, want unchanged %v", oldSession.CompletedAt, failedAt) } - if reused.CompletedAt != nil { - t.Error("FAILED reused session must have CompletedAt cleared") + if oldSession.IsPrimary { + t.Error("old FAILED session must remain non-primary") } } diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 69023f9b43..8255e126c2 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -258,6 +258,7 @@ type sessionExecutorStore interface { GetActiveTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) ListActiveTaskSessionsByTaskID(ctx context.Context, taskID string) ([]*models.TaskSession, error) SetSessionPrimary(ctx context.Context, sessionID string) error + SetSessionPrimaryIfNonterminal(ctx context.Context, sessionID string) (bool, error) RenameTaskSession(ctx context.Context, id, name string) error UpdateTaskSession(ctx context.Context, session *models.TaskSession) error UpdateTaskSessionIfCurrentState(ctx context.Context, session *models.TaskSession, expected models.TaskSessionState) (bool, error) diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index cf64b76737..eecd3e9ddb 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -2737,22 +2737,37 @@ func (r *Repository) GetPrimarySessionInfoByTaskIDs(ctx context.Context, taskIDs // (`SELECT ... FOR UPDATE`) before touching its sessions, so a second // concurrent promotion for the same task blocks until the first commits. func (r *Repository) SetSessionPrimary(ctx context.Context, sessionID string) error { + _, err := r.setSessionPrimary(ctx, sessionID, false) + return err +} + +// SetSessionPrimaryIfNonterminal marks a session primary only while it remains +// nonterminal. It is used by workflow profile switching so a completed agent +// cannot be promoted from a stale lookup and have its ACP conversation resumed. +func (r *Repository) SetSessionPrimaryIfNonterminal(ctx context.Context, sessionID string) (bool, error) { + return r.setSessionPrimary(ctx, sessionID, true) +} + +func (r *Repository) setSessionPrimary(ctx context.Context, sessionID string, requireNonterminal bool) (bool, error) { now := time.Now().UTC() tx, err := r.db.BeginTxx(ctx, nil) if err != nil { - return err + return false, err } defer func() { _ = tx.Rollback() }() - // First, get the task_id for this session + // First, get the task_id for this session. Do not lock the target row here: + // every primary promotion must take the owning task lock first so concurrent + // promotions keep one lock order. var taskID string - err = tx.QueryRowContext(ctx, r.db.Rebind(`SELECT task_id FROM task_sessions WHERE id = ?`), sessionID).Scan(&taskID) + query := `SELECT task_id FROM task_sessions WHERE id = ?` + err = tx.QueryRowContext(ctx, r.db.Rebind(query), sessionID).Scan(&taskID) if err == sql.ErrNoRows { - return fmt.Errorf("session not found: %s", sessionID) + return primarySessionNotPromoted(sessionID, requireNonterminal) } if err != nil { - return err + return false, err } // Serialize concurrent promotions for the same task across Postgres @@ -2762,7 +2777,20 @@ func (r *Repository) SetSessionPrimary(ctx context.Context, sessionID string) er var lockedTaskID string err := tx.QueryRowContext(ctx, r.db.Rebind(`SELECT id FROM tasks WHERE id = ? FOR UPDATE`), taskID).Scan(&lockedTaskID) if err != nil && err != sql.ErrNoRows { - return err + return false, err + } + } + + // Once the task lock is held, lock and validate the target row before + // promoting it. This serializes the nonterminal check with a concurrent + // state transition without reversing the task -> session lock order above. + if requireNonterminal { + valid, err := r.lockNonterminalPrimarySession(ctx, tx, sessionID) + if err != nil { + return false, err + } + if !valid { + return false, nil } } @@ -2771,20 +2799,25 @@ func (r *Repository) SetSessionPrimary(ctx context.Context, sessionID string) er UPDATE task_sessions SET is_primary = 0, updated_at = ? WHERE task_id = ? `), now, taskID) if err != nil { - return err + return false, err } // Set primary flag on the specified session - result, err := tx.ExecContext(ctx, r.db.Rebind(` - UPDATE task_sessions SET is_primary = 1, updated_at = ? WHERE id = ? - `), now, sessionID) + promoteQuery := `UPDATE task_sessions SET is_primary = 1, updated_at = ? WHERE id = ?` + if requireNonterminal { + promoteQuery += ` AND state IN ('CREATED', 'STARTING', 'RUNNING', 'IDLE', 'WAITING_FOR_INPUT')` + } + result, err := tx.ExecContext(ctx, r.db.Rebind(promoteQuery), now, sessionID) if err != nil { - return err + return false, err } rows, _ := result.RowsAffected() if rows == 0 { - return fmt.Errorf("session not found: %s", sessionID) + return primarySessionNotPromoted(sessionID, requireNonterminal) } - return tx.Commit() + if err := tx.Commit(); err != nil { + return false, err + } + return true, nil } diff --git a/apps/backend/internal/task/repository/sqlite/session_primary.go b/apps/backend/internal/task/repository/sqlite/session_primary.go new file mode 100644 index 0000000000..378abb9c44 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/session_primary.go @@ -0,0 +1,31 @@ +package sqlite + +import ( + "context" + "database/sql" + "fmt" + + "github.com/jmoiron/sqlx" + + "github.com/kandev/kandev/internal/db/dialect" +) + +func primarySessionNotPromoted(sessionID string, requireNonterminal bool) (bool, error) { + if requireNonterminal { + return false, nil + } + return false, fmt.Errorf("session not found: %s", sessionID) +} + +func (r *Repository) lockNonterminalPrimarySession(ctx context.Context, tx *sqlx.Tx, sessionID string) (bool, error) { + query := `SELECT id FROM task_sessions WHERE id = ? AND state IN ('CREATED', 'STARTING', 'RUNNING', 'IDLE', 'WAITING_FOR_INPUT')` + if dialect.IsPostgres(r.db.DriverName()) { + query += ` FOR UPDATE` + } + var lockedSessionID string + err := tx.QueryRowContext(ctx, r.db.Rebind(query), sessionID).Scan(&lockedSessionID) + if err == sql.ErrNoRows { + return false, nil + } + return true, err +}