fix: user workflow deadlock (#677)

apps/backend/internal/github/service_test.go

@@ -776,4 +776,3 @@ func TestTimeEqual(t *testing.T) {
 		})
 	}
 }
-

apps/backend/internal/orchestrator/event_handlers_agent.go

@@ -149,9 +149,20 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 	// Check for workflow transition based on session's current step.
 	// Uses the engine when available; falls back to legacy evaluation.
 	// The ViaEngine method handles setSessionWaitingForInput internally when no transition occurs.
-	s.processOnTurnCompleteViaEngine(ctx, data.TaskID, session)
+	transitioned := s.processOnTurnCompleteViaEngine(ctx, data.TaskID, session)
 
-	// ALWAYS check for queued messages after agent becomes ready, regardless of workflow transition
+	// When a workflow transition occurred (e.g. Work → Review), the new step's
+	// on_enter actions handle the next prompt (auto_start_agent launches a goroutine).
+	// Skip the queued-message check to avoid racing with that auto-start goroutine —
+	// both would try to call PromptTask and the loser's queued message would be lost.
+	if transitioned {
+		s.logger.Debug("workflow transition occurred, skipping queued message check",
+			zap.String("task_id", data.TaskID),
+			zap.String("session_id", data.SessionID))
+		return
+	}
+
+	// Check for queued messages when no workflow transition occurred.
 	queueStatus := s.messageQueue.GetStatus(ctx, data.SessionID)
 	s.logger.Info("checking for queued messages",
 		zap.String("session_id", data.SessionID),

apps/backend/internal/orchestrator/event_handlers_github_issue_test.go

@@ -117,7 +117,6 @@ func setupIssueTaskTest(t *testing.T) (*Service, *mockStepGetter) {
 	return createTestService(repo, stepGetter, newMockTaskRepo()), stepGetter
 }
 
-
 func TestCreateIssueTask_SkipsWhenAlreadyReserved(t *testing.T) {
 	svc, _ := setupIssueTaskTest(t)
 	ghSvc := &mockGitHubService{issueReserveReturn: false}
@@ -185,4 +184,4 @@ func TestCreateIssueTask_ReleasesOnTaskCreationFailure(t *testing.T) {
 	if ghSvc.issueAssignCalls != 0 {
 		t.Errorf("expected no Assign call after failure, got %d", ghSvc.issueAssignCalls)
 	}
-}
+}

apps/backend/internal/orchestrator/event_handlers_github_test.go

@@ -674,4 +674,3 @@ func TestListTasksNeedingPRWatch(t *testing.T) {
 		}
 	})
 }
-

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -511,7 +511,9 @@ func (s *Service) switchSessionForStep(ctx context.Context, taskID string, curre
 	}
 
 	// Promote the new session to primary so it's loaded when navigating back to this task.
