diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 2a65e88b90..afe0028fbc 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -594,6 +594,7 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session) return } + s.markIdleAfterReset(ctx, taskID, sessionID, session, step, isPassthrough) } hasAutoStart := false @@ -836,6 +837,43 @@ func (s *Service) queueAutoStartPrompt( return nil } +// 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, +// which loads the session before the turn finishes — the in-memory pointer +// still reads RUNNING even though the agent is now idle. Without this flip, +// queueAutoStartPromptIfRunning queues the message and PromptTask later +// rejects the drained queued send because the DB row also still reads RUNNING. +// +// Skip the flip when: +// - state was not RUNNING/STARTING (e.g. CREATED, where resetAgentContext +// early-returns true without restarting and autoStartStepPrompt routes +// the prompt through StartCreatedSession); +// - the session is passthrough AND auto_start_agent will write to PTY stdin +// next (the agent is actively processing, not idle). +// +// Uses updateTaskSessionState directly rather than setSessionWaitingForInput +// because the helper would also flip the task to TaskStateReview, which would +// be wrong here — auto_start_agent runs next and should leave the task as +// IN_PROGRESS. +func (s *Service) markIdleAfterReset( + ctx context.Context, + taskID, sessionID string, + session *models.TaskSession, + step *wfmodels.WorkflowStep, + isPassthrough bool, +) { + if session.State != models.TaskSessionStateRunning && + session.State != models.TaskSessionStateStarting { + return + } + if isPassthrough && step.HasOnEnterAction(wfmodels.OnEnterAutoStartAgent) { + return + } + s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false, session) + session.State = models.TaskSessionStateWaitingForInput +} + // resetAgentContext restarts the agent subprocess with a fresh ACP session, clearing // the agent's conversation context. The workspace environment is preserved. func (s *Service) resetAgentContext(ctx context.Context, taskID string, session *models.TaskSession, stepName string) bool { 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 4c72371a03..ef5be861b1 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go @@ -729,4 +729,101 @@ func TestProcessOnEnterResetAgentContext(t *testing.T) { t.Fatalf("expected session state %q, got %q", models.TaskSessionStateWaitingForInput, updated.State) } }) + + // Regression: on_turn_complete auto-transition enters processOnEnter with an + // in-memory session still marked RUNNING (loaded at the start of + // handleAgentReady before the turn finished). After reset_agent_context we + // must flip the session to WAITING_FOR_INPUT — otherwise a following + // auto_start_agent hits queueAutoStartPromptIfRunning (sees State=RUNNING) + // and queues the prompt instead of sending it, and PromptTask rejects the + // drained queued message for the same reason. The agent then sits idle + // until the user cancels. Both the in-memory pointer and the DB must be + // updated. + t.Run("reset_agent_context flips session to WAITING_FOR_INPUT for a running session", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") // seeds session.State = RUNNING + + session, _ := repo.GetTaskSession(ctx, "s1") + session.AgentExecutionID = "exec-abc" + session.Metadata = map[string]interface{}{"acp_session_id": "old-acp"} + _ = repo.UpdateTaskSession(ctx, session) + + agentMgr := &mockAgentManager{} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + step := &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Review Step", + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterResetAgentContext}, + }, + }, + } + + session, _ = repo.GetTaskSession(ctx, "s1") + if session.State != models.TaskSessionStateRunning { + t.Fatalf("precondition: seed should start session as RUNNING, got %q", session.State) + } + + svc.processOnEnter(ctx, "t1", session, step, "review task") + + // In-memory session must also be updated — queueAutoStartPromptIfRunning + // (further down in processOnEnter when auto_start_agent is present) + // checks the pointer, not the DB. + if session.State != models.TaskSessionStateWaitingForInput { + t.Errorf("in-memory session.State: want %q, got %q", + models.TaskSessionStateWaitingForInput, session.State) + } + + updated, _ := repo.GetTaskSession(ctx, "s1") + if updated.State != models.TaskSessionStateWaitingForInput { + t.Errorf("DB session.State: want %q, got %q", + models.TaskSessionStateWaitingForInput, updated.State) + } + + if len(agentMgr.restartProcessCalls) != 1 { + t.Errorf("expected 1 RestartAgentProcess call, got %d", len(agentMgr.restartProcessCalls)) + } + }) + + // Regression: passthrough sessions handle auto_start_agent by writing to + // PTY stdin via autoStartPassthroughPrompt — the agent is actively + // processing immediately after, not idle. We must NOT flip the session to + // WAITING_FOR_INPUT, otherwise WS subscribers see "ready for input" while + // the agent is mid-prompt. + t.Run("reset_agent_context preserves running state for passthrough+auto_start", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") // RUNNING + + session, _ := repo.GetTaskSession(ctx, "s1") + session.AgentExecutionID = "exec-pt" + _ = repo.UpdateTaskSession(ctx, session) + + agentMgr := &mockAgentManager{isPassthrough: true} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + + step := &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Review Step", + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterResetAgentContext}, + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + session, _ = repo.GetTaskSession(ctx, "s1") + svc.processOnEnter(ctx, "t1", session, step, "review task") + + // Positive assertion: pin the expected state to RUNNING so any other + // unintended mutation (COMPLETED, FAILED, etc.) also fails the test, + // not just an erroneous flip to WAITING_FOR_INPUT. + if session.State != models.TaskSessionStateRunning { + t.Errorf("in-memory session.State should remain RUNNING for passthrough+auto_start, got %q", session.State) + } + updated, _ := repo.GetTaskSession(ctx, "s1") + if updated.State != models.TaskSessionStateRunning { + t.Errorf("DB session.State should remain RUNNING for passthrough+auto_start, got %q", updated.State) + } + }) } diff --git a/apps/backend/internal/orchestrator/workflow_e2e_test.go b/apps/backend/internal/orchestrator/workflow_e2e_test.go index ea233f3e44..e97a235b23 100644 --- a/apps/backend/internal/orchestrator/workflow_e2e_test.go +++ b/apps/backend/internal/orchestrator/workflow_e2e_test.go @@ -7,8 +7,10 @@ import ( "fmt" "testing" + "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" "github.com/kandev/kandev/internal/workflow/engine" wfmodels "github.com/kandev/kandev/internal/workflow/models" ) @@ -112,15 +114,22 @@ var workflowTestCases = []workflowTestCase{ // Agent finishes at Backlog → In Progress (auto_start queues) {Trigger: engine.TriggerOnTurnComplete, ExpectStep: "In Progress", ExpectTransitioned: true, ExpectQueued: true, ExpectResets: 0}, - // Agent finishes at In Progress → New Context (reset + auto_start) + // Agent finishes at In Progress → New Context (reset + auto_start). + // After reset_agent_context, processOnEnter flips session state to + // WAITING_FOR_INPUT so auto_start sends directly instead of queueing + // against a stale RUNNING state. PromptTask then attempts the send — + // in this mock environment the executor has no running record for + // the session, so the send fails and autoStartStepPrompt's error + // path leaves the session at WAITING_FOR_INPUT. Crucially, the + // message is NOT sitting in the queue. {Trigger: engine.TriggerOnTurnComplete, SetRunning: true, ExpectStep: "New Context", - ExpectTransitioned: true, ExpectQueued: true, ExpectResets: 1}, + ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 1}, // Agent starts at New Context → on_turn_start → back to In Progress {Trigger: engine.TriggerOnTurnStart, SetRunning: true, ExpectStep: "In Progress", ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 1}, - // Agent finishes at In Progress → New Context again + // Agent finishes at In Progress → New Context again (same reset + auto_start path) {Trigger: engine.TriggerOnTurnComplete, SetRunning: true, ExpectStep: "New Context", - ExpectTransitioned: true, ExpectQueued: true, ExpectResets: 2}, + ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 2}, // Agent finishes at New Context → New Step (reset, no auto_start) {Trigger: engine.TriggerOnTurnComplete, SetRunning: true, ExpectStep: "New Step", ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 3}, @@ -283,7 +292,7 @@ func buildWorkflowFromJSON(t *testing.T, jsonStr string) (*mockStepGetter, map[s } // createEngineService creates a Service with the workflow engine initialized. -func createEngineService(t *testing.T, repo sessionExecutorStore, sg *mockStepGetter, agentMgr *mockAgentManager) *Service { +func createEngineService(t *testing.T, repo *sqliterepo.Repository, sg *mockStepGetter, agentMgr *mockAgentManager) *Service { t.Helper() log := testLogger() svc := &Service{ @@ -292,6 +301,7 @@ func createEngineService(t *testing.T, repo sessionExecutorStore, sg *mockStepGe taskRepo: newMockTaskRepo(), agentManager: agentMgr, messageQueue: messagequeue.NewService(log), + executor: executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}), } svc.SetWorkflowStepGetter(sg) return svc