From 2416b8dd49ab84ef4bfa423becd11a8622e5e2b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Tue, 28 Apr 2026 15:40:32 +0100 Subject: [PATCH] fix(orchestrator): drain queued message on workflow transition to non-auto-start step (#726) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 --- .../orchestrator/event_handlers_test.go | 20 ++++++-- .../orchestrator/event_handlers_workflow.go | 30 +++++++++++ .../event_handlers_workflow_triggers_test.go | 51 +++++++++++++++++++ 3 files changed, 97 insertions(+), 4 deletions(-) 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")