fix: reap prompt-dead executions on resume (#1602)

apps/backend/internal/orchestrator/event_handlers_test.go

@@ -165,6 +165,7 @@ type mockAgentManager struct {
 	mu                      sync.Mutex
 	stopAgentWithReasonArgs []stopAgentCall // tracks StopAgentWithReason calls
 	stopAgentWithReasonErr  error           // optional error to return from StopAgentWithReason
+	stopAgentWithReasonFunc func(context.Context, string, string, bool) error
 	stopAgentArgs           []stopAgentCall // tracks StopAgent calls (no reason)
 	stopAgentErr            error           // optional error to return from StopAgent
 
@@ -260,15 +261,20 @@ func (m *mockAgentManager) StopAgent(_ context.Context, agentExecutionID string,
 	m.stopAgentArgs = append(m.stopAgentArgs, stopAgentCall{ExecutionID: agentExecutionID, Force: force})
 	return m.stopAgentErr
 }
-func (m *mockAgentManager) StopAgentWithReason(_ context.Context, agentExecutionID, reason string, force bool) error {
+func (m *mockAgentManager) StopAgentWithReason(ctx context.Context, agentExecutionID, reason string, force bool) error {
 	m.mu.Lock()
-	defer m.mu.Unlock()
 	m.stopAgentWithReasonArgs = append(m.stopAgentWithReasonArgs, stopAgentCall{
 		ExecutionID: agentExecutionID,
 		Reason:      reason,
 		Force:       force,
 	})
-	return m.stopAgentWithReasonErr
+	hook := m.stopAgentWithReasonFunc
+	err := m.stopAgentWithReasonErr
+	m.mu.Unlock()
+	if hook != nil {
+		return hook(ctx, agentExecutionID, reason, force)
+	}
+	return err
 }
 func (m *mockAgentManager) PromptAgent(_ context.Context, executionID string, prompt string, _ []v1.MessageAttachment, dispatchOnly bool) (*executor.PromptResult, error) {
 	m.mu.Lock()

apps/backend/internal/orchestrator/prompt_idle_session_test.go

@@ -405,6 +405,44 @@ func TestEnsureSessionRunning_WaitsForPromptReadyWhenExecutionExists(t *testing.
 	}
 }
 
+func TestEnsureSessionRunning_CancelledContextDoesNotReapExecution(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateWaitingForInput)
+	session, err := repo.GetTaskSession(context.Background(), "session1")
+	if err != nil {
+		t.Fatalf("failed to load session: %v", err)
+	}
+	session.AgentProfileID = "profile1"
+	if err := repo.UpdateTaskSession(context.Background(), session); err != nil {
+		t.Fatalf("failed to update session: %v", err)
+	}
+	seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-live")
+
+	agentMgr := &mockAgentManager{
+		isAgentRunning:         true,
+		repoForExecutionLookup: repo,
+		isAgentReadyFn: func(context.Context, string) bool {
+			return false
+		},
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
+	svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+
+	err = svc.ensureSessionRunning(ctx, "session1", session)
+	agentMgr.mu.Lock()
+	stopCalls := append([]stopAgentCall(nil), agentMgr.stopAgentWithReasonArgs...)
+	agentMgr.mu.Unlock()
+	if len(stopCalls) != 0 {
+		t.Fatalf("expected canceled caller context not to stop live execution, got %#v", stopCalls)
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected caller cancellation, got %v", err)
+	}
+}
+
 type promptStreamRecoveringAgentManager struct {
 	*mockAgentManager
 	recovered atomic.Bool
@@ -461,6 +499,134 @@ func TestPromptTask_StalePromptStreamRecoversBeforeTimeout(t *testing.T) {
 	}
 }
 
+func TestPromptTask_ReapsPromptDeadExecutionBeforeSend(t *testing.T) {
+	oldReadyTimeout := agentPromptReadyTimeout
+	oldReadyInterval := agentPromptReadyInterval
+	agentPromptReadyTimeout = 20 * time.Millisecond
+	agentPromptReadyInterval = time.Millisecond
+	t.Cleanup(func() {
+		agentPromptReadyTimeout = oldReadyTimeout
+		agentPromptReadyInterval = oldReadyInterval
+	})
+
+	ctx := context.Background()
+	pollCtx, cancelPoll := context.WithCancel(context.Background())
+	t.Cleanup(cancelPoll)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateWaitingForInput)
+	session, err := repo.GetTaskSession(ctx, "session1")
+	if err != nil {
+		t.Fatalf("failed to load session: %v", err)
+	}
+	session.AgentProfileID = "profile1"
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("failed to update session: %v", err)
+	}
+
+	now := time.Now().UTC()
+	if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
+		ID:               "er1",
+		SessionID:        "session1",
+		TaskID:           "task1",
+		AgentExecutionID: "exec-zombie",
+		Status:           "running",
+		Resumable:        true,
+		ResumeToken:      "resume-token-123",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}); err != nil {
+		t.Fatalf("failed to seed executor running: %v", err)
+	}
+
+	var zombieRunning atomic.Bool
+	zombieRunning.Store(true)
+	var replacementReady atomic.Bool
+	var launchCalls atomic.Int32
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup: repo,
+		isAgentRunningFn: func(_ context.Context, _ string) bool {
+			return zombieRunning.Load()
+		},
+		isAgentReadyFn: func(_ context.Context, _ string) bool {
+			return replacementReady.Load()
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, _ string, _ string, _ bool) error {
+			zombieRunning.Store(false)
+			return nil
+		},
+		launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
+			launchCalls.Add(1)
+			if req.ACPSessionID != "resume-token-123" {
+				t.Errorf("expected preserved resume token, got %q", req.ACPSessionID)
+			}
+			running, err := repo.GetExecutorRunningBySessionID(context.Background(), req.SessionID)
+			if err != nil {
+				t.Errorf("reload executor running: %v", err)
+			} else {
+				running.AgentExecutionID = "exec-replacement"
+				running.Status = "ready"
+				if err := repo.UpsertExecutorRunning(context.Background(), running); err != nil {
+					t.Errorf("persist replacement executor running: %v", err)
+				}
+			}
+			go func(ctx context.Context, sessID string) {
+				tick := time.NewTicker(5 * time.Millisecond)
+				defer tick.Stop()
+				timeout := time.NewTimer(5 * time.Second)
+				defer timeout.Stop()
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-tick.C:
+						sess, err := repo.GetTaskSession(context.Background(), sessID)
+						if err == nil && sess != nil && sess.State == models.TaskSessionStateStarting {
+							sess.State = models.TaskSessionStateWaitingForInput
+							sess.UpdatedAt = time.Now().UTC()
+							_ = repo.UpdateTaskSession(context.Background(), sess)
+							replacementReady.Store(true)
+							return
+						}
+					case <-timeout.C:
+						return
+					}
+				}
+			}(pollCtx, req.SessionID)
+			return &executor.LaunchAgentResponse{AgentExecutionID: "exec-replacement"}, 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{})
+
+	if _, err := svc.PromptTask(ctx, "task1", "session1", "follow up", "", false, nil, false); err != nil {
+		t.Fatalf("expected prompt-dead execution to be replaced before send, got: %v", err)
+	}
+	if got := launchCalls.Load(); got != 1 {
+		t.Fatalf("expected one replacement launch, got %d", got)
+	}
+	if len(agentMgr.capturedPromptCalls) != 1 {
+		t.Fatalf("expected one delivered prompt, got %d", len(agentMgr.capturedPromptCalls))
+	}
+	if got := agentMgr.capturedPromptCalls[0].ExecutionID; got != "exec-replacement" {
+		t.Fatalf("expected prompt to target replacement execution, got %q", got)
+	}
+
+	running, err := repo.GetExecutorRunningBySessionID(ctx, "session1")
+	if err != nil {
+		t.Fatalf("failed to reload executor running row: %v", err)
+	}
+	if running.ResumeToken != "resume-token-123" {
+		t.Fatalf("expected resume token to be preserved, got %q", running.ResumeToken)
+	}
+}
+
 func TestPromptTask_LazyResumeExecutionNotFoundFallsBackToFreshLaunch(t *testing.T) {
 	ctx := context.Background()
 	repo := setupTestRepo(t)

apps/backend/internal/orchestrator/task_operations.go

@@ -39,6 +39,7 @@ const resumeReasonErrorRecovery = "error_recovery"
 const resumeReasonFailedSessionResumable = "failed_session_resumable"
 
 var ErrAgentPromptInProgress = errors.New("agent is currently processing a prompt")
+var ErrAgentNotReadyForPrompt = errors.New("agent not ready for prompt")
 var ErrSessionResetInProgress = errors.New("session reset in progress")
 
 // ErrSessionNotPromptable is returned when a session cannot accept a prompt
@@ -47,7 +48,7 @@ var ErrSessionResetInProgress = errors.New("session reset in progress")
 // the two misleads the UI and any caller doing errors.Is checks.
 var ErrSessionNotPromptable = errors.New("session not promptable")
 
-const (
+var (
 	// Backend restart recovery can restore the session state before the ACP
 	// stream is promptable again. Keep this above CI's slow-start tail so a
 	// valid resume waits instead of surfacing "Failed to send message to agent".
@@ -55,6 +56,8 @@ const (
 	agentPromptReadyInterval = 100 * time.Millisecond
 )
 
+const promptReadinessRecoveryStopReason = "prompt readiness recovery"
+
 type agentPromptStreamRecoverer interface {
 	RecoverAgentPromptStream(ctx context.Context, sessionID string) error
 }
@@ -1052,47 +1055,48 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin
 	// Session resume can take time and shouldn't be tied to the WS request lifecycle.
 	resumeCtx := context.WithoutCancel(ctx)
 	execution, err := s.executor.ResumeSession(resumeCtx, session, true)
+	var readySession *models.TaskSession
 	if err != nil {
 		// If the execution is already running (duplicate resume request), return it as success.
 		if errors.Is(err, executor.ErrExecutionAlreadyRunning) {
-			if existing, ok := s.executor.GetExecutionBySession(sessionID); ok && existing != nil {
-				readySession, waitErr := s.waitForResumedSessionReady(ctx, sessionID)
-				if waitErr != nil {
-					return nil, waitErr
-				}
-				existing.SessionState = v1.TaskSessionState(readySession.State)
-				return existing, nil
+			execution, readySession, err = s.recoverAlreadyRunningResume(ctx, resumeCtx, taskID, sessionID)
+			if err != nil && errors.Is(err, ErrAgentNotReadyForPrompt) {
+				return nil, err
 			}
 		}
-		// Task was archived while the resume was in flight — return the error
-		// without mutating task/session state (archive already handled cleanup).
-		// Check both the sentinel (early rejection) and re-read the task to catch
-		// the race where archive completed after the executor's archived check.
-		if errors.Is(err, executor.ErrTaskArchived) {
+		if err != nil {
+			// Task was archived while the resume was in flight — return the error
+			// without mutating task/session state (archive already handled cleanup).
+			// Check both the sentinel (early rejection) and re-read the task to catch
+			// the race where archive completed after the executor's archived check.
+			if errors.Is(err, executor.ErrTaskArchived) {
+				return nil, err
+			}
+			if task, taskErr := s.repo.GetTask(resumeCtx, taskID); taskErr == nil && task != nil && task.ArchivedAt != nil {
+				return nil, executor.ErrTaskArchived
+			}
+			// Use resumeCtx (WithoutCancel) for the failure-recording writes too —
+			// if the caller's ctx was already cancelled (e.g. WS client navigated
+			// away), the SessionStateFailed and TaskStateFailed updates would
+			// themselves fail with "context canceled" and leave the task stuck
+			// looking "running" forever.
+			s.updateTaskSessionState(resumeCtx, taskID, sessionID, models.TaskSessionStateFailed, err.Error(), false, session)
+			if stateErr := s.taskRepo.UpdateTaskState(resumeCtx, taskID, v1.TaskStateFailed); stateErr != nil {
+				s.logger.Warn("failed to update task state to FAILED after resume error",
+					zap.String("task_id", taskID),
+					zap.String("session_id", sessionID),
+					zap.Error(stateErr))
+			} else {
+				s.processParentChildrenCompletedForTaskState(resumeCtx, taskID, v1.TaskStateFailed)
+			}
 			return nil, err
 		}
-		if task, taskErr := s.repo.GetTask(resumeCtx, taskID); taskErr == nil && task != nil && task.ArchivedAt != nil {
-			return nil, executor.ErrTaskArchived
-		}
-		// Use resumeCtx (WithoutCancel) for the failure-recording writes too —
-		// if the caller's ctx was already cancelled (e.g. WS client navigated
-		// away), the SessionStateFailed and TaskStateFailed updates would
-		// themselves fail with "context canceled" and leave the task stuck
-		// looking "running" forever.
-		s.updateTaskSessionState(resumeCtx, taskID, sessionID, models.TaskSessionStateFailed, err.Error(), false, session)
-		if stateErr := s.taskRepo.UpdateTaskState(resumeCtx, taskID, v1.TaskStateFailed); stateErr != nil {
-			s.logger.Warn("failed to update task state to FAILED after resume error",
-				zap.String("task_id", taskID),
-				zap.String("session_id", sessionID),
-				zap.Error(stateErr))
-		} else {
-			s.processParentChildrenCompletedForTaskState(resumeCtx, taskID, v1.TaskStateFailed)
-		}
-		return nil, err
 	}
-	readySession, err := s.waitForResumedSessionReady(ctx, sessionID)
-	if err != nil {
-		return nil, err
+	if readySession == nil {
+		readySession, err = s.waitForResumedSessionReady(ctx, sessionID)
+		if err != nil {
+			return nil, err
+		}
 	}
 	execution.SessionState = v1.TaskSessionState(readySession.State)
 
@@ -1129,6 +1133,50 @@ func (s *Service) waitForResumedSessionReady(ctx context.Context, sessionID stri
 	return s.waitForSessionAndAgentReady(ctx, sessionID, "after resume")
 }
 
+func (s *Service) recoverAlreadyRunningResume(
+	ctx context.Context,
+	resumeCtx context.Context,
+	taskID string,
+	sessionID string,
+) (*executor.TaskExecution, *models.TaskSession, error) {
+	existing, ok := s.executor.GetExecutionBySession(sessionID)
+	if !ok || existing == nil {
+		return nil, nil, executor.ErrExecutionAlreadyRunning
+	}
+
+	readySession, waitErr := s.waitForResumedSessionReady(ctx, sessionID)
+	if waitErr == nil {
+		existing.SessionState = v1.TaskSessionState(readySession.State)
+		return existing, readySession, nil
+	}
+	if !errors.Is(waitErr, ErrAgentNotReadyForPrompt) {
+		return nil, nil, waitErr
+	}
+
+	if err := s.reapPromptUnreadyExecution(resumeCtx, sessionID, waitErr); err != nil {
+		return nil, nil, fmt.Errorf("%w: failed to stop prompt-unready execution: %w", ErrAgentNotReadyForPrompt, err)
+	}
+
+	session, err := s.repo.GetTaskSession(resumeCtx, sessionID)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to reload session after prompt-readiness recovery: %w", err)
+	}
+	if session.TaskID != taskID {
+		return nil, nil, fmt.Errorf("task session does not belong to task")
+	}
+
+	execution, err := s.executor.ResumeSession(resumeCtx, session, true)
+	if err != nil {
+		return nil, nil, err
+	}
+	readySession, err = s.waitForResumedSessionReady(ctx, sessionID)
+	if err != nil {
+		return nil, nil, err
+	}
+	execution.SessionState = v1.TaskSessionState(readySession.State)
+	return execution, readySession, nil
+}
+
 func (s *Service) waitForSessionAndAgentReady(ctx context.Context, sessionID, waitContext string) (*models.TaskSession, error) {
 	if err := s.waitForSessionReady(ctx, sessionID); err != nil {
 		return nil, fmt.Errorf("session not ready %s: %w", waitContext, err)
@@ -1249,9 +1297,21 @@ func (s *Service) ensureSessionRunning(ctx context.Context, sessionID string, se
 	if exec, ok := s.executor.GetExecutionBySession(sessionID); ok && exec != nil {
 		s.recoverAgentPromptStreamIfNeeded(ctx, sessionID)
 		if err := s.waitForAgentPromptReady(ctx, sessionID); err != nil {
-			return err
+			if !errors.Is(err, ErrAgentNotReadyForPrompt) {
+				return err
+			}
+			recoveryCtx := context.WithoutCancel(ctx)
+			if stopErr := s.reapPromptUnreadyExecution(recoveryCtx, sessionID, err); stopErr != nil {
+				return fmt.Errorf("failed to stop prompt-unready execution: %w", stopErr)
+			}
+			refreshed, refreshErr := s.repo.GetTaskSession(recoveryCtx, sessionID)
+			if refreshErr != nil {
+				return fmt.Errorf("failed to reload session after prompt-readiness recovery: %w", refreshErr)
+			}
+			session = refreshed
+		} else {
+			return nil
 		}
-		return nil
 	}
 
 	s.logger.Debug("agent not running for session, attempting resume",
@@ -1286,7 +1346,11 @@ func (s *Service) ensureSessionRunning(ctx context.Context, sessionID string, se
 	resumeCtx := context.WithoutCancel(ctx)
 	if _, err = s.executor.ResumeSession(resumeCtx, session, true); err != nil {
 		if errors.Is(err, executor.ErrExecutionAlreadyRunning) {
-			return nil // Agent is already running, nothing to do
+			s.recoverAgentPromptStreamIfNeeded(ctx, sessionID)
+			if readyErr := s.waitForAgentPromptReady(ctx, sessionID); readyErr != nil {
+				return fmt.Errorf("agent not ready after resume race: %w", readyErr)
+			}
+			return nil
 		}
 		s.updateTaskSessionState(ctx, session.TaskID, sessionID, models.TaskSessionStateFailed, err.Error(), false, session)
 		if stateErr := s.taskRepo.UpdateTaskState(ctx, session.TaskID, v1.TaskStateFailed); stateErr != nil {
@@ -1328,6 +1392,33 @@ func (s *Service) recoverAgentPromptStreamIfNeeded(ctx context.Context, sessionI
 	}
 }
 
+func (s *Service) reapPromptUnreadyExecution(ctx context.Context, sessionID string, cause error) error {
+	if s.agentManager == nil {
+		return fmt.Errorf("agent manager is not configured")
+	}
+	executionID, err := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
+	if err != nil || executionID == "" {
+		if running, runErr := s.repo.GetExecutorRunningBySessionID(ctx, sessionID); runErr == nil && running != nil {
+			executionID = running.AgentExecutionID
+		}
+	}
+	if executionID == "" {
+		if err != nil {
+			return fmt.Errorf("execution id for session %s: %w", sessionID, err)
+		}
+		return fmt.Errorf("execution id for session %s not found", sessionID)
+	}
+
+	s.logger.Warn("stopping prompt-unready agent execution before resume",
+		zap.String("session_id", sessionID),
+		zap.String("agent_execution_id", executionID),
+		zap.Error(cause))
+	if err := s.executor.StopExecution(ctx, executionID, promptReadinessRecoveryStopReason, true); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (s *Service) waitForAgentPromptReady(ctx context.Context, sessionID string) error {
 	if s.agentManager == nil {
 		return nil
@@ -1346,7 +1437,10 @@ func (s *Service) waitForAgentPromptReady(ctx context.Context, sessionID string)
 
 		select {
 		case <-readyCtx.Done():
-			return fmt.Errorf("agent not ready for prompt: %w", readyCtx.Err())
+			if err := ctx.Err(); err != nil {
+				return err
+			}
+			return fmt.Errorf("%w: %w", ErrAgentNotReadyForPrompt, readyCtx.Err())
 		case <-ticker.C:
 		}
 	}

apps/backend/internal/orchestrator/task_operations_resume_test.go

@@ -2,6 +2,8 @@ package orchestrator
 
 import (
 	"context"
+	"errors"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -429,3 +431,220 @@ func TestResumeTaskSession_WaitsForPromptReady(t *testing.T) {
 		t.Fatal("ResumeTaskSession did not return after prompt readiness")
 	}
 }
+
+func TestResumeTaskSession_ReapsPromptDeadExecutionAndRelaunches(t *testing.T) {
+	oldReadyTimeout := agentPromptReadyTimeout
+	oldReadyInterval := agentPromptReadyInterval
+	agentPromptReadyTimeout = 20 * time.Millisecond
+	agentPromptReadyInterval = time.Millisecond
+	t.Cleanup(func() {
+		agentPromptReadyTimeout = oldReadyTimeout
+		agentPromptReadyInterval = oldReadyInterval
+	})
+
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateWaitingForInput)
+
+	session, err := repo.GetTaskSession(ctx, "session1")
+	if err != nil {
+		t.Fatalf("failed to load session: %v", err)
+	}
+	session.AgentProfileID = "profile1"
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("failed to update session: %v", err)
+	}
+
+	now := time.Now().UTC()
+	if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
+		ID:               "er1",
+		SessionID:        "session1",
+		TaskID:           "task1",
+		AgentExecutionID: "exec-zombie",
+		Status:           "running",
+		Resumable:        true,
+		ResumeToken:      "resume-token-123",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}); err != nil {
+		t.Fatalf("failed to seed executor running: %v", err)
+	}
+
+	var zombieRunning atomic.Bool
+	zombieRunning.Store(true)
+	var replacementReady atomic.Bool
+	var launchCalls atomic.Int32
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup: repo,
+		isAgentRunningFn: func(_ context.Context, _ string) bool {
+			return zombieRunning.Load()
+		},
+		isAgentReadyFn: func(_ context.Context, _ string) bool {
+			return replacementReady.Load()
+		},
+		stopAgentWithReasonFunc: func(_ context.Context, _ string, _ string, _ bool) error {
+			zombieRunning.Store(false)
+			return nil
+		},
+		launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
+			launchCalls.Add(1)
+			if req.ACPSessionID != "resume-token-123" {
+				t.Errorf("expected preserved resume token, got %q", req.ACPSessionID)
+			}
+			go func(sessID string) {
+				tick := time.NewTicker(5 * time.Millisecond)
+				defer tick.Stop()
+				timeout := time.After(5 * time.Second)
+				for {
+					select {
+					case <-tick.C:
+						sess, err := repo.GetTaskSession(context.Background(), sessID)
+						if err == nil && sess != nil && sess.State == models.TaskSessionStateStarting {
+							sess.State = models.TaskSessionStateWaitingForInput
+							sess.UpdatedAt = time.Now().UTC()
+							_ = repo.UpdateTaskSession(context.Background(), sess)
+							replacementReady.Store(true)
+							return
+						}
+					case <-timeout:
+						return
+					}
+				}
+			}(req.SessionID)
+			return &executor.LaunchAgentResponse{AgentExecutionID: "exec-replacement"}, 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("expected prompt-dead execution to be reaped and relaunched, got: %v", err)
+	}
+	if exec == nil {
+		t.Fatal("expected replacement execution")
+	}
+	if exec.AgentExecutionID != "exec-replacement" {
+		t.Fatalf("expected replacement execution id, got %q", exec.AgentExecutionID)
+	}
+	if got := launchCalls.Load(); got != 1 {
+		t.Fatalf("expected one replacement launch, got %d", got)
+	}
+
+	agentMgr.mu.Lock()
+	stopCalls := append([]stopAgentCall(nil), agentMgr.stopAgentWithReasonArgs...)
+	agentMgr.mu.Unlock()
+	if len(stopCalls) != 1 {
+		t.Fatalf("expected one zombie stop call, got %d", len(stopCalls))
+	}
+	if stopCalls[0].ExecutionID != "exec-zombie" || !stopCalls[0].Force {
+		t.Fatalf("unexpected zombie stop call: %#v", stopCalls[0])
+	}
+
+	running, err := repo.GetExecutorRunningBySessionID(ctx, "session1")
+	if err != nil {
+		t.Fatalf("failed to reload executor running row: %v", err)
+	}
+	if running.ResumeToken != "resume-token-123" {
+		t.Fatalf("expected resume token to be preserved, got %q", running.ResumeToken)
+	}
+}
+
+func TestResumeTaskSession_ReapStopFailureDoesNotFailTask(t *testing.T) {
+	oldReadyTimeout := agentPromptReadyTimeout
+	oldReadyInterval := agentPromptReadyInterval
+	agentPromptReadyTimeout = 20 * time.Millisecond
+	agentPromptReadyInterval = time.Millisecond
+	t.Cleanup(func() {
+		agentPromptReadyTimeout = oldReadyTimeout
+		agentPromptReadyInterval = oldReadyInterval
+	})
+
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateWaitingForInput)
+
+	session, err := repo.GetTaskSession(ctx, "session1")
+	if err != nil {
+		t.Fatalf("failed to load session: %v", err)
+	}
+	session.AgentProfileID = "profile1"
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("failed to update session: %v", err)
+	}
+
+	now := time.Now().UTC()
+	if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
+		ID:               "er1",
+		SessionID:        "session1",
+		TaskID:           "task1",
+		AgentExecutionID: "exec-zombie",
+		Status:           "running",
+		Resumable:        true,
+		ResumeToken:      "resume-token-123",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}); err != nil {
+		t.Fatalf("failed to seed executor running: %v", err)
+	}
+
+	var launchCalls atomic.Int32
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup: repo,
+		isAgentRunningFn: func(_ context.Context, _ string) bool {
+			return true
+		},
+		isAgentReadyFn: func(_ context.Context, _ string) bool {
+			return false
+		},
+		stopAgentWithReasonErr: errors.New("stop timeout"),
+		launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
+			launchCalls.Add(1)
+			return &executor.LaunchAgentResponse{AgentExecutionID: "exec-replacement"}, 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.Fatal("expected stop failure to be returned")
+	}
+	if exec != nil {
+		t.Fatalf("expected no execution on stop failure, got %#v", exec)
+	}
+	if !errors.Is(err, ErrAgentNotReadyForPrompt) {
+		t.Fatalf("expected prompt-not-ready error, got: %v", err)
+	}
+	if got := launchCalls.Load(); got != 0 {
+		t.Fatalf("expected no replacement launch after failed stop, got %d", got)
+	}
+
+	refreshed, err := repo.GetTaskSession(ctx, "session1")
+	if err != nil {
+		t.Fatalf("failed to reload session: %v", err)
+	}
+	if refreshed.State == models.TaskSessionStateFailed {
+		t.Fatalf("expected session to remain retryable, got state %s", refreshed.State)
+	}
+	if got := taskRepo.updatedStates["task1"]; got == v1.TaskStateFailed {
+		t.Fatal("expected task state not to be marked failed on transient stop failure")
+	}
+	if got := taskRepo.stateWrites["task1"]; got != 0 {
+		t.Fatalf("expected no task state writes on transient stop failure, got %d", got)
+	}
+}