From dcab8aaa05aaf934d20b43212023ab3bfc7ce6f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Tue, 26 May 2026 16:02:15 +0100 Subject: [PATCH] fix(backend): drain stuck auto-start prompts after step transitions (#1087) * fix(backend): drain stuck auto-start prompts after step transitions Adds flipStaleRunningToWaiting, a helper that detects when a session's in-memory state reports RUNNING/STARTING but the orchestrator's authoritative activeTurns map has no entry, and pre-flips the session to WAITING_FOR_INPUT. processOnEnter now calls this before reaching queueAutoStartPromptIfRunning, so the auto-start prompt is sent directly via PromptTask instead of being queued against a session whose agent.ready will never fire. Co-Authored-By: Claude Sonnet 4.6 * docs(backend): clarify reset guard scope and add STARTING-state test - Expand isSessionResetInProgress comment to note it guards concurrent goroutines only; sequential reset_agent_context OnEnter actions run after flipStaleRunningToWaiting returns (markIdleAfterReset no-ops if state is already WAITING_FOR_INPUT). - Add call-site comment explaining why the pre-flip must happen before the len(step.Events.OnEnter)==0 early-return check. - Add test: flipStaleRunningToWaiting flips STARTING session with no active turn (addresses reviewer gap noted in code review). Co-Authored-By: Claude Sonnet 4.6 --------- Co-authored-by: Kandev Agent Co-authored-by: Claude Sonnet 4.6 --- .../orchestrator/event_handlers_workflow.go | 74 +++++++++ .../event_handlers_workflow_moved_test.go | 144 ++++++++++++++++++ .../event_handlers_workflow_triggers_test.go | 4 + 3 files changed, 222 insertions(+) diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 8490bd0a44..17fc93d3aa 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -812,6 +812,19 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo sessionID := session.ID isPassthrough := s.agentManager.IsPassthroughSession(ctx, sessionID) + // Stale session.State left over from a previous turn (e.g. the agent's + // agent.ready fired before this goroutine resumed) would otherwise trick + // queueAutoStartPromptIfRunning into queueing the auto-start prompt against + // a session that's actually idle — and nothing would drain it because no + // future agent.ready is pending. Flip to WAITING_FOR_INPUT when activeTurns + // confirms no in-flight turn. + // + // This must run before the len(step.Events.OnEnter)==0 early-return below. + // A stale-RUNNING session on a no-OnEnter step should still transition to + // WAITING_FOR_INPUT and drain its queue; without the pre-flip it would + // early-return unchanged, leaving session.State==RUNNING with no drain path. + s.flipStaleRunningToWaiting(ctx, taskID, session, isPassthrough) + hasPlanMode := s.resolveStepPlanMode(ctx, session, step, isPassthrough) if len(step.Events.OnEnter) == 0 && !sessionSwitched { @@ -1396,6 +1409,12 @@ func (s *Service) recordAutoStartMessage( } } +// queueAutoStartPromptIfRunning queues the workflow auto-start prompt only +// when the session is currently RUNNING/STARTING, returning queued=true on +// success. Callers must ensure session.State is fresh — a stale RUNNING flag +// (no in-flight turn) would queue a prompt that nothing drains. processOnEnter +// runs flipStaleRunningToWaiting before reaching this path; applyEngineTransition +// flips the same way inline. New call sites must do the same. func (s *Service) queueAutoStartPromptIfRunning( ctx context.Context, taskID string, session *models.TaskSession, prompt string, @@ -1455,6 +1474,61 @@ func (s *Service) queueAutoStartPrompt( return nil } +// flipStaleRunningToWaiting flips the session to WAITING_FOR_INPUT when its +// state claims RUNNING/STARTING but the orchestrator's authoritative +// activeTurns map shows no in-flight turn. This catches the manual-move race +// where processOnEnter runs from the task.moved goroutine with a session +// pointer loaded *before* the previous turn's agent.ready fired (or after that +// agent.ready already completed but the DB hadn't propagated yet). Without +// this flip, queueAutoStartPromptIfRunning would queue the auto-start prompt +// against a session no longer mid-turn, and no future agent.ready would fire +// to drain it. +// +// Skip when: +// - session.State is not RUNNING/STARTING (CREATED routes through +// StartCreatedSession; terminal states are immutable); +// - the session is passthrough (PTY-driven, manages its own RUNNING/idle +// transitions via MarkPassthroughRunning); +// - a context reset is in progress (resetAgentContext owns the state +// machine until it completes and runs markIdleAfterReset); +// - activeTurns has an entry for the session (a turn is genuinely in flight, +// queueing is correct and agent.ready will drain). +// +// applyEngineTransition (engine-driven on_turn_complete path) already does +// this flip inline at the call site; that path stays untouched — the check +// here is idempotent, so even if both fired, the second is a no-op. +// +// Returns true when the flip happened (mostly useful for tests and logs). +func (s *Service) flipStaleRunningToWaiting(ctx context.Context, taskID string, session *models.TaskSession, isPassthrough bool) bool { + if session.State != models.TaskSessionStateRunning && + session.State != models.TaskSessionStateStarting { + return false + } + if isPassthrough { + return false + } + // Guards against a concurrent goroutine running resetAgentContext for the + // same session — not the sequential reset_agent_context OnEnter action that + // may run later in processOnEnter after this function returns. If both + // reset_agent_context and auto_start_agent appear in OnEnter, the flip here + // is still correct: resetAgentContext runs after this returns, then + // markIdleAfterReset sees WAITING_FOR_INPUT and no-ops (idempotent). + if s.isSessionResetInProgress(session.ID) { + return false + } + if _, hasActiveTurn := s.activeTurns.Load(session.ID); hasActiveTurn { + return false + } + priorState := session.State + s.updateTaskSessionState(ctx, taskID, session.ID, models.TaskSessionStateWaitingForInput, "", false, session) + session.State = models.TaskSessionStateWaitingForInput + s.logger.Info("flipped stale RUNNING session to WAITING_FOR_INPUT (no active turn registered)", + zap.String("task_id", taskID), + zap.String("session_id", session.ID), + zap.String("prior_state", string(priorState))) + return true +} + // markIdleAfterReset flips a freshly-reset session to WAITING_FOR_INPUT so a // following auto_start_agent sends the prompt directly instead of queueing // against a stale RUNNING state. processOnEnter runs from handleAgentReady, diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_moved_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_moved_test.go index 890af2d9a2..69572d27d3 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_moved_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_moved_test.go @@ -265,6 +265,12 @@ func TestHandleTaskMovedWithSession(t *testing.T) { } svc := createTestService(repo, stepGetter, newMockTaskRepo()) + // Seed an active-turn entry so flipStaleRunningToWaiting treats the + // session as genuinely mid-turn and leaves auto-start on the queue + // path. Without this, the helper would flip state to WAITING_FOR_INPUT + // and autoStartStepPrompt would fall through to PromptTask (which + // dereferences the executor that's nil in this minimal test setup). + svc.activeTurns.Store("s1", "turn-1") svc.handleTaskMovedWithSession(ctx, watcher.TaskMovedEventData{ TaskID: "t1", SessionID: "s1", @@ -301,6 +307,10 @@ func TestHandleTaskMovedWithSession(t *testing.T) { } svc := createTestService(repo, stepGetter, newMockTaskRepo()) + // Seed an active-turn entry so flipStaleRunningToWaiting recognises the + // session as genuinely mid-turn and the auto-start prompt is queued + // (the behavior this subtest asserts). + svc.activeTurns.Store("s1", "turn-1") svc.handleTaskMovedWithSession(ctx, watcher.TaskMovedEventData{ TaskID: "t1", SessionID: "s1", @@ -361,6 +371,140 @@ func TestHandleTaskMovedWithSession(t *testing.T) { } }) + // Regression: an in-memory session pointer can read RUNNING/STARTING even + // when the agent's turn is actually done (the manual-move goroutine loaded + // the row before agent.ready fired, or the previous turn's bookkeeping + // hasn't propagated). Without flipStaleRunningToWaiting, processOnEnter + // would queue the auto-start prompt against this stale state and nothing + // would drain it — the symptom users observe as "QUEUED 1 of N" stuck in + // the chat after a step move. + t.Run("flipStaleRunningToWaiting flips RUNNING session with no active turn", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") // seeds session.State = RUNNING + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + // Intentionally do NOT seed svc.activeTurns — this simulates the stale + // state where session.State claims RUNNING but no turn is in flight. + + session, _ := repo.GetTaskSession(ctx, "s1") + if session.State != models.TaskSessionStateRunning { + t.Fatalf("precondition: session should be RUNNING, got %q", session.State) + } + + flipped := svc.flipStaleRunningToWaiting(ctx, "t1", session, false) + if !flipped { + t.Fatal("expected flipStaleRunningToWaiting to flip stale RUNNING session") + } + if session.State != models.TaskSessionStateWaitingForInput { + t.Errorf("in-memory state: want WAITING_FOR_INPUT, got %q", session.State) + } + updated, _ := repo.GetTaskSession(ctx, "s1") + if updated.State != models.TaskSessionStateWaitingForInput { + t.Errorf("DB state: want WAITING_FOR_INPUT, got %q", updated.State) + } + }) + + // Regression for #1036: when activeTurns has an entry, the session is + // genuinely mid-turn; agent.ready will drain the queue. The helper must + // leave state alone so the queue path stays the right answer. + t.Run("flipStaleRunningToWaiting leaves session running when active turn registered", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.activeTurns.Store("s1", "turn-active") + + session, _ := repo.GetTaskSession(ctx, "s1") + flipped := svc.flipStaleRunningToWaiting(ctx, "t1", session, false) + if flipped { + t.Fatal("expected flipStaleRunningToWaiting to no-op when an active turn is registered") + } + if session.State != models.TaskSessionStateRunning { + t.Errorf("in-memory state: want RUNNING preserved, got %q", session.State) + } + updated, _ := repo.GetTaskSession(ctx, "s1") + if updated.State != models.TaskSessionStateRunning { + t.Errorf("DB state: want RUNNING preserved, got %q", updated.State) + } + }) + + t.Run("flipStaleRunningToWaiting flips STARTING session with no active turn", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + session, _ := repo.GetTaskSession(ctx, "s1") + session.State = models.TaskSessionStateStarting + _ = repo.UpdateTaskSession(ctx, session) + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + // No activeTurns entry — STARTING with no registered turn should flip. + flipped := svc.flipStaleRunningToWaiting(ctx, "t1", session, false) + if !flipped { + t.Fatal("expected STARTING session with no active turn to be flipped to WAITING_FOR_INPUT") + } + if session.State != models.TaskSessionStateWaitingForInput { + t.Errorf("in-memory state: want WAITING_FOR_INPUT, got %q", session.State) + } + updated, _ := repo.GetTaskSession(ctx, "s1") + if updated.State != models.TaskSessionStateWaitingForInput { + t.Errorf("DB state: want WAITING_FOR_INPUT, got %q", updated.State) + } + }) + + t.Run("flipStaleRunningToWaiting no-ops for passthrough sessions", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + session, _ := repo.GetTaskSession(ctx, "s1") + + flipped := svc.flipStaleRunningToWaiting(ctx, "t1", session, true) + if flipped { + t.Fatal("expected flipStaleRunningToWaiting to no-op for passthrough sessions") + } + if session.State != models.TaskSessionStateRunning { + t.Errorf("passthrough session state must be left alone, got %q", session.State) + } + }) + + t.Run("flipStaleRunningToWaiting no-ops for CREATED sessions", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + session, _ := repo.GetTaskSession(ctx, "s1") + session.State = models.TaskSessionStateCreated + _ = repo.UpdateTaskSession(ctx, session) + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + flipped := svc.flipStaleRunningToWaiting(ctx, "t1", session, false) + if flipped { + t.Fatal("expected flipStaleRunningToWaiting to no-op for CREATED sessions") + } + if session.State != models.TaskSessionStateCreated { + t.Errorf("CREATED session state must be left alone, got %q", session.State) + } + }) + + t.Run("flipStaleRunningToWaiting no-ops while session reset is in progress", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + // resetAgentContext owns the state machine during reset; the helper + // must not race with it. + svc.setSessionResetInProgress("s1", true) + t.Cleanup(func() { svc.setSessionResetInProgress("s1", false) }) + + session, _ := repo.GetTaskSession(ctx, "s1") + flipped := svc.flipStaleRunningToWaiting(ctx, "t1", session, false) + if flipped { + t.Fatal("expected flipStaleRunningToWaiting to no-op while reset is in progress") + } + if session.State != models.TaskSessionStateRunning { + t.Errorf("session state must be left alone during reset, got %q", session.State) + } + }) + t.Run("handles missing from-step gracefully", func(t *testing.T) { synctest.Test(t, func(t *testing.T) { repo := setupTestRepo(t) 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 15012a482b..2552dd86aa 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go @@ -415,6 +415,10 @@ func TestProcessOnEnter(t *testing.T) { taskRepo := newMockTaskRepo() svc := createTestService(repo, newMockStepGetter(), taskRepo) + // Seed an active-turn entry so flipStaleRunningToWaiting treats the + // session as genuinely mid-turn and the auto-start prompt is queued + // (the behavior this subtest asserts). + svc.activeTurns.Store("s1", "turn-1") attachments := []messagequeue.MessageAttachment{ {Type: "image", Data: "base64-data", MimeType: "image/png"}, }