fix(backend): drain orphaned queued messages on agent boot ready (#1096)
* fix(backend): drain orphaned queued messages on agent boot ready When autoStartStepPrompt hits a transient PromptTask error (e.g. "agent stream disconnected" mid-prompt), it records the auto-start user message and then queues the same prompt for retry. The drain only fires from handleAgentReady (turn-end) — but the failed prompt never reached the agent, so no agent.ready ever arrives for it. After the user resumes the session, handleAgentBootReady only flipped state and returned, leaving the workflow auto-start prompt stuck on the queue forever (the "1 queued" that wouldn't go away after resume). handleAgentBootReady now drains via drainQueuedMessageAfterTransition after the WAITING_FOR_INPUT flip — also on the early-return path where state was already WAITING (boot signal raced persistResumeState). Terminal sessions short-circuit before the drain so cancelled/completed sessions don't revive their queued prompts. Tests: TestAutoStartTransientError_BootReadyDrainsOrphanedQueue reproduces the production scenario end-to-end (transient PromptAgent failure → record + queue → boot_ready → drained). Plus TestHandleAgentBootReady_DrainsOrphanedQueuedMessage covers both STARTING→WAITING and the persistResumeState-race path, and TestHandleAgentBootReady_DoesNotDrainForTerminalSession guards the terminal-session short-circuit. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * test(backend): wait for drain goroutine and verify prompt dispatch in repro Greptile flagged that the queue-empty assertion in TestAutoStartTransientError_BootReadyDrainsOrphanedQueue raced the drain's executeQueuedMessage goroutine: the goroutine would re-enter the same isTransientPromptError path and re-queue the message immediately, but the assertion would usually win on timing. Clear agentMgr.promptErr before Phase 2 (simulating the agent recovering), then poll capturedPrompts until the drained prompt actually reaches PromptAgent. This turns a timing-fragile assertion into a deterministic end-to-end check: drain → executeQueuedMessage → PromptTask → PromptAgent → queue empty. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * test(backend): wire executor in boot-ready drain test to fix nil-deref under -race Claude and cubic flagged that TestHandleAgentBootReady_DrainsOrphanedQueuedMessage constructed Service without an executor field. The drainQueuedMessageAfterTransition goroutine spawned by handleAgentBootReady would nil-deref s.executor via PromptTask -> ensureSessionRunning -> s.executor.GetExecutionBySession. Passed today by luck of scheduling but would panic under -race. Wire a real executor (built from the same mockAgentManager and repo as the other tests) and seed an executors_running row so the drained PromptTask lands on a working code path instead of crashing. Verified passes with `go test -race -count=3`. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * test(backend): replace time.Sleep polling with promptDone channel sync in repro Claude flagged that the polling loops in TestAutoStartTransientError_BootReadyDrainsOrphanedQueue used time.Sleep where mockAgentManager.promptDone (closes on first PromptAgent call) gives a deterministic sync point per AGENTS.md. - Phase 1: wait on firstPromptCalled, then a 500ms / 1ms tight poll covers the synchronous PromptAgent-returns → queue-row-commits gap. - Phase 2: swap in a fresh secondPromptCalled channel (the mock's `first := len(capturedPrompts) == 0` guard requires resetting captured* slices so the close re-fires for the drain's second call), then wait on that signal before the queue-empty assertion. Verified with `go test -race -count=5` — all pass, no goroutine leaks. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> --------- Co-authored-by: Kandev Agent <agent@kandev.dev> Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
João Salavisa
@@ -109,9 +109,13 @@ func (s *Service) requeueMessage(ctx context.Context, queuedMsg *messagequeue.Qu
|
||||
// agent.ready (turn-end) so the orchestrator never has to disambiguate the
|
||||
// two with race-prone flags.
|
||||
//
|
||||
// The only job here is to flip the session to WAITING_FOR_INPUT so callers
|
||||
// Two jobs here: (1) flip the session to WAITING_FOR_INPUT so callers
|
||||
// that are gating on that state (e.g. PromptTask's waitForSessionReady after
|
||||
// ensureSessionRunning kicked off ResumeSession) can proceed. Crucially we do
|
||||
// ensureSessionRunning kicked off ResumeSession) can proceed; (2) drain any
|
||||
// orphaned queued message. Without the drain, a workflow auto-start prompt
|
||||
// queued against a session that died mid-turn (or before its first prompt)
|
||||
// would sit forever after the user resumed it — agent.ready (the usual drain
|
||||
// trigger) never fires for a turn that never completed. Crucially we do
|
||||
// NOT call processOnTurnCompleteViaEngine — there's no turn to complete, and
|
||||
// stepping the workflow off a boot signal is what caused the production
|
||||
// ping-pong bug.
|
||||
@@ -159,15 +163,23 @@ func (s *Service) handleAgentBootReady(ctx context.Context, data watcher.AgentEv
|
||||
|
||||
// Idempotent: if the session is already WAITING_FOR_INPUT (e.g. revived
|
||||
// from a previously launched session and the boot signal arrived faster
|
||||
// than persistResumeState wrote STARTING), skip — there's nothing to flip
|
||||
// and waitForSessionReady's poll will pick it up.
|
||||
// than persistResumeState wrote STARTING), skip the flip — but still
|
||||
// fall through to the drain below: an orphaned queued message would
|
||||
// otherwise sit forever.
|
||||
if session.State == models.TaskSessionStateWaitingForInput {
|
||||
s.logger.Debug("agent.boot_ready: session already WAITING_FOR_INPUT, no-op",
|
||||
s.logger.Debug("agent.boot_ready: session already WAITING_FOR_INPUT, skipping flip",
|
||||
zap.String("session_id", data.SessionID))
|
||||
return
|
||||
} else {
|
||||
s.setSessionWaitingForInput(ctx, data.TaskID, data.SessionID, session)
|
||||
}
|
||||
|
||||
s.setSessionWaitingForInput(ctx, data.TaskID, data.SessionID, session)
|
||||
// Drain any orphaned queued message. handleAgentReady drains on turn-end,
|
||||
// but a session that crashed mid-turn (or never started its first turn)
|
||||
// won't fire agent.ready — leaving e.g. workflow auto-start prompts stuck
|
||||
// on the queue until the user manually sends another message. After the
|
||||
// agent has booted and the session is back to WAITING_FOR_INPUT it's safe
|
||||
// to dispatch any pending message.
|
||||
s.drainQueuedMessageAfterTransition(ctx, data.SessionID)
|
||||
}
|
||||
|
||||
// handleAgentReady handles turn-end ready events: the agent finished processing
|
||||
|
||||
@@ -0,0 +1,248 @@
|
||||
package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/orchestrator/executor"
|
||||
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
|
||||
"github.com/kandev/kandev/internal/orchestrator/watcher"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
wfmodels "github.com/kandev/kandev/internal/workflow/models"
|
||||
v1 "github.com/kandev/kandev/pkg/api/v1"
|
||||
)
|
||||
|
||||
// TestAutoStartTransientError_BootReadyDrainsOrphanedQueue is the regression
|
||||
// test for the production stuck-queue symptom on task 9378f7cf.
|
||||
//
|
||||
// Setup mirrors the production scenario verbatim:
|
||||
// - Step "Fixup" has auto_start_agent + on_turn_complete=move_to_next.
|
||||
// - Step "Merge" has only auto_start_agent (no on_turn_complete).
|
||||
// - Task at Fixup, session.State=RUNNING (Fixup turn in flight).
|
||||
// - The underlying agent stream drops mid-prompt — PromptAgent returns
|
||||
// "agent stream disconnected", which isTransientPromptError matches.
|
||||
//
|
||||
// What happens without the boot-ready drain (the bug):
|
||||
// 1. handleAgentReady → applyEngineTransition flips state, transitions
|
||||
// Fixup→Merge, spawns processOnEnter(Merge) in a goroutine.
|
||||
// 2. processOnEnter → autoStartStepPrompt:
|
||||
// a. recordAutoStartMessage records the Merge prompt as a user msg.
|
||||
// b. PromptTask → executor.Prompt → PromptAgent returns the transient
|
||||
// error after ~5 s in production.
|
||||
// c. handlePromptError reverts state + completeTurnForSession (this
|
||||
// is the 5-second "ghost turn" observed in the production DB).
|
||||
// d. autoStartStepPrompt's retry loop matches isTransientPromptError
|
||||
// and queues the same Merge prompt with queued_by=workflow.
|
||||
// 3. The queue sits forever because:
|
||||
// - The agent never produced an agent.ready for this prompt (the
|
||||
// stream dropped first), so handleAgentReady's drain never fires.
|
||||
// - The user manually resumes the session → handleAgentBootReady
|
||||
// fires → BEFORE the fix, that handler only flipped state and
|
||||
// returned, leaving the queue orphaned.
|
||||
//
|
||||
// With the boot-ready drain fix in handleAgentBootReady, step 3 drains the
|
||||
// queue when the session is resumed — the test asserts both halves end-to-end:
|
||||
// the duplicate is created on transient failure, then boot_ready drains it.
|
||||
func TestAutoStartTransientError_BootReadyDrainsOrphanedQueue(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const (
|
||||
taskID = "task-1"
|
||||
sessionID = "sess-1"
|
||||
executionID = "exec-1"
|
||||
fixupStepID = "step-fixup"
|
||||
mergeStepID = "step-merge"
|
||||
profile = "profile-impl"
|
||||
taskWorkflow = "wf1"
|
||||
)
|
||||
|
||||
repo := setupTestRepo(t)
|
||||
now := time.Now().UTC()
|
||||
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workspace: %v", err)
|
||||
}
|
||||
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: taskWorkflow, WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workflow: %v", err)
|
||||
}
|
||||
|
||||
stepGetter := newMockStepGetter()
|
||||
stepGetter.steps[fixupStepID] = &wfmodels.WorkflowStep{
|
||||
ID: fixupStepID, WorkflowID: taskWorkflow, Name: "Fixup after PR", Position: 1,
|
||||
AgentProfileID: profile,
|
||||
Events: wfmodels.StepEvents{
|
||||
OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}},
|
||||
OnTurnComplete: []wfmodels.OnTurnCompleteAction{
|
||||
{Type: wfmodels.OnTurnCompleteMoveToStep, Config: map[string]interface{}{"step_id": mergeStepID}},
|
||||
},
|
||||
},
|
||||
}
|
||||
stepGetter.steps[mergeStepID] = &wfmodels.WorkflowStep{
|
||||
ID: mergeStepID, WorkflowID: taskWorkflow, Name: "Merge", Position: 2,
|
||||
AgentProfileID: profile,
|
||||
Events: wfmodels.StepEvents{
|
||||
OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}},
|
||||
},
|
||||
}
|
||||
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: taskID, WorkflowID: taskWorkflow, WorkflowStepID: fixupStepID,
|
||||
Title: "Test", Description: "Test task",
|
||||
State: v1.TaskStateInProgress,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
if err := repo.CreateTaskSession(ctx, &models.TaskSession{
|
||||
ID: sessionID,
|
||||
TaskID: taskID,
|
||||
AgentProfileID: profile,
|
||||
AgentExecutionID: executionID,
|
||||
State: models.TaskSessionStateRunning, // mid-turn for Fixup
|
||||
IsPrimary: true,
|
||||
StartedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create session: %v", err)
|
||||
}
|
||||
seedExecutorRunning(t, repo, sessionID, taskID, executionID)
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
taskRepo.tasks[taskID] = &v1.Task{
|
||||
ID: taskID, WorkflowID: taskWorkflow, State: v1.TaskStateInProgress,
|
||||
}
|
||||
|
||||
// Stream-disconnected error matches isTransientPromptError → autoStartStepPrompt
|
||||
// records the user msg, then queues on the retry path. promptDone closes on
|
||||
// the first PromptAgent call so the test can sync on it without time.Sleep.
|
||||
firstPromptCalled := make(chan struct{})
|
||||
agentMgr := &mockAgentManager{
|
||||
repoForExecutionLookup: repo,
|
||||
isAgentRunning: true, // skip ensureSessionRunning's resume
|
||||
promptErr: errors.New("agent stream disconnected mid-prompt"),
|
||||
promptDone: firstPromptCalled,
|
||||
}
|
||||
|
||||
msgCreator := &mockMessageCreator{}
|
||||
|
||||
exec := executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
|
||||
svc := &Service{
|
||||
logger: testLogger(),
|
||||
repo: repo,
|
||||
taskRepo: taskRepo,
|
||||
agentManager: agentMgr,
|
||||
messageQueue: messagequeue.NewServiceMemory(testLogger()),
|
||||
executor: exec,
|
||||
messageCreator: msgCreator,
|
||||
}
|
||||
svc.SetWorkflowStepGetter(stepGetter)
|
||||
|
||||
// --- Phase 1: agent.ready triggers the transient-failure auto-start path ---
|
||||
|
||||
doneFired := make(chan struct{})
|
||||
go func() {
|
||||
defer close(doneFired)
|
||||
svc.handleAgentReady(ctx, watcher.AgentEventData{
|
||||
TaskID: taskID,
|
||||
SessionID: sessionID,
|
||||
AgentExecutionID: executionID,
|
||||
AgentProfileID: profile,
|
||||
})
|
||||
}()
|
||||
select {
|
||||
case <-doneFired:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatalf("handleAgentReady did not return within 3s")
|
||||
}
|
||||
|
||||
// Sync on PromptAgent's first call (closes firstPromptCalled). After that,
|
||||
// the goroutine returns from PromptAgent and runs handlePromptError +
|
||||
// autoStartStepPrompt's queue branch — a fully synchronous sequence that
|
||||
// ends with messageQueue.QueueMessageWithMetadata. A tight poll covers
|
||||
// the few-microsecond gap between PromptAgent returning and the queue
|
||||
// row being committed (an unbuffered channel isn't available there).
|
||||
select {
|
||||
case <-firstPromptCalled:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatalf("PromptAgent was not called within 3s")
|
||||
}
|
||||
deadline := time.Now().Add(500 * time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
if svc.messageQueue.GetStatus(ctx, sessionID).Count > 0 {
|
||||
break
|
||||
}
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
}
|
||||
|
||||
// Verify the workflow transitioned to Merge.
|
||||
updatedTask, err := repo.GetTask(ctx, taskID)
|
||||
if err != nil {
|
||||
t.Fatalf("load task: %v", err)
|
||||
}
|
||||
if updatedTask.WorkflowStepID != mergeStepID {
|
||||
t.Errorf("task step = %q, want %q", updatedTask.WorkflowStepID, mergeStepID)
|
||||
}
|
||||
|
||||
// The transient-failure path records the user msg AND queues the prompt
|
||||
// — both for the same Merge auto-start. This is the production data shape
|
||||
// (one chat row + one queue entry, both with workflow_step_name=Merge).
|
||||
queueStatus := svc.messageQueue.GetStatus(ctx, sessionID)
|
||||
if queueStatus.Count != 1 {
|
||||
t.Fatalf("expected exactly 1 queued message after transient failure, got %d", queueStatus.Count)
|
||||
}
|
||||
queued := queueStatus.Entries[0]
|
||||
if queued.QueuedBy != messagequeue.QueuedByWorkflow {
|
||||
t.Errorf("queued_by = %q, want %q", queued.QueuedBy, messagequeue.QueuedByWorkflow)
|
||||
}
|
||||
if name, _ := queued.Metadata["workflow_step_name"].(string); name != "Merge" {
|
||||
t.Errorf("queued metadata workflow_step_name = %q, want Merge", name)
|
||||
}
|
||||
mergeUserMsgs := 0
|
||||
for _, m := range msgCreator.userMessages {
|
||||
if name, _ := m.metadata["workflow_step_name"].(string); name == "Merge" {
|
||||
mergeUserMsgs++
|
||||
}
|
||||
}
|
||||
if mergeUserMsgs != 1 {
|
||||
t.Errorf("expected exactly 1 Merge user_message (from recordAutoStartMessage before PromptTask failed), got %d", mergeUserMsgs)
|
||||
}
|
||||
|
||||
// --- Phase 2: user resumes the session — boot_ready must drain the queue ---
|
||||
|
||||
// Clear the transient error so the drain's executeQueuedMessage → PromptTask
|
||||
// → PromptAgent succeeds. Otherwise the goroutine that runs after the queue
|
||||
// take would re-enter the same transient-retry path and re-queue the message
|
||||
// immediately, racing the queue-empty assertion below. Also swap in a fresh
|
||||
// promptDone channel to signal the second PromptAgent call (the original one
|
||||
// was already closed by Phase 1).
|
||||
secondPromptCalled := make(chan struct{})
|
||||
agentMgr.mu.Lock()
|
||||
agentMgr.promptErr = nil
|
||||
agentMgr.promptDone = secondPromptCalled
|
||||
// Reset capturedPrompts' bookkeeping so the channel re-fires; the mock's
|
||||
// `first := len(m.capturedPrompts) == 0` guard would otherwise skip the close.
|
||||
agentMgr.capturedPrompts = agentMgr.capturedPrompts[:0]
|
||||
agentMgr.capturedPromptCalls = agentMgr.capturedPromptCalls[:0]
|
||||
agentMgr.mu.Unlock()
|
||||
|
||||
// Simulate the resume: agentctl's ACP session has re-initialized so the
|
||||
// lifecycle manager fires events.AgentBootReady. Without the drain fix,
|
||||
// the queue stays orphaned. With the fix, it gets drained.
|
||||
svc.handleAgentBootReady(ctx, watcher.AgentEventData{
|
||||
TaskID: taskID,
|
||||
SessionID: sessionID,
|
||||
})
|
||||
|
||||
// drainQueuedMessageAfterTransition pops the queue synchronously and fires
|
||||
// `go executeQueuedMessage(...)`. The goroutine creates a user message and
|
||||
// calls PromptTask → PromptAgent. The mock closes secondPromptCalled on
|
||||
// that call, so the assertion below has a deterministic sync point.
|
||||
select {
|
||||
case <-secondPromptCalled:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatalf("boot_ready drain did not reach PromptAgent within 3s")
|
||||
}
|
||||
|
||||
if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 {
|
||||
t.Errorf("queue not drained after boot_ready: %d messages still queued (the orphaned-queue bug is back)", got)
|
||||
}
|
||||
}
|
||||
@@ -601,3 +601,174 @@ func TestHandleAgentBootReady_DoesNotTriggerOnTurnComplete(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandleAgentBootReady_DrainsOrphanedQueuedMessage reproduces the
|
||||
// production stuck-queue symptom: a workflow auto-start prompt is queued
|
||||
// against a session, the agent dies before the turn completes (so no
|
||||
// agent.ready fires to drain it), and the user resumes the session. The
|
||||
// session ends up WAITING_FOR_INPUT with the message still on the queue —
|
||||
// "1 queued" displayed in the UI forever — because handleAgentBootReady
|
||||
// only flipped state but never drained.
|
||||
//
|
||||
// After the fix, handleAgentBootReady takes the queued message and dispatches
|
||||
// it (via executeQueuedMessage in a goroutine). The test wires a full
|
||||
// executor + seeded executors_running row so the goroutine's PromptTask
|
||||
// call lands on a working code path instead of nil-derefing s.executor
|
||||
// under -race.
|
||||
func TestHandleAgentBootReady_DrainsOrphanedQueuedMessage(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
startSt models.TaskSessionState
|
||||
}{
|
||||
{"STARTING -> WAITING_FOR_INPUT (boot completes resume)", models.TaskSessionStateStarting},
|
||||
{"already WAITING_FOR_INPUT (boot raced persistResumeState)", models.TaskSessionStateWaitingForInput},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workspace: %v", err)
|
||||
}
|
||||
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workflow: %v", err)
|
||||
}
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: "task-1", WorkflowID: "wf1", WorkflowStepID: "step-merge",
|
||||
Title: "T", Description: "D", State: v1.TaskStateInProgress,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
|
||||
sessionID := "s1"
|
||||
const executionID = "exec-1"
|
||||
if err := repo.CreateTaskSession(ctx, &models.TaskSession{
|
||||
ID: sessionID, TaskID: "task-1", AgentProfileID: "profile-impl",
|
||||
AgentExecutionID: executionID,
|
||||
State: tc.startSt,
|
||||
IsPrimary: true,
|
||||
StartedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create session: %v", err)
|
||||
}
|
||||
// Seed executors_running so PromptTask -> ensureSessionRunning ->
|
||||
// executor.GetExecutionBySession finds the agent and skips resume.
|
||||
seedExecutorRunning(t, repo, sessionID, "task-1", executionID)
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
taskRepo.tasks["task-1"] = &v1.Task{
|
||||
ID: "task-1", WorkflowID: "wf1", State: v1.TaskStateInProgress,
|
||||
}
|
||||
|
||||
log := testLogger()
|
||||
agentMgr := &mockAgentManager{
|
||||
repoForExecutionLookup: repo,
|
||||
isAgentRunning: true, // satisfy GetExecutionBySession's IsAgentRunningForSession check
|
||||
}
|
||||
svc := &Service{
|
||||
logger: log,
|
||||
repo: repo,
|
||||
taskRepo: taskRepo,
|
||||
agentManager: agentMgr,
|
||||
messageQueue: messagequeue.NewServiceMemory(log),
|
||||
// Wire a real executor so the executeQueuedMessage goroutine
|
||||
// spawned by drainQueuedMessageAfterTransition can safely call
|
||||
// PromptTask -> executor.GetExecutionBySession without nil-derefing.
|
||||
executor: executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}),
|
||||
}
|
||||
|
||||
// Seed an orphaned workflow auto-start prompt — what the production
|
||||
// bug looked like in the DB at task 9378f7cf.
|
||||
if _, err := svc.messageQueue.QueueMessage(
|
||||
ctx, sessionID, "task-1", "ROLE: Merge operator. ...", "",
|
||||
messagequeue.QueuedByWorkflow, false, nil,
|
||||
); err != nil {
|
||||
t.Fatalf("queue orphaned prompt: %v", err)
|
||||
}
|
||||
if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 1 {
|
||||
t.Fatalf("precondition: queue count = %d, want 1", got)
|
||||
}
|
||||
|
||||
svc.handleAgentBootReady(ctx, watcher.AgentEventData{
|
||||
TaskID: "task-1", SessionID: sessionID,
|
||||
})
|
||||
|
||||
if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 {
|
||||
t.Errorf("queue count after boot ready = %d, want 0 (orphaned message must be drained)", got)
|
||||
}
|
||||
|
||||
// Session must still end up WAITING_FOR_INPUT — the drain is
|
||||
// additive, not a replacement for the existing flip behavior.
|
||||
finalSess, err := repo.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("load session: %v", err)
|
||||
}
|
||||
if finalSess.State != models.TaskSessionStateWaitingForInput {
|
||||
t.Errorf("session.State = %q, want WAITING_FOR_INPUT", finalSess.State)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandleAgentBootReady_DoesNotDrainForTerminalSession guards against
|
||||
// reviving a queued message on a session that was cancelled or completed —
|
||||
// the user explicitly stopped this session, the queued prompt should NOT be
|
||||
// dispatched, and the early-return for terminal states must continue to
|
||||
// short-circuit before the drain.
|
||||
func TestHandleAgentBootReady_DoesNotDrainForTerminalSession(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
|
||||
repo := setupTestRepo(t)
|
||||
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workspace: %v", err)
|
||||
}
|
||||
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now}); err != nil {
|
||||
t.Fatalf("create workflow: %v", err)
|
||||
}
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: "task-1", WorkflowID: "wf1", WorkflowStepID: "step-merge",
|
||||
Title: "T", Description: "D", State: v1.TaskStateInProgress,
|
||||
CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
|
||||
sessionID := "s1"
|
||||
if err := repo.CreateTaskSession(ctx, &models.TaskSession{
|
||||
ID: sessionID, TaskID: "task-1", AgentProfileID: "profile-impl",
|
||||
State: models.TaskSessionStateCancelled,
|
||||
IsPrimary: true,
|
||||
StartedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create session: %v", err)
|
||||
}
|
||||
|
||||
log := testLogger()
|
||||
svc := &Service{
|
||||
logger: log,
|
||||
repo: repo,
|
||||
taskRepo: newMockTaskRepo(),
|
||||
agentManager: &mockAgentManager{repoForExecutionLookup: repo},
|
||||
messageQueue: messagequeue.NewServiceMemory(log),
|
||||
}
|
||||
|
||||
if _, err := svc.messageQueue.QueueMessage(
|
||||
ctx, sessionID, "task-1", "stuck prompt", "",
|
||||
messagequeue.QueuedByWorkflow, false, nil,
|
||||
); err != nil {
|
||||
t.Fatalf("queue prompt: %v", err)
|
||||
}
|
||||
|
||||
svc.handleAgentBootReady(ctx, watcher.AgentEventData{
|
||||
TaskID: "task-1", SessionID: sessionID,
|
||||
})
|
||||
|
||||
if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 1 {
|
||||
t.Errorf("queue count after boot ready on terminal session = %d, want 1 (must not drain)", got)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user