diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index b9fc73a57b..c291377350 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -109,9 +109,13 @@ func (s *Service) requeueMessage(ctx context.Context, queuedMsg *messagequeue.Qu // agent.ready (turn-end) so the orchestrator never has to disambiguate the // two with race-prone flags. // -// The only job here is to flip the session to WAITING_FOR_INPUT so callers +// Two jobs here: (1) flip the session to WAITING_FOR_INPUT so callers // that are gating on that state (e.g. PromptTask's waitForSessionReady after -// ensureSessionRunning kicked off ResumeSession) can proceed. Crucially we do +// ensureSessionRunning kicked off ResumeSession) can proceed; (2) drain any +// orphaned queued message. Without the drain, a workflow auto-start prompt +// queued against a session that died mid-turn (or before its first prompt) +// would sit forever after the user resumed it — agent.ready (the usual drain +// trigger) never fires for a turn that never completed. Crucially we do // NOT call processOnTurnCompleteViaEngine — there's no turn to complete, and // stepping the workflow off a boot signal is what caused the production // ping-pong bug. @@ -159,15 +163,23 @@ func (s *Service) handleAgentBootReady(ctx context.Context, data watcher.AgentEv // Idempotent: if the session is already WAITING_FOR_INPUT (e.g. revived // from a previously launched session and the boot signal arrived faster - // than persistResumeState wrote STARTING), skip — there's nothing to flip - // and waitForSessionReady's poll will pick it up. + // than persistResumeState wrote STARTING), skip the flip — but still + // fall through to the drain below: an orphaned queued message would + // otherwise sit forever. if session.State == models.TaskSessionStateWaitingForInput { - s.logger.Debug("agent.boot_ready: session already WAITING_FOR_INPUT, no-op", + s.logger.Debug("agent.boot_ready: session already WAITING_FOR_INPUT, skipping flip", zap.String("session_id", data.SessionID)) - return + } else { + s.setSessionWaitingForInput(ctx, data.TaskID, data.SessionID, session) } - s.setSessionWaitingForInput(ctx, data.TaskID, data.SessionID, session) + // Drain any orphaned queued message. handleAgentReady drains on turn-end, + // but a session that crashed mid-turn (or never started its first turn) + // won't fire agent.ready — leaving e.g. workflow auto-start prompts stuck + // on the queue until the user manually sends another message. After the + // agent has booted and the session is back to WAITING_FOR_INPUT it's safe + // to dispatch any pending message. + s.drainQueuedMessageAfterTransition(ctx, data.SessionID) } // handleAgentReady handles turn-end ready events: the agent finished processing diff --git a/apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go b/apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go new file mode 100644 index 0000000000..552f20dc05 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go @@ -0,0 +1,248 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestAutoStartTransientError_BootReadyDrainsOrphanedQueue is the regression +// test for the production stuck-queue symptom on task 9378f7cf. +// +// Setup mirrors the production scenario verbatim: +// - Step "Fixup" has auto_start_agent + on_turn_complete=move_to_next. +// - Step "Merge" has only auto_start_agent (no on_turn_complete). +// - Task at Fixup, session.State=RUNNING (Fixup turn in flight). +// - The underlying agent stream drops mid-prompt — PromptAgent returns +// "agent stream disconnected", which isTransientPromptError matches. +// +// What happens without the boot-ready drain (the bug): +// 1. handleAgentReady → applyEngineTransition flips state, transitions +// Fixup→Merge, spawns processOnEnter(Merge) in a goroutine. +// 2. processOnEnter → autoStartStepPrompt: +// a. recordAutoStartMessage records the Merge prompt as a user msg. +// b. PromptTask → executor.Prompt → PromptAgent returns the transient +// error after ~5 s in production. +// c. handlePromptError reverts state + completeTurnForSession (this +// is the 5-second "ghost turn" observed in the production DB). +// d. autoStartStepPrompt's retry loop matches isTransientPromptError +// and queues the same Merge prompt with queued_by=workflow. +// 3. The queue sits forever because: +// - The agent never produced an agent.ready for this prompt (the +// stream dropped first), so handleAgentReady's drain never fires. +// - The user manually resumes the session → handleAgentBootReady +// fires → BEFORE the fix, that handler only flipped state and +// returned, leaving the queue orphaned. +// +// With the boot-ready drain fix in handleAgentBootReady, step 3 drains the +// queue when the session is resumed — the test asserts both halves end-to-end: +// the duplicate is created on transient failure, then boot_ready drains it. +func TestAutoStartTransientError_BootReadyDrainsOrphanedQueue(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-1" + sessionID = "sess-1" + executionID = "exec-1" + fixupStepID = "step-fixup" + mergeStepID = "step-merge" + profile = "profile-impl" + taskWorkflow = "wf1" + ) + + repo := setupTestRepo(t) + now := time.Now().UTC() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: taskWorkflow, WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workflow: %v", err) + } + + stepGetter := newMockStepGetter() + stepGetter.steps[fixupStepID] = &wfmodels.WorkflowStep{ + ID: fixupStepID, WorkflowID: taskWorkflow, Name: "Fixup after PR", Position: 1, + AgentProfileID: profile, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}}, + OnTurnComplete: []wfmodels.OnTurnCompleteAction{ + {Type: wfmodels.OnTurnCompleteMoveToStep, Config: map[string]interface{}{"step_id": mergeStepID}}, + }, + }, + } + stepGetter.steps[mergeStepID] = &wfmodels.WorkflowStep{ + ID: mergeStepID, WorkflowID: taskWorkflow, Name: "Merge", Position: 2, + AgentProfileID: profile, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}}, + }, + } + + if err := repo.CreateTask(ctx, &models.Task{ + ID: taskID, WorkflowID: taskWorkflow, WorkflowStepID: fixupStepID, + Title: "Test", Description: "Test task", + State: v1.TaskStateInProgress, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create task: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + AgentProfileID: profile, + AgentExecutionID: executionID, + State: models.TaskSessionStateRunning, // mid-turn for Fixup + IsPrimary: true, + StartedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create session: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ + ID: taskID, WorkflowID: taskWorkflow, State: v1.TaskStateInProgress, + } + + // Stream-disconnected error matches isTransientPromptError → autoStartStepPrompt + // records the user msg, then queues on the retry path. promptDone closes on + // the first PromptAgent call so the test can sync on it without time.Sleep. + firstPromptCalled := make(chan struct{}) + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: true, // skip ensureSessionRunning's resume + promptErr: errors.New("agent stream disconnected mid-prompt"), + promptDone: firstPromptCalled, + } + + msgCreator := &mockMessageCreator{} + + exec := executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + svc := &Service{ + logger: testLogger(), + repo: repo, + taskRepo: taskRepo, + agentManager: agentMgr, + messageQueue: messagequeue.NewServiceMemory(testLogger()), + executor: exec, + messageCreator: msgCreator, + } + svc.SetWorkflowStepGetter(stepGetter) + + // --- Phase 1: agent.ready triggers the transient-failure auto-start path --- + + doneFired := make(chan struct{}) + go func() { + defer close(doneFired) + svc.handleAgentReady(ctx, watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + AgentProfileID: profile, + }) + }() + select { + case <-doneFired: + case <-time.After(3 * time.Second): + t.Fatalf("handleAgentReady did not return within 3s") + } + + // Sync on PromptAgent's first call (closes firstPromptCalled). After that, + // the goroutine returns from PromptAgent and runs handlePromptError + + // autoStartStepPrompt's queue branch — a fully synchronous sequence that + // ends with messageQueue.QueueMessageWithMetadata. A tight poll covers + // the few-microsecond gap between PromptAgent returning and the queue + // row being committed (an unbuffered channel isn't available there). + select { + case <-firstPromptCalled: + case <-time.After(3 * time.Second): + t.Fatalf("PromptAgent was not called within 3s") + } + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + if svc.messageQueue.GetStatus(ctx, sessionID).Count > 0 { + break + } + time.Sleep(1 * time.Millisecond) + } + + // Verify the workflow transitioned to Merge. + updatedTask, err := repo.GetTask(ctx, taskID) + if err != nil { + t.Fatalf("load task: %v", err) + } + if updatedTask.WorkflowStepID != mergeStepID { + t.Errorf("task step = %q, want %q", updatedTask.WorkflowStepID, mergeStepID) + } + + // The transient-failure path records the user msg AND queues the prompt + // — both for the same Merge auto-start. This is the production data shape + // (one chat row + one queue entry, both with workflow_step_name=Merge). + queueStatus := svc.messageQueue.GetStatus(ctx, sessionID) + if queueStatus.Count != 1 { + t.Fatalf("expected exactly 1 queued message after transient failure, got %d", queueStatus.Count) + } + queued := queueStatus.Entries[0] + if queued.QueuedBy != messagequeue.QueuedByWorkflow { + t.Errorf("queued_by = %q, want %q", queued.QueuedBy, messagequeue.QueuedByWorkflow) + } + if name, _ := queued.Metadata["workflow_step_name"].(string); name != "Merge" { + t.Errorf("queued metadata workflow_step_name = %q, want Merge", name) + } + mergeUserMsgs := 0 + for _, m := range msgCreator.userMessages { + if name, _ := m.metadata["workflow_step_name"].(string); name == "Merge" { + mergeUserMsgs++ + } + } + if mergeUserMsgs != 1 { + t.Errorf("expected exactly 1 Merge user_message (from recordAutoStartMessage before PromptTask failed), got %d", mergeUserMsgs) + } + + // --- Phase 2: user resumes the session — boot_ready must drain the queue --- + + // Clear the transient error so the drain's executeQueuedMessage → PromptTask + // → PromptAgent succeeds. Otherwise the goroutine that runs after the queue + // take would re-enter the same transient-retry path and re-queue the message + // immediately, racing the queue-empty assertion below. Also swap in a fresh + // promptDone channel to signal the second PromptAgent call (the original one + // was already closed by Phase 1). + secondPromptCalled := make(chan struct{}) + agentMgr.mu.Lock() + agentMgr.promptErr = nil + agentMgr.promptDone = secondPromptCalled + // Reset capturedPrompts' bookkeeping so the channel re-fires; the mock's + // `first := len(m.capturedPrompts) == 0` guard would otherwise skip the close. + agentMgr.capturedPrompts = agentMgr.capturedPrompts[:0] + agentMgr.capturedPromptCalls = agentMgr.capturedPromptCalls[:0] + agentMgr.mu.Unlock() + + // Simulate the resume: agentctl's ACP session has re-initialized so the + // lifecycle manager fires events.AgentBootReady. Without the drain fix, + // the queue stays orphaned. With the fix, it gets drained. + svc.handleAgentBootReady(ctx, watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + }) + + // drainQueuedMessageAfterTransition pops the queue synchronously and fires + // `go executeQueuedMessage(...)`. The goroutine creates a user message and + // calls PromptTask → PromptAgent. The mock closes secondPromptCalled on + // that call, so the assertion below has a deterministic sync point. + select { + case <-secondPromptCalled: + case <-time.After(3 * time.Second): + t.Fatalf("boot_ready drain did not reach PromptAgent within 3s") + } + + if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 { + t.Errorf("queue not drained after boot_ready: %d messages still queued (the orphaned-queue bug is back)", got) + } +} 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 b6ca06ac01..d6fecf04cf 100644 --- a/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go @@ -601,3 +601,174 @@ func TestHandleAgentBootReady_DoesNotTriggerOnTurnComplete(t *testing.T) { }) } } + +// TestHandleAgentBootReady_DrainsOrphanedQueuedMessage reproduces the +// production stuck-queue symptom: a workflow auto-start prompt is queued +// against a session, the agent dies before the turn completes (so no +// agent.ready fires to drain it), and the user resumes the session. The +// session ends up WAITING_FOR_INPUT with the message still on the queue — +// "1 queued" displayed in the UI forever — because handleAgentBootReady +// only flipped state but never drained. +// +// After the fix, handleAgentBootReady takes the queued message and dispatches +// it (via executeQueuedMessage in a goroutine). The test wires a full +// executor + seeded executors_running row so the goroutine's PromptTask +// call lands on a working code path instead of nil-derefing s.executor +// under -race. +func TestHandleAgentBootReady_DrainsOrphanedQueuedMessage(t *testing.T) { + ctx := context.Background() + now := time.Now().UTC() + + cases := []struct { + name string + startSt models.TaskSessionState + }{ + {"STARTING -> WAITING_FOR_INPUT (boot completes resume)", models.TaskSessionStateStarting}, + {"already WAITING_FOR_INPUT (boot raced persistResumeState)", models.TaskSessionStateWaitingForInput}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + repo := setupTestRepo(t) + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-1", WorkflowID: "wf1", WorkflowStepID: "step-merge", + Title: "T", Description: "D", State: v1.TaskStateInProgress, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create task: %v", err) + } + + sessionID := "s1" + const executionID = "exec-1" + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, TaskID: "task-1", AgentProfileID: "profile-impl", + AgentExecutionID: executionID, + State: tc.startSt, + IsPrimary: true, + StartedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create session: %v", err) + } + // Seed executors_running so PromptTask -> ensureSessionRunning -> + // executor.GetExecutionBySession finds the agent and skips resume. + seedExecutorRunning(t, repo, sessionID, "task-1", executionID) + + taskRepo := newMockTaskRepo() + taskRepo.tasks["task-1"] = &v1.Task{ + ID: "task-1", WorkflowID: "wf1", State: v1.TaskStateInProgress, + } + + log := testLogger() + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: true, // satisfy GetExecutionBySession's IsAgentRunningForSession check + } + svc := &Service{ + logger: log, + repo: repo, + taskRepo: taskRepo, + agentManager: agentMgr, + messageQueue: messagequeue.NewServiceMemory(log), + // Wire a real executor so the executeQueuedMessage goroutine + // spawned by drainQueuedMessageAfterTransition can safely call + // PromptTask -> executor.GetExecutionBySession without nil-derefing. + executor: executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}), + } + + // Seed an orphaned workflow auto-start prompt — what the production + // bug looked like in the DB at task 9378f7cf. + if _, err := svc.messageQueue.QueueMessage( + ctx, sessionID, "task-1", "ROLE: Merge operator. ...", "", + messagequeue.QueuedByWorkflow, false, nil, + ); err != nil { + t.Fatalf("queue orphaned prompt: %v", err) + } + if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 1 { + t.Fatalf("precondition: queue count = %d, want 1", got) + } + + svc.handleAgentBootReady(ctx, watcher.AgentEventData{ + TaskID: "task-1", SessionID: sessionID, + }) + + if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 { + t.Errorf("queue count after boot ready = %d, want 0 (orphaned message must be drained)", got) + } + + // Session must still end up WAITING_FOR_INPUT — the drain is + // additive, not a replacement for the existing flip behavior. + finalSess, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("load session: %v", err) + } + if finalSess.State != models.TaskSessionStateWaitingForInput { + t.Errorf("session.State = %q, want WAITING_FOR_INPUT", finalSess.State) + } + }) + } +} + +// TestHandleAgentBootReady_DoesNotDrainForTerminalSession guards against +// reviving a queued message on a session that was cancelled or completed — +// the user explicitly stopped this session, the queued prompt should NOT be +// dispatched, and the early-return for terminal states must continue to +// short-circuit before the drain. +func TestHandleAgentBootReady_DoesNotDrainForTerminalSession(t *testing.T) { + ctx := context.Background() + now := time.Now().UTC() + + repo := setupTestRepo(t) + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-1", WorkflowID: "wf1", WorkflowStepID: "step-merge", + Title: "T", Description: "D", State: v1.TaskStateInProgress, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create task: %v", err) + } + + sessionID := "s1" + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, TaskID: "task-1", AgentProfileID: "profile-impl", + State: models.TaskSessionStateCancelled, + IsPrimary: true, + StartedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create session: %v", err) + } + + log := testLogger() + svc := &Service{ + logger: log, + repo: repo, + taskRepo: newMockTaskRepo(), + agentManager: &mockAgentManager{repoForExecutionLookup: repo}, + messageQueue: messagequeue.NewServiceMemory(log), + } + + if _, err := svc.messageQueue.QueueMessage( + ctx, sessionID, "task-1", "stuck prompt", "", + messagequeue.QueuedByWorkflow, false, nil, + ); err != nil { + t.Fatalf("queue prompt: %v", err) + } + + svc.handleAgentBootReady(ctx, watcher.AgentEventData{ + TaskID: "task-1", SessionID: sessionID, + }) + + if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 1 { + t.Errorf("queue count after boot ready on terminal session = %d, want 1 (must not drain)", got) + } +}