diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index c291377350..4a861936de 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -369,7 +369,7 @@ func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messag zap.String("queue_id", queuedMsg.ID), zap.Error(err)) - if isAgentPromptInProgressError(err) || isTransientPromptError(err) || isSessionResetInProgressError(err) { + if isSessionBusyError(err) || isTransientPromptError(err) || isSessionResetInProgressError(err) { s.logger.Warn("queued message execution failed transiently; requeueing", zap.String("session_id", callerSessionID), zap.String("task_id", queuedMsg.TaskID), diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 17fc93d3aa..7a446aa14e 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -1259,7 +1259,7 @@ func (s *Service) autoStartStepPrompt( return nil } - if !isAgentPromptInProgressError(err) && !isTransientPromptError(err) && !isSessionResetInProgressError(err) { + if !isSessionBusyError(err) && !isTransientPromptError(err) && !isSessionResetInProgressError(err) { requeueTaken() return err } diff --git a/apps/backend/internal/orchestrator/prompt_idle_session_test.go b/apps/backend/internal/orchestrator/prompt_idle_session_test.go new file mode 100644 index 0000000000..83cd325ba6 --- /dev/null +++ b/apps/backend/internal/orchestrator/prompt_idle_session_test.go @@ -0,0 +1,232 @@ +package orchestrator + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestPromptTask_IdleSession_IsPromptable reproduces the office-mode IDLE +// re-engagement bug. An IDLE office session has its conversation parked +// (ACP session preserved) but no live agent process. Sending a prompt MUST +// wake the session up rather than be bounced by the "agent is currently +// processing a prompt" guard. +// +// Before the fix, checkSessionPromptable's default branch returned +// ErrAgentPromptInProgress for IDLE — the misleading error the user reported. +func TestPromptTask_IdleSession_IsPromptable(t *testing.T) { + repo := setupTestRepo(t) + taskRepo := newMockTaskRepo() + + // Model the post-resume state: agent is running, executors_running row + // exists. This keeps the test focused on the checkSessionPromptable bug; + // the IDLE→resume path itself is covered by + // TestEnsureSessionRunning_IdleSessionTriggersResume. + agentMgr := &mockAgentManager{ + isAgentRunning: true, + repoForExecutionLookup: repo, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateIdle) + session, err := repo.GetTaskSession(context.Background(), "session1") + if err != nil { + t.Fatalf("failed to load session: %v", err) + } + session.AgentExecutionID = "exec-idle-1" + seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-idle-1") + if err := repo.UpdateTaskSession(context.Background(), session); err != nil { + t.Fatalf("failed to update session: %v", err) + } + + _, err = svc.PromptTask(context.Background(), "task1", "session1", "follow-up", "", false, nil, false) + if err != nil { + if errors.Is(err, ErrAgentPromptInProgress) { + t.Fatalf("IDLE session must not surface ErrAgentPromptInProgress: %v", err) + } + if strings.Contains(err.Error(), "agent is currently processing a prompt") { + t.Fatalf("must not return 'agent is currently processing a prompt' for IDLE: %v", err) + } + t.Fatalf("expected IDLE session to accept prompt, got: %v", err) + } + + // The prompt was forwarded to the (post-resume) live agent. + if len(agentMgr.capturedPrompts) != 1 { + t.Fatalf("expected one captured prompt, got %d", len(agentMgr.capturedPrompts)) + } +} + +// TestIsSessionBusyError pins down the requeue umbrella: STARTING / CREATED / +// RUNNING all need to come back true so executeQueuedMessage and the auto-start +// loop in event_handlers_workflow.go requeue rather than drop. Without this +// helper, splitting ErrAgentPromptInProgress and ErrSessionNotPromptable +// would silently lose queued messages that hit a session mid-startup. +func TestIsSessionBusyError(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"unrelated", errors.New("something else"), false}, + {"in-progress sentinel", ErrAgentPromptInProgress, true}, + {"wrapped in-progress", fmt.Errorf("outer: %w", ErrAgentPromptInProgress), true}, + {"not-promptable sentinel", ErrSessionNotPromptable, true}, + {"wrapped not-promptable", fmt.Errorf("outer: %w", ErrSessionNotPromptable), true}, + {"reset is NOT busy", ErrSessionResetInProgress, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := isSessionBusyError(tc.err); got != tc.want { + t.Errorf("isSessionBusyError(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +// TestCheckSessionPromptable_StateMatrix exercises every state to lock in the +// IDLE acceptance and document which states are rejected and with which error. +// RUNNING → ErrAgentPromptInProgress (someone else is mid-turn). +// IDLE / COMPLETED / WAITING_FOR_INPUT → accepted. +// Everything else → ErrSessionNotPromptable (NOT ErrAgentPromptInProgress). +func TestCheckSessionPromptable_StateMatrix(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + cases := []struct { + state models.TaskSessionState + wantErr bool + wantIs error + }{ + {models.TaskSessionStateWaitingForInput, false, nil}, + {models.TaskSessionStateCompleted, false, nil}, + {models.TaskSessionStateIdle, false, nil}, + {models.TaskSessionStateRunning, true, ErrAgentPromptInProgress}, + {models.TaskSessionStateStarting, true, ErrSessionNotPromptable}, + {models.TaskSessionStateFailed, true, ErrSessionNotPromptable}, + {models.TaskSessionStateCancelled, true, ErrSessionNotPromptable}, + {models.TaskSessionStateCreated, true, ErrSessionNotPromptable}, + } + for _, tc := range cases { + t.Run(string(tc.state), func(t *testing.T) { + err := svc.checkSessionPromptable("t", "s", tc.state) + if tc.wantErr { + if err == nil { + t.Fatalf("expected error for state %q", tc.state) + } + if tc.wantIs != nil && !errors.Is(err, tc.wantIs) { + t.Fatalf("state %q: expected errors.Is(%v); got %v", tc.state, tc.wantIs, err) + } + // Non-RUNNING states must NOT be classified as + // "agent in progress" — that misleads the UI and any caller + // doing errors.Is(err, ErrAgentPromptInProgress) checks. + if tc.state != models.TaskSessionStateRunning && errors.Is(err, ErrAgentPromptInProgress) { + t.Fatalf("state %q must not wrap ErrAgentPromptInProgress: %v", tc.state, err) + } + } else if err != nil { + t.Fatalf("expected nil for state %q, got: %v", tc.state, err) + } + }) + } +} + +// TestEnsureSessionRunning_IdleSessionTriggersResume confirms that an IDLE +// session with a preserved executors_running row (the normal office-mode IDLE +// shape, since handleOfficeTurnComplete tears down the agent process but leaves +// the row intact) routes through ResumeSession when the in-memory execution +// store has no live entry. This pins down the second half of the bug: even +// with checkSessionPromptable fixed, ensureSessionRunning must not bounce IDLE +// — it must resume. +func TestEnsureSessionRunning_IdleSessionTriggersResume(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateIdle) + session, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("failed to load session: %v", err) + } + session.AgentExecutionID = "exec-idle-1" + session.AgentProfileID = "profile1" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("failed to update session: %v", err) + } + seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-idle-1") + + // Mock launch: report success and capture the call so the test can confirm + // the resume path triggered. isAgentRunning starts false (true IDLE shape). + // After LaunchAgent returns, ResumeSession's persistResumeState forces the + // session into STARTING; we then need to flip to WAITING_FOR_INPUT so the + // downstream waitForSessionReady poll exits. We watch for STARTING to + // appear (deterministic — no time.Sleep race against persistResumeState) + // then flip, mirroring the real AgentBootReady → handleAgentBootReady flow. + launchCalled := make(chan struct{}, 1) + agentMgr := &mockAgentManager{ + isAgentRunning: false, + repoForExecutionLookup: repo, + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + select { + case launchCalled <- struct{}{}: + default: + } + go func(sessID string) { + // Watch for STARTING (committed by persistResumeState) then flip + // — guarantees ordering without time.Sleep. Ticker + select per + // AGENTS.md "channel-based synchronization over sleep-based + // waits" guidance (testing/synctest doesn't help here because + // the SQLite repo runs real I/O goroutines that synctest can't + // fully observe as idle). + 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) + return + } + case <-timeout: + return + } + } + }(req.SessionID) + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-resumed-1"}, 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{}) + + // Re-load + session, err = repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("reload: %v", err) + } + + if err := svc.ensureSessionRunning(ctx, "session1", session); err != nil { + t.Fatalf("ensureSessionRunning failed for IDLE session: %v", err) + } + select { + case <-launchCalled: + default: + t.Fatal("expected ResumeSession to call LaunchAgent on IDLE session, but it never fired") + } +} diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 64b0681c37..2df622c8e8 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -40,10 +40,32 @@ const resumeReasonFailedSessionResumable = "failed_session_resumable" var ErrAgentPromptInProgress = errors.New("agent is currently processing a prompt") var ErrSessionResetInProgress = errors.New("session reset in progress") +// ErrSessionNotPromptable is returned when a session cannot accept a prompt +// because of its lifecycle state (STARTING, CREATED, FAILED, CANCELLED). +// Distinct from ErrAgentPromptInProgress, which is RUNNING-only — confusing +// the two misleads the UI and any caller doing errors.Is checks. +var ErrSessionNotPromptable = errors.New("session not promptable") + func isAgentPromptInProgressError(err error) bool { return err != nil && errors.Is(err, ErrAgentPromptInProgress) } +// isSessionBusyError reports whether the session is in a state where a queued +// or auto-started prompt should be retried later rather than dropped. Covers +// both "the agent is mid-turn" (ErrAgentPromptInProgress, RUNNING) and "the +// session isn't yet ready to accept input" (ErrSessionNotPromptable — +// STARTING, CREATED, FAILED, CANCELLED). The pre-PR code path collapsed both +// into ErrAgentPromptInProgress; this helper preserves the requeue behaviour +// after the error split so queued messages targeting a session that is +// briefly STARTING/CREATED don't get silently dropped (see TODO about the +// missing dead-letter queue in executeQueuedMessage). +func isSessionBusyError(err error) bool { + if err == nil { + return false + } + return errors.Is(err, ErrAgentPromptInProgress) || errors.Is(err, ErrSessionNotPromptable) +} + func isSessionResetInProgressError(err error) bool { return err != nil && errors.Is(err, ErrSessionResetInProgress) } @@ -1933,10 +1955,18 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom } // checkSessionPromptable returns nil when the session's state accepts a new -// prompt, or an ErrAgentPromptInProgress-wrapped error otherwise. +// prompt. RUNNING is rejected with ErrAgentPromptInProgress; any other +// non-acceptable state (STARTING / CREATED / FAILED / CANCELLED) is rejected +// with ErrSessionNotPromptable so callers can distinguish "wait for the +// current turn" from "this session is not in a state where it can take a +// prompt". IDLE is acceptable: office sessions intentionally park in IDLE +// between turns (agent torn down, ACP session preserved) and the next prompt +// resumes them — see ensureSessionRunning. func (s *Service) checkSessionPromptable(taskID, sessionID string, state models.TaskSessionState) error { switch state { - case models.TaskSessionStateWaitingForInput, models.TaskSessionStateCompleted: + case models.TaskSessionStateWaitingForInput, + models.TaskSessionStateCompleted, + models.TaskSessionStateIdle: return nil case models.TaskSessionStateRunning: s.logger.Warn("rejected prompt while agent is already running", @@ -1949,7 +1979,7 @@ func (s *Service) checkSessionPromptable(taskID, sessionID string, state models. zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.String("session_state", string(state))) - return fmt.Errorf("%w, session is in %s state", ErrAgentPromptInProgress, state) + return fmt.Errorf("%w: session is in %s state", ErrSessionNotPromptable, state) } }