fix(orchestrator): drain queued message on workflow transition to non-auto-start step (#726)

* fix(orchestrator): drain queued message on workflow transition to non-auto-start step A workflow transition (e.g. In Progress → Review) made handleAgentReady early-return before draining the message queue, on the assumption that the new step's auto_start_agent would handle the next prompt (#677). For steps without auto_start_agent (Review, etc.) nothing ever drained the queue and user-queued messages stuck forever. Add drainQueuedMessageAfterTransition and call it from the two processOnEnter paths that don't auto-start the agent (no on_enter actions, and the default switch case). Restores pre-#677 behavior for non-auto-start transitions while keeping the deadlock fix for auto-start steps intact. Includes regression test covering the no-on_enter-actions case. * fix(orchestrator): address PR review feedback on queue drain - Log a Warn when drainQueuedMessageAfterTransition discards an empty queued message, matching handleAgentReady's behavior. (Claude) - Fix data race on mockAgentManager.capturedPrompts surfaced by `-race`: PromptAgent now appends under the existing mu, and adds an opt-in promptDone channel so tests can wait deterministically. (Claude) - Replace sleep-based polling in the new regression test with a select on promptDone — matches the project's no-time.Sleep guideline. (Claude, CodeRabbit) - Drain the queue in the sessionSwitched + step.Prompt error handler goroutine in processOnEnter — same orphan-queue class of bug as the rest of this PR, just on a different no-auto-start path. (Greptile P1) --------- Co-authored-by: Kandev Agent <agent@kandev.dev>
2026-04-28 15:40:32 +01:00
parent fbe2410fdb
commit 2416b8dd49
3 changed files with 97 additions and 4 deletions
@@ -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
 }
@@ -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.
@@ -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")