-	if err := s.repo.SetSessionPrimary(ctx, newSession.ID); err != nil {
+	// Use SetPrimarySession (not repo.SetSessionPrimary) to broadcast a task.updated WS
+	// event — the frontend reads primarySessionId from the task to render the star icon.
+	if err := s.SetPrimarySession(ctx, newSession.ID); err != nil {
 		s.logger.Warn("failed to set new session as primary",
 			zap.String("session_id", newSession.ID), zap.Error(err))
 	}
@@ -526,8 +528,14 @@ func (s *Service) switchSessionForStep(ctx context.Context, taskID string, curre
 	}
 
 	// Mark the current session as completed.
-	currentSession.State = models.TaskSessionStateCompleted
+	// Use updateTaskSessionState to publish a session.state_changed WS event so the
+	// frontend learns the old session is terminal and can adopt the new one.
+	s.updateTaskSessionState(ctx, taskID, currentSession.ID, models.TaskSessionStateCompleted, "", false)
+	// Also set CompletedAt timestamp for bookkeeping (updateTaskSessionState only sets state).
+	// Clear IsPrimary to avoid overwriting the SetPrimarySession call above when writing back.
 	now := time.Now().UTC()
+	currentSession.State = models.TaskSessionStateCompleted
+	currentSession.IsPrimary = false
 	currentSession.CompletedAt = &now
 	currentSession.UpdatedAt = now
 	if err := s.repo.UpdateTaskSession(ctx, currentSession); err != nil {
@@ -542,6 +550,9 @@ func (s *Service) switchSessionForStep(ctx context.Context, taskID string, curre
 
 // maybySwitchSessionForProfile checks whether the step requires a different agent profile
 // and switches the session if so. Passthrough sessions are returned unchanged.
+// When the step has no agent_profile override, falls back to the task's original
+// agent profile (from task metadata) so that moving to a "plain" step reverts
+// to the default agent instead of keeping the previous step's override.
 // Returns the effective session (new or original) and whether processing should continue.
 // A false return means the switch failed; the caller should return immediately.
 func (s *Service) maybySwitchSessionForProfile(
@@ -551,6 +562,18 @@ func (s *Service) maybySwitchSessionForProfile(
 		return session, true
 	}
 	effectiveProfile := s.resolveStepAgentProfile(ctx, step)
+
+	// When the step has no override, fall back to the task's original agent profile
+	// so that moving to a step without an agent_profile reverts to the default.
+	if effectiveProfile == "" {
+		task, err := s.repo.GetTask(ctx, taskID)
+		if err == nil && task.Metadata != nil {
+			if pid, ok := task.Metadata[models.MetaKeyAgentProfileID].(string); ok && pid != "" {
+				effectiveProfile = pid
+			}
+		}
+	}
+
 	if effectiveProfile == "" || effectiveProfile == session.AgentProfileID {
 		return session, true
 	}
@@ -629,18 +652,18 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
 
 	case hasAutoStart:
 		// ACP path: build prompt from step configuration.
-		// Run auto-start inline so queue state is visible before handleAgentReady
-		// checks for queued messages.
+		// When called from applyEngineTransition (on_turn_complete), processOnEnter
+		// runs in a goroutine and the session is already WAITING_FOR_INPUT, so
+		// autoStartStepPrompt sends the prompt directly via PromptTask.
 		effectivePrompt := s.buildWorkflowPrompt(taskDescription, step, taskID, sessionID)
-		planMode := hasPlanMode
-		err := s.autoStartStepPrompt(ctx, taskID, session, step.Name, effectivePrompt, planMode, true)
-		if err != nil {
+		if err := s.autoStartStepPrompt(ctx, taskID, session, step.Name, effectivePrompt, hasPlanMode, true); err != nil {
 			s.logger.Error("failed to auto-start agent for step",
 				zap.String("task_id", taskID),
 				zap.String("session_id", sessionID),
 				zap.String("step_name", step.Name),
 				zap.Error(err))
 			s.setSessionWaitingForInput(ctx, taskID, sessionID, session)
+			s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session)
 		}
 
 	default:
@@ -650,19 +673,27 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
 		if sessionSwitched && step.Prompt != "" {
 			effectivePrompt := s.buildWorkflowPrompt(taskDescription, step, taskID, sessionID)
 			planMode := hasPlanMode
+			stepName := step.Name
+			stepID := step.ID
 			s.logger.Info("auto-launching agent after profile switch (no explicit auto_start)",
 				zap.String("task_id", taskID),
 				zap.String("session_id", sessionID),
-				zap.String("step_name", step.Name))
-			err := s.autoStartStepPrompt(ctx, taskID, session, step.Name, effectivePrompt, planMode, true)
-			if err != nil {
-				s.logger.Error("failed to launch agent after profile switch",
-					zap.String("task_id", taskID),
-					zap.String("session_id", sessionID),
-					zap.Error(err))
-				s.setSessionWaitingForInput(ctx, taskID, sessionID, session)
-				s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session)
-			}
+				zap.String("step_name", stepName))
+			// Launch asynchronously because processOnEnter may also be called
+			// synchronously from finalizeStepEnter (manual task move). In that path,
+			// autoStartStepPrompt would block the caller's goroutine.
+			go func() {
+				asyncCtx := context.WithoutCancel(ctx)
+				err := s.autoStartStepPrompt(asyncCtx, taskID, session, stepName, effectivePrompt, planMode, true)
+				if err != nil {
+					s.logger.Error("failed to launch agent after profile switch",
+						zap.String("task_id", taskID),
+						zap.String("session_id", sessionID),
+						zap.Error(err))
+					s.setSessionWaitingForInput(asyncCtx, taskID, sessionID, session)
+					s.publishSessionWaitingEvent(asyncCtx, taskID, sessionID, stepID, session)
+				}
+			}()
 			return
 		}
 		s.setSessionWaitingForInput(ctx, taskID, sessionID, session)
@@ -1232,7 +1263,24 @@ func (s *Service) applyEngineTransition(
 		return true
 	}
 
-	s.processOnEnter(ctx, taskID, session, targetStep, taskDescription)
+	// When triggered from on_turn_complete, the agent has finished its turn but
+	// handleAgentReady returns early without setting WAITING_FOR_INPUT (because the
+	// transition already occurred). The session is still RUNNING in the DB.
+	// Flip to WAITING_FOR_INPUT so that autoStartStepPrompt in processOnEnter sends
+	// the prompt directly instead of queueing it — the queue would never be drained
+	// because handleAgentReady already returned.
+	if session.State == models.TaskSessionStateRunning || session.State == models.TaskSessionStateStarting {
+		s.updateTaskSessionState(ctx, taskID, session.ID, models.TaskSessionStateWaitingForInput, "", false, session)
+		session.State = models.TaskSessionStateWaitingForInput
+	}
+
+	// Launch processOnEnter asynchronously to avoid blocking the stream reader goroutine.
+	// When triggered from on_turn_complete, the entire call chain runs in the WebSocket
+	// stream reader goroutine (G_reader). processOnEnter may call resetAgentContext →
+	// ResetAgentContext → sendStreamRequest, which blocks G_reader waiting for a response
+	// that can only be delivered by G_reader reading from the same WebSocket — a deadlock.
+	// The DB transition is already persisted above, so it's safe to process on_enter async.
+	go s.processOnEnter(context.WithoutCancel(ctx), taskID, session, targetStep, taskDescription)
 	return true
 }
 

apps/backend/internal/orchestrator/task_operations.go

@@ -273,6 +273,41 @@ func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, ag
 		return nil, fmt.Errorf("agent_profile_id is required")
 	}
 
