diff --git a/apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go b/apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go index 7da85784cd..eacb0363a8 100644 --- a/apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go @@ -3,6 +3,7 @@ package orchestrator import ( "context" "errors" + "sync/atomic" "testing" "time" @@ -262,3 +263,267 @@ func TestAutoStartTransientError_BootReadyDrainsOrphanedQueue(t *testing.T) { t.Errorf("expected exactly 1 Merge user_message after boot_ready drain, got %d (duplicate-prompt bug is back)", mergeUserMsgsAfterDrain) } } + +// TestAutoStartTransientError_AutoResumesWhenAgentDead is the regression test +// for the silent-idle production symptom (task 648ca65e: 81-minute stuck queue). +// +// When PromptAgent returns a transient disconnect error, autoStartStepPrompt +// queues the Merge prompt with queued_by=workflow. Before this fix, the queue +// sat orphaned forever — there was no live agent process to fire agent.ready +// (which is what drains the queue normally), and no code re-launched the agent. +// +// scheduleAutoResumeForWorkflowQueue (added in queueAutoStartPrompt) fixes this: +// it fires tryEnsureExecution in the background when the agent is dead, driving +// ResumeSession → agent.boot_ready → handleAgentBootReady → drain. This test +// asserts that full chain end-to-end. +func TestAutoStartTransientError_AutoResumesWhenAgentDead(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-ar-1" + sessionID = "sess-ar-1" + executionID = "exec-ar-1" + fixupStepID = "step-ar-fixup" + mergeStepID = "step-ar-merge" + profile = "profile-ar-impl" + taskWorkflow = "wf-ar-1" + ) + + repo := setupTestRepo(t) + now := time.Now().UTC() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-ar", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: taskWorkflow, WorkspaceID: "ws-ar", 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, + 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, + } + + // firstPromptCalled closes on the first PromptAgent call (the transient failure). + // launchCalled closes when auto-resume fires LaunchAgent. + // secondPromptCalled closes on the second PromptAgent call (queue drain after resume). + firstPromptCalled := make(chan struct{}) + launchCalled := make(chan struct{}) + secondPromptCalled := make(chan struct{}) + // resumeDone is closed after handleAgentBootReady writes WAITING_FOR_INPUT and + // waitForSessionReady has had one full poll cycle (500ms) to observe it — this + // ensures the tryEnsureExecution goroutine exits before goleak runs. + resumeDone := make(chan struct{}) + var agentResumed atomic.Bool + + // var svc declared here so the launchAgentFunc closure can capture it. By the + // time LaunchAgent is called, svc is already constructed below. + var svc *Service + + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: true, + promptErr: errors.New("agent stream disconnected mid-prompt"), + promptDone: firstPromptCalled, + } + + // isAgentRunningFn models the production state transition: + // alive before the first PromptAgent → dead after the stream disconnect → + // alive again after LaunchAgent resumes the agent. + agentMgr.isAgentRunningFn = func(_ context.Context, _ string) bool { + if agentResumed.Load() { + return true + } + select { + case <-firstPromptCalled: + return false // agent died after the first prompt attempt + default: + return true // agent alive before the first prompt + } + } + + // launchAgentFunc simulates a successful agent re-launch. It: + // 1. Marks the agent as running (so subsequent IsAgentRunningForSession calls + // return true and PromptTask's ensureSessionRunning is a no-op). + // 2. Signals launchCalled so the test can assert the resume path was taken. + // 3. Spawns a goroutine that polls for STARTING (written by persistResumeState + // immediately after LaunchAgent returns), then calls handleAgentBootReady — + // mirroring the production AgentBootReady lifecycle event — which sets the + // session to WAITING_FOR_INPUT (unblocking waitForSessionReady) and drains + // the queue. + // 4. Sleeps 600ms after handleAgentBootReady returns so that waitForSessionReady + // (which polls every 500ms) has at least one full cycle to pick up + // WAITING_FOR_INPUT and exit — preventing a goroutine leak caught by goleak. + agentMgr.launchAgentFunc = func(lctx context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentResumed.Store(true) // mark alive before anything polls for it + close(launchCalled) + go func() { + defer close(resumeDone) + tick := time.NewTicker(5 * time.Millisecond) + defer tick.Stop() + timeout := time.After(5 * time.Second) + for { + select { + case <-tick.C: + sess, err := repo.GetTaskSession(context.Background(), sessionID) + if err != nil || sess == nil || sess.State != models.TaskSessionStateStarting { + continue + } + // Clear promptErr and swap in the second channel before handleAgentBootReady + // triggers the drain — executeQueuedMessage must see nil error. + agentMgr.mu.Lock() + agentMgr.promptErr = nil + agentMgr.promptDone = secondPromptCalled + agentMgr.capturedPrompts = agentMgr.capturedPrompts[:0] + agentMgr.capturedPromptCalls = agentMgr.capturedPromptCalls[:0] + agentMgr.mu.Unlock() + svc.handleAgentBootReady(context.Background(), watcher.AgentEventData{ + TaskID: req.TaskID, + SessionID: req.SessionID, + }) + // handleAgentBootReady wrote WAITING_FOR_INPUT to DB. + // waitForSessionReady polls every 500ms — sleep one full cycle so + // it exits before goleak checks for leaked goroutines. + time.Sleep(600 * time.Millisecond) + return + case <-timeout: + return + } + } + }() + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-ar-resumed"}, nil + } + + exec := executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + msgCreator := &mockMessageCreator{} + 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 → transient failure → workflow prompt queued --- + // handleAgentReady fires the Fixup on_turn_complete → transitions to Merge → + // processOnEnter(Merge) → autoStartStepPrompt → PromptAgent returns transient + // error → queueAutoStartPrompt → scheduleAutoResumeForWorkflowQueue fires. + + 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") + } + + select { + case <-firstPromptCalled: + case <-time.After(3 * time.Second): + t.Fatalf("PromptAgent was not called within 3s") + } + // Poll for the queue entry (tiny gap between PromptAgent returning and the + // queue write completing — both are synchronous in the processOnEnter goroutine). + 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) + } + if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 1 { + t.Fatalf("expected 1 queued message after transient failure, got %d", got) + } + queued := svc.messageQueue.GetStatus(ctx, sessionID).Entries[0] + if queued.QueuedBy != messagequeue.QueuedByWorkflow { + t.Errorf("queued_by = %q, want %q", queued.QueuedBy, messagequeue.QueuedByWorkflow) + } + + // --- Phase 2: auto-resume --- + // scheduleAutoResumeForWorkflowQueue (called from queueAutoStartPrompt) fires + // tryEnsureExecution in the background. Assert LaunchAgent was called, then + // wait for the drain to reach the second PromptAgent call. + + select { + case <-launchCalled: + case <-time.After(5 * time.Second): + t.Fatalf("scheduleAutoResumeForWorkflowQueue did not trigger LaunchAgent within 5s (auto-resume bug?)") + } + + // Wait for secondPromptCalled (drain reached PromptAgent) and resumeDone + // (waitForSessionReady exited) concurrently — both must fire before we assert. + select { + case <-secondPromptCalled: + case <-time.After(5 * time.Second): + t.Fatalf("queue was not drained after auto-resume (no second PromptAgent within 5s)") + } + select { + case <-resumeDone: + case <-time.After(3 * time.Second): + t.Fatalf("tryEnsureExecution goroutine did not complete within 3s") + } + + if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 { + t.Errorf("queue not empty after auto-resume drain: %d messages remain", got) + } + + // Exactly one Merge user message must exist — no duplicate. + 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, got %d", mergeUserMsgs) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index b851bc6b90..99195fbaab 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -144,8 +144,12 @@ func (m *mockTaskRepo) UpdateTaskStateIfCurrentIn( // mockAgentManager is a minimal mock of executor.AgentManagerClient for testing. type mockAgentManager struct { - isPassthrough bool - isAgentRunning bool + isPassthrough bool + isAgentRunning bool + // isAgentRunningFn, when non-nil, overrides isAgentRunning for + // IsAgentRunningForSession. Lets tests model state changes mid-sequence + // (e.g. stream disconnect between PromptAgent call and queue write). + isAgentRunningFn func(context.Context, string) bool resolveProfileErr error restartProcessCalls []string // tracks execution IDs passed to RestartAgentProcess restartProcessErr error @@ -287,7 +291,10 @@ func (m *mockAgentManager) CancelAgent(_ context.Context, _ string) error { func (m *mockAgentManager) RespondToPermissionBySessionID(_ context.Context, _, _, _ string, _ bool) error { return nil } -func (m *mockAgentManager) IsAgentRunningForSession(_ context.Context, _ string) bool { +func (m *mockAgentManager) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { + if m.isAgentRunningFn != nil { + return m.isAgentRunningFn(ctx, sessionID) + } return m.isAgentRunning } func (m *mockAgentManager) ResolveAgentProfile(_ context.Context, _ string) (*executor.AgentProfileInfo, error) { diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 716d27bd74..76b04ffbf3 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -1514,9 +1514,31 @@ func (s *Service) queueAutoStartPrompt( return fmt.Errorf("failed to queue workflow auto-start prompt: %w", err) } s.publishQueueStatusEvent(ctx, sessionID) + s.scheduleAutoResumeForWorkflowQueue(ctx, sessionID) return nil } +// scheduleAutoResumeForWorkflowQueue kicks off a background resume when a +// workflow auto-start prompt was just queued but no live agent process exists +// to drain it. No-op when the agent is already running — handleAgentReady +// will drain on the next turn end. Uses the same tryEnsureExecution path as +// EnsureSession (office panels), which drives ResumeSession → agent.boot_ready +// → handleAgentBootReady → drainQueuedMessageAfterTransition. +// +// Covers the case where the execution is dead at queue time (e.g. agent +// crashed just before the on_enter transition). If the agent is alive when +// the queue is written but dies later, the queue is drained by the next +// handleAgentBootReady (manual or automatic resume). +func (s *Service) scheduleAutoResumeForWorkflowQueue(ctx context.Context, sessionID string) { + if s.executor == nil { + return + } + if exec, ok := s.executor.GetExecutionBySession(sessionID); ok && exec != nil { + return + } + go s.tryEnsureExecution(context.WithoutCancel(ctx), sessionID) +} + // 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