diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 435233064e..77c8a5c37a 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -128,6 +128,9 @@ type mockAgentManager struct { // Prompt tracking capturedPrompts []string // tracks prompts passed to PromptAgent + // Optional: closed once on the first PromptAgent call so tests can wait + // deterministically without polling. Tests opt in by initializing the channel. + promptDone chan struct{} // Passthrough stdin tracking passthroughStdinCalls []passthroughStdinCall @@ -169,12 +172,21 @@ func (m *mockAgentManager) StopAgentWithReason(_ context.Context, agentExecution return nil } func (m *mockAgentManager) PromptAgent(_ context.Context, _ string, prompt string, _ []v1.MessageAttachment) (*executor.PromptResult, error) { + m.mu.Lock() + first := len(m.capturedPrompts) == 0 m.capturedPrompts = append(m.capturedPrompts, prompt) - if m.promptErr != nil { - return nil, m.promptErr + promptErr := m.promptErr + promptResult := m.promptResult + doneCh := m.promptDone + m.mu.Unlock() + if first && doneCh != nil { + close(doneCh) } - if m.promptResult != nil { - return m.promptResult, nil + if promptErr != nil { + return nil, promptErr + } + if promptResult != nil { + return promptResult, nil } return &executor.PromptResult{}, nil } diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 5867c9e7c8..9057554da6 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -610,6 +610,7 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo if len(step.Events.OnEnter) == 0 && !sessionSwitched { s.setSessionWaitingForInput(ctx, taskID, sessionID, session) s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session) + s.drainQueuedMessageAfterTransition(ctx, sessionID) return } @@ -696,15 +697,44 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo zap.Error(err)) s.setSessionWaitingForInput(asyncCtx, taskID, sessionID, session) s.publishSessionWaitingEvent(asyncCtx, taskID, sessionID, stepID, session) + s.drainQueuedMessageAfterTransition(asyncCtx, sessionID) } }() return } s.setSessionWaitingForInput(ctx, taskID, sessionID, session) s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session) + // handleAgentReady early-returns when a workflow transition occurs (#677), + // so user-queued messages would otherwise stick forever on transitions to + // steps without auto_start_agent (e.g. Review). Drain here to match the + // pre-#677 behavior where handleAgentReady always drained after returning + // from inline processOnEnter. + s.drainQueuedMessageAfterTransition(ctx, sessionID) } } +// drainQueuedMessageAfterTransition takes any user-queued message and dispatches +// it for execution. Used by processOnEnter branches that don't auto-start the +// agent — without this, the message is orphaned because handleAgentReady skips +// the queue when a workflow transition occurred. +func (s *Service) drainQueuedMessageAfterTransition(ctx context.Context, sessionID string) { + if s.messageQueue == nil { + return + } + queuedMsg, ok := s.messageQueue.TakeQueued(ctx, sessionID) + if !ok || queuedMsg == nil { + return + } + s.publishQueueStatusEvent(ctx, sessionID) + if queuedMsg.Content == "" && len(queuedMsg.Attachments) == 0 { + s.logger.Warn("skipping empty queued message after transition", + zap.String("session_id", sessionID), + zap.String("queue_id", queuedMsg.ID)) + return + } + go s.executeQueuedMessage(sessionID, queuedMsg) +} + // deliverPassthroughPrompt writes a prompt to PTY stdin and marks the session as running. // Appends \r (carriage return) to simulate pressing Enter — TUI agents in raw terminal mode // expect CR, not LF, as the submit key. Returns an error only if writing fails. diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go index ef5be861b1..df47c510e7 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/task/models" wfmodels "github.com/kandev/kandev/internal/workflow/models" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -355,6 +356,56 @@ func TestProcessOnEnter(t *testing.T) { } }) + // Regression: a workflow transition to a step without auto_start_agent + // (e.g. Review) must still drain a user-queued message. Pre-#677 the drain + // happened in handleAgentReady after inline processOnEnter returned; + // #677 made handleAgentReady early-return on transition, which orphaned + // the queue for any step that didn't auto-start. + t.Run("drains user-queued message when entering step without auto_start_agent", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + // Mirror applyEngineTransition's pre-flip so processOnEnter sees the + // session in the same state it would in production. + session, _ := repo.GetTaskSession(ctx, "s1") + session.State = models.TaskSessionStateWaitingForInput + session.AgentExecutionID = "exec-1" + _ = repo.UpdateTaskSession(ctx, session) + + agentMgr := &mockAgentManager{isAgentRunning: true, promptDone: make(chan struct{})} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + + if _, err := svc.messageQueue.QueueMessage(ctx, "s1", "t1", "user queued msg", "", "user", false, nil); err != nil { + t.Fatalf("failed to seed queued message: %v", err) + } + + reviewStep := &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Review", Position: 1, + // No on_enter actions — this is the broken case before the fix. + } + + session, _ = repo.GetTaskSession(ctx, "s1") + svc.processOnEnter(ctx, "t1", session, reviewStep, "task description") + + select { + case <-agentMgr.promptDone: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for queued message to be sent to agent") + } + + agentMgr.mu.Lock() + got := agentMgr.capturedPrompts[0] + agentMgr.mu.Unlock() + if got != "user queued msg" { + t.Fatalf("agent received %q, want %q", got, "user queued msg") + } + + if status := svc.messageQueue.GetStatus(ctx, "s1"); status.IsQueued { + t.Fatalf("expected queue to be drained, still queued: %+v", status.Message) + } + }) + t.Run("auto_start_agent queues prompt when session is already running", func(t *testing.T) { repo := setupTestRepo(t) seedSession(t, repo, "t1", "s1", "step1")