+	// Override with workflow step's agent profile if one is configured.
+	effectiveProfileID = s.resolveEffectiveAgentProfile(ctx, taskID, "", effectiveProfileID)
+
+	// If the workflow step overrode the profile, update the session record in DB
+	// so the frontend tab displays the correct agent (it reads session.agent_profile_id).
+	if effectiveProfileID != session.AgentProfileID {
+		s.logger.Info("updating session agent profile for workflow step override",
+			zap.String("session_id", sessionID),
+			zap.String("old_profile", session.AgentProfileID),
+			zap.String("new_profile", effectiveProfileID))
+		session.AgentProfileID = effectiveProfileID
+		// Re-resolve the agent profile snapshot so the tab shows the correct agent logo/name.
+		// Set a minimal snapshot first so stale data is never persisted if resolution fails.
+		session.AgentProfileSnapshot = map[string]interface{}{"id": effectiveProfileID}
+		if profileInfo, err := s.agentManager.ResolveAgentProfile(ctx, effectiveProfileID); err != nil {
+			s.logger.Warn("failed to resolve agent profile snapshot for workflow step override",
+				zap.String("session_id", sessionID),
+				zap.String("profile_id", effectiveProfileID),
+				zap.Error(err))
+		} else if profileInfo != nil {
+			session.AgentProfileSnapshot = map[string]interface{}{
+				"id":         profileInfo.ProfileID,
+				"name":       profileInfo.ProfileName,
+				"agent_id":   profileInfo.AgentID,
+				"agent_name": profileInfo.AgentName,
+				"model":      profileInfo.Model,
+			}
+		}
+		if err := s.repo.UpdateTaskSession(ctx, session); err != nil {
+			s.logger.Warn("failed to update session agent profile",
+				zap.String("session_id", sessionID),
+				zap.Error(err))
+		}
+	}
+
 	// Transition task state: CREATED → SCHEDULING → (IN_PROGRESS via executor)
 	if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateScheduling); err != nil {
 		s.logger.Warn("failed to update task state to SCHEDULING",
@@ -391,6 +426,11 @@ func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID s
 
 	s.moveTaskToWorkflowStep(ctx, taskID, workflowStepID)
 
+	// Resolve the workflow step's agent profile override.
+	// The frontend may pass the workspace default profile, but the step may
+	// require a different agent (e.g., Codex on "In Progress", Auggie on "Review").
+	agentProfileID = s.resolveEffectiveAgentProfile(ctx, taskID, workflowStepID, agentProfileID)
+
 	// Fetch the task from the repository to get complete task info
 	task, err := s.scheduler.GetTask(ctx, taskID)
 	if err != nil {
@@ -471,6 +511,75 @@ func (s *Service) moveTaskToWorkflowStep(ctx context.Context, taskID, workflowSt
 	}
 }
 
+// resolveEffectiveAgentProfile checks whether the task's workflow step overrides
+// the agent profile. If the step (or workflow default) specifies a different
+// profile, that profile is returned instead of the caller-provided one.
+// This ensures the initial task start uses the step's agent — not just the
+// workspace default the frontend sends.
+func (s *Service) resolveEffectiveAgentProfile(ctx context.Context, taskID, workflowStepID, callerProfileID string) string {
+	if s.workflowStepGetter == nil {
+		s.logger.Debug("resolveEffectiveAgentProfile: no workflowStepGetter, using caller profile",
+			zap.String("task_id", taskID),
+			zap.String("caller_profile", callerProfileID))
+		return callerProfileID
+	}
+
+	// Determine the effective step ID: explicit param > task's current step.
+	effectiveStepID := workflowStepID
+	if effectiveStepID == "" {
+		dbTask, err := s.repo.GetTask(ctx, taskID)
+		if err != nil {
+			s.logger.Debug("resolveEffectiveAgentProfile: failed to load task from DB",
+				zap.String("task_id", taskID),
+				zap.Error(err))
+			return callerProfileID
+		}
+		s.logger.Debug("resolveEffectiveAgentProfile: loaded task from DB",
+			zap.String("task_id", taskID),
+			zap.String("db_workflow_step_id", dbTask.WorkflowStepID))
+		if dbTask.WorkflowStepID == "" {
+			s.logger.Debug("resolveEffectiveAgentProfile: task has no workflow step, using caller profile",
+				zap.String("task_id", taskID))
+			return callerProfileID
+		}
+		effectiveStepID = dbTask.WorkflowStepID
+	}
+
+	step, err := s.workflowStepGetter.GetStep(ctx, effectiveStepID)
+	if err != nil || step == nil {
+		s.logger.Debug("resolveEffectiveAgentProfile: failed to load step",
+			zap.String("task_id", taskID),
+			zap.String("step_id", effectiveStepID),
+			zap.Error(err))
+		return callerProfileID
+	}
+
+	s.logger.Debug("resolveEffectiveAgentProfile: loaded step",
+		zap.String("task_id", taskID),
+		zap.String("step_id", effectiveStepID),
+		zap.String("step_name", step.Name),
+		zap.String("step_agent_profile_id", step.AgentProfileID),
+		zap.String("step_workflow_id", step.WorkflowID))
+
+	stepProfile := s.resolveStepAgentProfile(ctx, step)
+	s.logger.Debug("resolveEffectiveAgentProfile: resolved step profile",
+		zap.String("task_id", taskID),
+		zap.String("step_profile", stepProfile),
+		zap.String("caller_profile", callerProfileID))
+
+	if stepProfile == "" || stepProfile == callerProfileID {
+		return callerProfileID
+	}
+
+	s.logger.Info("overriding agent profile with workflow step profile",
+		zap.String("task_id", taskID),
+		zap.String("step_id", effectiveStepID),
+		zap.String("step_name", step.Name),
+		zap.String("caller_profile", callerProfileID),
+		zap.String("step_profile", stepProfile))
+	return stepProfile
+}
+
 // postLaunchStart records the initial message and sets plan mode after a successful launch.
 func (s *Service) postLaunchStart(ctx context.Context, taskID string, execution *executor.TaskExecution, prompt string, recordPlanMode, setPlanMode bool, attachments []v1.MessageAttachment) {
 	if execution.SessionID != "" {

apps/backend/internal/orchestrator/workflow_e2e_test.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/kandev/kandev/internal/orchestrator/executor"
 	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
@@ -111,9 +112,13 @@ var workflowTestCases = []workflowTestCase{
 		WorkflowJSON: developmentWorkflowJSON,
 		StartStep:    "Backlog",
 		Events: []testEvent{
-			// Agent finishes at Backlog → In Progress (auto_start queues)
+			// Agent finishes at Backlog → In Progress (auto_start_agent on_enter).
+			// applyEngineTransition flips RUNNING → WAITING_FOR_INPUT before the
+			// async processOnEnter goroutine, so autoStartStepPrompt attempts a
+			// direct send instead of queueing. In this mock environment the send
+			// fails (no executor), so the message is NOT queued.
 			{Trigger: engine.TriggerOnTurnComplete, ExpectStep: "In Progress",
-				ExpectTransitioned: true, ExpectQueued: true, ExpectResets: 0},
+				ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 0},
 			// 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
@@ -244,6 +249,12 @@ func runSingleEvent(
 		t.Fatalf("unsupported trigger: %s", ev.Trigger)
 	}
 
+	// processOnEnter runs asynchronously (goroutine) after engine transitions.
+	// Give it time to complete before checking side effects.
+	if transitioned {
+		time.Sleep(100 * time.Millisecond)
+	}
+
 	if transitioned != ev.ExpectTransitioned {
 		t.Errorf("transitioned = %v, want %v", transitioned, ev.ExpectTransitioned)
 	}
@@ -364,13 +375,23 @@ func assertQueueState(t *testing.T, ctx context.Context, svc *Service, sessionID
 }
 
 // assertSessionState verifies the session's current state.
+// Polls briefly because processOnEnter runs asynchronously (goroutine) and
+// the state change may not be visible immediately after the trigger returns.
 func assertSessionState(t *testing.T, ctx context.Context, repo sessionExecutorStore, sessionID string, expect models.TaskSessionState) {
 	t.Helper()
-	session, err := repo.GetTaskSession(ctx, sessionID)
-	if err != nil {
-		t.Fatalf("failed to load session: %v", err)
-	}
-	if session.State != expect {
-		t.Errorf("session state = %q, want %q", session.State, expect)
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for {
+		session, err := repo.GetTaskSession(ctx, sessionID)
+		if err != nil {
+			t.Fatalf("failed to load session: %v", err)
+		}
+		if session.State == expect {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Errorf("session state = %q, want %q (after polling)", session.State, expect)
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
 	}
 }

apps/backend/internal/task/handlers/task_http_handlers.go

@@ -408,10 +408,11 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) {
 		return
 	}
 
-	// When not starting the agent immediately, persist profile IDs in task metadata
-	// so handleTaskMovedNoSession can retrieve them when the task is later dragged
-	// to a step with auto_start_agent.
-	if !body.StartAgent && !body.PrepareSession && body.AgentProfileID != "" {
+	// Always persist profile IDs in task metadata so they can be used as the
+	// task's "default" agent profile. This is needed both for deferred agent start
+	// (handleTaskMovedNoSession) and for reverting to the default agent when a
+	// workflow step has no agent_profile override.
+	if body.AgentProfileID != "" {
 		if body.Metadata == nil {
 			body.Metadata = make(map[string]interface{})
 		}

apps/backend/internal/task/handlers/task_ws_handlers.go

@@ -128,10 +128,11 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M
 		})
 	}
 
-	// When not starting the agent immediately, persist profile IDs in task metadata
-	// so handleTaskMovedNoSession can retrieve them when the task is later dragged
-	// to a step with auto_start_agent.
-	if !req.StartAgent && req.AgentProfileID != "" {
+	// Always persist profile IDs in task metadata so they can be used as the
+	// task's "default" agent profile. This is needed both for deferred agent start
+	// (handleTaskMovedNoSession) and for reverting to the default agent when a
+	// workflow step has no agent_profile override.
+	if req.AgentProfileID != "" {
 		if req.Metadata == nil {
 			req.Metadata = make(map[string]interface{})
 		}