diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index a3da52ac14..7a5a6aa426 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/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() diff --git a/apps/backend/internal/orchestrator/prompt_idle_session_test.go b/apps/backend/internal/orchestrator/prompt_idle_session_test.go index 2e6fd8c170..c37cbd904f 100644 --- a/apps/backend/internal/orchestrator/prompt_idle_session_test.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index a4c9fed82d..9204121e28 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/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: } } diff --git a/apps/backend/internal/orchestrator/task_operations_resume_test.go b/apps/backend/internal/orchestrator/task_operations_resume_test.go index aa7a6f3e5d..8b60fd3e6a 100644 --- a/apps/backend/internal/orchestrator/task_operations_resume_test.go +++ b/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) + } +}