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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> --------- Co-authored-by: Kandev Agent <agent@kandev.dev> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
João Salavisa
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user