From 96871ccdea74bbf6e174fd4a6e98b3387c90ab9b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Mon, 1 Jun 2026 11:42:05 +0100 Subject: [PATCH] fix(orchestrator): auto-resume dormant sessions after workflow queue (#1163) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(orchestrator): auto-resume dormant sessions after workflow queue When a workflow auto-start prompt is queued after a transient ACP stream disconnect, no live agent exists to drain the queue via agent.ready. The prompt sits orphaned until the user manually resumes the session. Add scheduleAutoResumeForWorkflowQueue (called from queueAutoStartPrompt) which fires tryEnsureExecution in the background when GetExecutionBySession shows no live agent. This drives ResumeSession → agent.boot_ready → handleAgentBootReady → drainQueuedMessageAfterTransition — the same path a manual resume would trigger. Guard with s.executor == nil so existing tests that don't wire the executor are unaffected. Add isAgentRunningFn override to mockAgentManager for mid-sequence state transitions, and add a regression test that exercises the full chain end-to-end. Co-Authored-By: Claude Sonnet 4.6 * docs(orchestrator): clarify scheduleAutoResumeForWorkflowQueue boundary condition --------- Co-authored-by: Kandev Agent Co-authored-by: Claude Sonnet 4.6 --- ...event_handlers_duplicate_autostart_test.go | 265 ++++++++++++++++++ .../orchestrator/event_handlers_test.go | 13 +- .../orchestrator/event_handlers_workflow.go | 22 ++ 3 files changed, 297 insertions(+), 3 deletions(-) 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