fix(orchestrator): prevent terminal session revival on workflow re-entry

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")

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -1479,10 +1479,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),
@@ -1510,10 +1514,19 @@ func (s *Service) switchSessionForStep(ctx context.Context, taskID string, curre
 	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 +1543,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 +1553,14 @@ 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). findReusableSessionForProfile guarantees `existing` is
+// never terminal (COMPLETED/FAILED/CANCELLED) — see its doc comment for why
+// reviving a terminal session's ACP conversation here would be unsafe.
 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,10 +1569,6 @@ 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 {
@@ -1595,35 +1595,6 @@ 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))
-	}
-}
-
 // createNewSessionForStep is the original switch-and-create-fresh-session path,
 // used when there is no existing session for the target profile.
 func (s *Service) createNewSessionForStep(ctx context.Context, taskID string, currentSession *models.TaskSession, newAgentProfileID string) (*models.TaskSession, error) {