fix: preserve workflow prompts after async startup failure (#3753) (#3768)

apps/backend/internal/orchestrator/dynamic_launch.go

@@ -313,11 +313,12 @@ func (s *Service) mirrorDynamicRouteProjection(
 // process start succeeds. LaunchPreparedSession returns before this point.
 func (s *Service) handleAgentProcessStarted(
 	ctx context.Context,
-	_, sessionID, agentExecutionID string,
+	taskID, sessionID, agentExecutionID string,
 ) {
 	if !s.ceilingCallbackOwnsSession(ctx, sessionID, agentExecutionID) {
 		return
 	}
+	s.retireWorkflowStartPromptAttempt(ctx, taskID, sessionID, agentExecutionID)
 	// AC-52's acceptance edge, composed first and unconditionally on
 	// sessionID alone (AC-56a): profileExecutionResolver is a dynamic-routing
 	// precondition, not a launch one, so an instance without it configured

apps/backend/internal/orchestrator/event_handlers_agent.go

@@ -1153,6 +1153,42 @@ func queuedMessagePromptContent(queuedMsg *messagequeue.QueuedMessage) string {
 	return appendStepHandoffToPrompt(content, stepHandoffFromQueuedMetadata(queuedMsg.Metadata))
 }
 
+func (s *Service) queuedMessageHasDispatchInput(ctx context.Context, queuedMsg *messagequeue.QueuedMessage) (bool, error) {
+	if queuedMsg == nil {
+		return false, nil
+	}
+	if present, ok := queuedMsg.Metadata[metaKeyWorkflowDispatchInputPresent].(bool); ok {
+		return present, nil
+	}
+	if strings.TrimSpace(queuedMessagePromptContent(queuedMsg)) != "" ||
+		len(queuedMsg.Attachments) > 0 || queuedMsg.PlanMode {
+		return true, nil
+	}
+	session, err := s.repo.GetTaskSession(ctx, queuedMsg.SessionID)
+	if err != nil {
+		if errors.Is(err, models.ErrTaskSessionNotFound) || errors.Is(err, context.Canceled) {
+			return false, nil
+		}
+		return false, err
+	}
+	if session == nil {
+		return false, nil
+	}
+	configMode, _ := session.Metadata["config_mode"].(bool)
+	return configMode, nil
+}
+
+func workflowQueuedConfigModeOverride(queuedMsg *messagequeue.QueuedMessage) *bool {
+	if queuedMsg == nil {
+		return nil
+	}
+	configMode, ok := queuedMsg.Metadata[metaKeyWorkflowConfigMode].(bool)
+	if !ok {
+		return nil
+	}
+	return &configMode
+}
+
 // prepareQueuedCIAutoFixOutcomeProtocol selects the protocol name from the
 // current session execution. It rewrites only the server-owned protocol block;
 // task prompt text and historical transcript content remain untouched.
@@ -1558,6 +1594,7 @@ func (s *Service) executeQueuedMessageWithReservation(
 			afterDispatch:        afterDispatch,
 			beforeDispatch:       beforeDispatch,
 			disableDispatchRetry: queuedMsg.IsDurablePlanComment(),
+			configModeOverride:   workflowQueuedConfigModeOverride(queuedMsg),
 			onAccepted: func(turnID string) {
 				s.bindQueuedCIAutoFixAttempt(promptCtx, queuedMsg, turnID)
 			},
@@ -3477,6 +3514,7 @@ func (s *Service) handleAgentStartFailed(ctx context.Context, taskID, sessionID,
 			}
 			return true
 		}
+		s.preserveWorkflowStartPromptAfterFailure(ctx, taskID, sessionID, agentExecutionID)
 	}
 	if failureData.FailureCode == string(routingerr.CodeManagedRuntimeNpmResolution) {
 		s.logger.Info("managed npm runtime startup failure is recoverable",

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -5336,7 +5336,20 @@ func (s *Service) dispatchTakenQueuedMessageForSession(
 		}
 	}
 	s.publishQueueStatusEventForIdentity(ctx, identity)
-	if queuedMsg.Content == "" && len(queuedMsg.Attachments) == 0 {
+	hasInput, inputErr := s.queuedMessageHasDispatchInput(ctx, queuedMsg)
+	if inputErr != nil {
+		s.logger.Warn("failed to inspect queued message input; retaining reservation",
+			zap.String("session_id", identity.SessionID),
+			zap.String("queue_id", queuedMsg.ID),
+			zap.Error(inputErr))
+		if identity.SessionIncarnationID != "" {
+			s.requeueMessageForReservation(ctx, &queuedDispatchReservation{identity: identity}, queuedMsg, queuedMsg.QueuedBy)
+		} else {
+			s.requeueMessage(ctx, queuedMsg, queuedMsg.QueuedBy)
+		}
+		return false
+	}
+	if !hasInput {
 		s.logger.Warn("skipping empty queued message after transition",
 			zap.String("session_id", identity.SessionID),
 			zap.String("queue_id", queuedMsg.ID))
@@ -5380,7 +5393,16 @@ func (s *Service) dispatchTakenQueuedMessage(
 		}
 	}
 	s.publishQueueStatusEvent(ctx, sessionID)
-	if queuedMsg.Content == "" && len(queuedMsg.Attachments) == 0 {
+	hasInput, inputErr := s.queuedMessageHasDispatchInput(ctx, queuedMsg)
+	if inputErr != nil {
+		s.logger.Warn("failed to inspect queued message input; retaining reservation",
+			zap.String("session_id", sessionID),
+			zap.String("queue_id", queuedMsg.ID),
+			zap.Error(inputErr))
+		s.requeueMessage(ctx, queuedMsg, queuedMsg.QueuedBy)
+		return false
+	}
+	if !hasInput {
 		s.logger.Warn("discarding empty queued message after transition",
 			zap.String("session_id", sessionID),
 			zap.String("queue_id", queuedMsg.ID))
@@ -5485,6 +5507,17 @@ func (s *Service) autoStartPassthroughPrompt(
 // failed transiently and the queue was later drained via boot_ready.
 const metaKeyUserMessageRecorded = "user_message_recorded"
 
+// metaKeyWorkflowDispatchInputPresent records that the original workflow
+// dispatch had effective input, including mode-generated instructions. It is
+// written with the raw queue form so a recovery drain does not need to infer
+// actionability from mutable session metadata.
+const metaKeyWorkflowDispatchInputPresent = "workflow_dispatch_input_present"
+
+// metaKeyWorkflowConfigMode preserves the mode that contributed effective
+// input to a deferred workflow launch. The queue keeps the prompt raw and the
+// mode as metadata so recovery can apply the transform exactly once.
+const metaKeyWorkflowConfigMode = "workflow_config_mode"
+
 // MetaKeyTurnStartAlreadyProcessed marks a queued prompt whose on_turn_start
 // hook already ran synchronously before queuing (see
 // task/handlers.queuePromptIfRuntimeUnavailable, which queues after
@@ -5757,6 +5790,12 @@ func (s *Service) autoStartStepPrompt(
 			}, referenceContext, pullRequestTargetContext)
 		}
 	}
+	// Use the fully composed forms after references, handoff, and first-turn
+	// context have been applied. This captures mode-generated input for an
+	// empty raw prompt while the queue still receives the raw prompt plus mode
+	// metadata, avoiding a second system block during recovery.
+	dispatchInputPresent := strings.TrimSpace(recordedPrompt) != "" ||
+		strings.TrimSpace(dispatchPrompt) != "" || len(attachments) > 0
 	userMsgRecorded := s.recordAutoStartMessage(ctx, taskID, sessionID, recordedPrompt, planMode, origin, references, attachments)
 
 	// If the session is in CREATED state, the agent was never started (e.g. workspace-only
@@ -5793,20 +5832,37 @@ func (s *Service) autoStartStepPrompt(
 				release()
 			})
 		}
+		defer heldRelease()
 
 		s.logger.Info("auto-start: session is CREATED, launching agent via StartCreatedSession",
 			zap.String("task_id", taskID),
 			zap.String("session_id", sessionID),
 			zap.String("step_name", stepName))
-		_, err := s.startCreatedSessionWithComposedPrompt(
-			ctx, taskID, sessionID, session.AgentProfileID,
+		queueIdentity, workflowEntry, workflowEntryRequired, captureErr := s.captureWorkflowStartPromptAdmission(ctx, taskID, fresh)
+		if captureErr != nil {
+			requeueTaken()
+			return captureErr
+		}
+		workflowAttempt := newWorkflowStartPromptAttemptWithAdmission(
+			taskID, sessionID, origin, prompt, planMode, attachments, references,
+			handoffForQueue, userMsgRecorded, dispatchInputPresent,
+			queueIdentity, workflowEntry, workflowEntryRequired,
+		)
+		workflowAttempt.workflowEntryRequired = workflowEntryRequired
+		workflowAttempt.configMode, _ = fresh.Metadata["config_mode"].(bool)
+		launchCtx := withWorkflowStartPromptAttempt(ctx, workflowAttempt)
+		execution, err := s.startCreatedSessionWithComposedPrompt(
+			launchCtx, taskID, sessionID, session.AgentProfileID,
 			recordedPrompt, agentPrompt, true, planMode, true, attachments, references,
 		)
+		if execution == nil {
+			workflowAttempt.retire()
+		}
 		// Release the guard as soon as StartCreatedSession has admitted the
 		// launch (succeeded or failed definitively). The deferred release
 		// via heldRelease ensures the guard is always released even on panic.
-		defer heldRelease()
 		if err != nil {
+			workflowAttempt.retire()
 			s.handleCreatedAutoStartLaunchFailure(
 				ctx, taskID, sessionID, stepName, prompt, err,
 				planMode, shouldQueueIfBusy, userMsgRecorded,
@@ -6189,8 +6245,79 @@ func (s *Service) queueAutoStartPrompt(
 	references []v1.EntityReference,
 	handoffText string,
 ) error {
+	queued, err := s.persistAutoStartPrompt(
+		ctx, taskID, sessionID, prompt, planMode, attachments, origin,
+		userMessageRecorded, references, handoffText,
+	)
+	if err != nil {
+		return err
+	}
+	s.scheduleAutoResumeForWorkflowQueue(ctx, sessionID, queued.ID)
+	return nil
+}
+
+// persistAutoStartPrompt writes a workflow auto-start prompt without starting
+// a resume. Failure callbacks use this boundary so the original launch error
+// remains the recovery signal and cleanup can finish before queue draining.
+func (s *Service) persistAutoStartPrompt(
+	ctx context.Context,
+	taskID, sessionID, prompt string,
+	planMode bool,
+	attachments []v1.MessageAttachment,
+	origin workflowMessageOrigin,
+	userMessageRecorded bool,
+	references []v1.EntityReference,
+	handoffText string,
+) (*messagequeue.QueuedMessage, error) {
+	snapshot := s.workflowStartPromptInputSnapshot(
+		ctx, sessionID, prompt, planMode, attachments, references, handoffText,
+	)
+	return s.persistAutoStartPromptWithAdmission(
+		ctx, taskID, sessionID, prompt, planMode, attachments, origin,
+		userMessageRecorded, references, handoffText,
+		snapshot.dispatchInputPresentPtr(), snapshot.configModePtr(), nil, nil,
+	)
+}
+
+func (s *Service) persistAutoStartPromptAtWorkflowEntry(
+	ctx context.Context,
+	taskID, sessionID, prompt string,
+	planMode bool,
+	attachments []v1.MessageAttachment,
+	origin workflowMessageOrigin,
+	userMessageRecorded bool,
+	references []v1.EntityReference,
+	handoffText string,
+	dispatchInputPresent bool,
+	configMode bool,
+	identity messagequeue.QueueSessionIdentity,
+	entry messagequeue.WorkflowEntryIdentity,
+) (*messagequeue.QueuedMessage, error) {
+	inputPresent := dispatchInputPresent
+	launchConfigMode := configMode
+	return s.persistAutoStartPromptWithAdmission(
+		ctx, taskID, sessionID, prompt, planMode, attachments, origin,
+		userMessageRecorded, references, handoffText,
+		&inputPresent, &launchConfigMode, &identity, &entry,
+	)
+}
+
+func (s *Service) persistAutoStartPromptWithAdmission(
+	ctx context.Context,
+	taskID, sessionID, prompt string,
+	planMode bool,
+	attachments []v1.MessageAttachment,
+	origin workflowMessageOrigin,
+	userMessageRecorded bool,
+	references []v1.EntityReference,
+	handoffText string,
+	dispatchInputPresent *bool,
+	configMode *bool,
+	identity *messagequeue.QueueSessionIdentity,
+	entry *messagequeue.WorkflowEntryIdentity,
+) (*messagequeue.QueuedMessage, error) {
 	if s.messageQueue == nil {
-		return fmt.Errorf("message queue is not configured")
+		return nil, fmt.Errorf("message queue is not configured")
 	}
 	meta := workflowMessageMetadata(planMode, origin, references)
 	if userMessageRecorded {
@@ -6199,23 +6326,30 @@ func (s *Service) queueAutoStartPrompt(
 	if handoffText != "" {
 		meta[messagequeue.MetadataStepHandoff] = handoffText
 	}
-	queued, err := s.messageQueue.QueueMessageWithMetadata(
-		ctx,
-		sessionID,
-		taskID,
-		prompt,
-		"",
-		messagequeue.QueuedByWorkflow,
-		planMode,
-		toQueuedAttachments(attachments),
-		meta,
-	)
+	if dispatchInputPresent != nil {
+		meta[metaKeyWorkflowDispatchInputPresent] = *dispatchInputPresent
+	}
+	if configMode != nil {
+		meta[metaKeyWorkflowConfigMode] = *configMode
+	}
+	var queued *messagequeue.QueuedMessage
+	var err error
+	if identity != nil && entry != nil {
+		queued, err = s.messageQueue.QueueMessageWithMetadataForSessionAtWorkflowEntry(
+			ctx, *identity, *entry, prompt, "", messagequeue.QueuedByWorkflow,
+			planMode, toQueuedAttachments(attachments), meta,
+		)
+	} else {
+		queued, err = s.messageQueue.QueueMessageWithMetadata(
+			ctx, sessionID, taskID, prompt, "", messagequeue.QueuedByWorkflow,
+			planMode, toQueuedAttachments(attachments), meta,
+		)
+	}
 	if err != nil {
-		return fmt.Errorf("failed to queue workflow auto-start prompt: %w", err)
+		return nil, fmt.Errorf("failed to queue workflow auto-start prompt: %w", err)
 	}
 	s.publishQueueStatusEvent(ctx, sessionID)
-	s.scheduleAutoResumeForWorkflowQueue(ctx, sessionID, queued.ID)
-	return nil
+	return queued, nil
 }
 
 // scheduleAutoResumeForWorkflowQueue kicks off a background resume when a

apps/backend/internal/orchestrator/messagequeue/repository_memory.go

@@ -305,6 +305,22 @@ func (r *memoryRepository) InsertForSessionWithPolicy(
 	return r.insertLocked(msg, maxPerSession)
 }
 
+func (r *memoryRepository) InsertForSessionWithWorkflowEntry(
+	context.Context,
+	QueueSessionIdentity,
+	WorkflowEntryIdentity,
+	*QueuedMessage,
+	*QueueAttachmentClaim,
+	int,
+	*AutoMergePolicy,
+) error {
+	// The in-memory queue has no shared task repository or workflow-transition
+	// transaction, so accepting a captured entry would falsely claim to fence
+	// workflow moves. Callers must use the legacy path explicitly or provide a
+	// transactional queue repository.
+	return ErrQueueAdmissionUnavailable
+}
+
 // Restore reinserts a previously dequeued entry at its original FIFO position.
 func (r *memoryRepository) Restore(_ context.Context, msg *QueuedMessage, maxPerSession int) error {
 	r.mu.Lock()
@@ -1965,6 +1981,20 @@ func (r *memoryRepository) AutoMergeCandidateIntoAboveForSessionWithPolicy(
 	return r.autoMergeCandidateIntoAboveLocked(candidate)
 }
 
+func (r *memoryRepository) AutoMergeCandidateIntoAboveForSessionWithWorkflowEntry(
+	context.Context,
+	QueueSessionIdentity,
+	WorkflowEntryIdentity,
+	*QueuedMessage,
+	*QueueAttachmentClaim,
+	*AutoMergePolicy,
+) (*QueuedMessage, bool, error) {
+	// Candidate folding is an admission mutation too. It needs the same
+	// shared task/queue transaction as insertion before it can honor an entry
+	// fence.
+	return nil, false, ErrQueueAdmissionUnavailable
+}
+
 func (r *memoryRepository) autoMergeCandidateIntoAboveLocked(candidate *QueuedMessage) (*QueuedMessage, bool, error) {
 	list := r.entries[candidate.SessionID]
 	var target *QueuedMessage

apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go

@@ -33,8 +33,9 @@ type sqliteRepository struct {
 	// OUTSIDE any transaction because a failed statement on PostgreSQL aborts
 	// the whole transaction — the guard must never issue its UPDATE against a
 	// missing table inside a tx.
-	tasksTablePresent        bool
-	taskSessionsTablePresent bool
+	tasksTablePresent          bool
+	taskSessionsTablePresent   bool
+	taskStepTransitionsPresent bool
 }
 
 // NewSQLiteRepository creates a SQLite-backed Repository. The supplied writer
@@ -72,6 +73,10 @@ func NewSQLiteRepository(writer, reader *sqlx.DB) (Repository, error) {
 		return nil, fmt.Errorf("messagequeue: resolve task_sessions table presence: %w", err)
 	}
 	r.taskSessionsTablePresent = present
+	r.taskStepTransitionsPresent, err = r.sharedTablePresent("task_step_transitions")
+	if err != nil {
+		return nil, fmt.Errorf("messagequeue: resolve task step transitions table presence: %w", err)
+	}
 	if present {
 		// Older isolated queue fixtures can provide a task_sessions table
 		// without the incarnation column. They still use the legacy queue API,
@@ -226,6 +231,62 @@ func (r *sqliteRepository) guardActiveTaskTx(ctx context.Context, tx *sqlx.Tx, t
 	return nil
 }
 
+// validateWorkflowEntryTx checks the launch-time entry after the task row has
+// been locked by guardActiveTaskTx. Workflow moves take the same task-row lock
+// before writing their transition ledger row, so this check and queue insert
+// form one admission boundary rather than a read-then-write race.
+func (r *sqliteRepository) validateWorkflowEntryTx(
+	ctx context.Context,
+	tx *sqlx.Tx,
+	taskID string,
+	entry *WorkflowEntryIdentity,
+) error {
+	if entry == nil {
+		return nil
+	}
+	if !r.tasksTablePresent || !r.taskStepTransitionsPresent || entry.TransitionID <= 0 {
+		return ErrWorkflowEntryMismatch
+	}
+	query := `SELECT COALESCE(workflow_id, ''), COALESCE(workflow_step_id, '')
+		FROM tasks WHERE id = ? AND archived_at IS NULL`
+	if r.db.DriverName() == "pgx" {
+		query += ` FOR UPDATE`
+	}
+	var workflowID, workflowStepID string
+	if err := tx.QueryRowxContext(ctx, r.db.Rebind(query), taskID).Scan(&workflowID, &workflowStepID); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrTaskInactive
+		}
+		return fmt.Errorf("read workflow entry for queue admission: %w", err)
+	}
+	if workflowID != entry.WorkflowID || workflowStepID != entry.WorkflowStepID {
+		return ErrWorkflowEntryMismatch
+	}
+	var transitionID int64
+	if err := tx.GetContext(ctx, &transitionID, r.db.Rebind(`
+		SELECT id FROM task_step_transitions
+		WHERE task_id = ?
+		ORDER BY id DESC
+		LIMIT 1
+	`), taskID); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrWorkflowEntryMismatch
+		}
+		return fmt.Errorf("read workflow transition for queue admission: %w", err)
+	}
+	if transitionID != entry.TransitionID {
+		return ErrWorkflowEntryMismatch
+	}
+	generation, err := getLifecycleGenerationTx(ctx, tx, r.db, taskID)
+	if err != nil {
+		return err
+	}
+	if generation != entry.LifecycleGeneration {
+		return ErrLifecycleCancelled
+	}
+	return nil
+}
+
 // guardSessionTx verifies that the owning task session still exists. The
 // session lock acquired before this call already rejects active transfers.
 func (r *sqliteRepository) guardSessionTx(ctx context.Context, tx *sqlx.Tx, sessionID, taskID string) error {
@@ -583,21 +644,21 @@ func decodeEntryOptions(encoded string) (*workflowmove.EntryOptions, error) {
 
 // Insert appends a new entry at the tail of the session's FIFO queue.
 func (r *sqliteRepository) Insert(ctx context.Context, msg *QueuedMessage, maxPerSession int) error {
-	return r.insert(ctx, nil, msg, nil, maxPerSession, nil)
+	return r.insert(ctx, nil, msg, nil, maxPerSession, nil, nil)
 }
 
 func (r *sqliteRepository) InsertForSession(ctx context.Context, identity QueueSessionIdentity, msg *QueuedMessage, maxPerSession int) error {
 	if msg == nil || msg.SessionID != identity.SessionID || msg.TaskID != identity.TaskID {
 		return ErrSessionIdentityMismatch
 	}
-	return r.insert(ctx, &identity, msg, nil, maxPerSession, nil)
+	return r.insert(ctx, &identity, msg, nil, maxPerSession, nil, nil)
 }
 
 func (r *sqliteRepository) InsertForSessionWithClaim(ctx context.Context, identity QueueSessionIdentity, msg *QueuedMessage, claim QueueAttachmentClaim, maxPerSession int) error {
 	if msg == nil || msg.SessionID != identity.SessionID || msg.TaskID != identity.TaskID {
 		return ErrSessionIdentityMismatch
 	}
-	return r.insert(ctx, &identity, msg, &claim, maxPerSession, nil)
+	return r.insert(ctx, &identity, msg, &claim, maxPerSession, nil, nil)
 }
 
 func (r *sqliteRepository) InsertForSessionWithPolicy(
@@ -611,7 +672,22 @@ func (r *sqliteRepository) InsertForSessionWithPolicy(
 	if msg == nil || msg.SessionID != identity.SessionID || msg.TaskID != identity.TaskID {
 		return ErrSessionIdentityMismatch
 	}
-	return r.insert(ctx, &identity, msg, claim, maxPerSession, &policy)
+	return r.insert(ctx, &identity, msg, claim, maxPerSession, &policy, nil)
+}
+
+func (r *sqliteRepository) InsertForSessionWithWorkflowEntry(
+	ctx context.Context,
+	identity QueueSessionIdentity,
+	entry WorkflowEntryIdentity,
+	msg *QueuedMessage,
+	claim *QueueAttachmentClaim,
+	maxPerSession int,
+	policy *AutoMergePolicy,
+) error {
+	if msg == nil || msg.SessionID != identity.SessionID || msg.TaskID != identity.TaskID {
+		return ErrSessionIdentityMismatch
+	}
+	return r.insert(ctx, &identity, msg, claim, maxPerSession, policy, &entry)
 }
 
 func (r *sqliteRepository) insert(
@@ -621,6 +697,7 @@ func (r *sqliteRepository) insert(
 	claim *QueueAttachmentClaim,
 	maxPerSession int,
 	policy *AutoMergePolicy,
+	workflowEntry *WorkflowEntryIdentity,
 ) error {
 	tx, err := r.db.BeginTxx(ctx, nil)
 	if err != nil {
@@ -630,6 +707,9 @@ func (r *sqliteRepository) insert(
 	if err := r.guardActiveTaskTx(ctx, tx, msg.TaskID); err != nil {
 		return err
 	}
+	if err := r.validateWorkflowEntryTx(ctx, tx, msg.TaskID, workflowEntry); err != nil {
+		return err
+	}
 	if err := r.lockSessionTx(ctx, tx, msg.SessionID); err != nil {
 		return err
 	}
@@ -4677,21 +4757,21 @@ func (r *sqliteRepository) AutoMergeCandidateIntoAbove(ctx context.Context, cand
 	if err := r.ensureEditLeaseSchema(ctx); err != nil {
 		return nil, false, err
 	}
-	return r.autoMergeCandidateIntoAbove(ctx, nil, candidate, nil, nil)
+	return r.autoMergeCandidateIntoAbove(ctx, nil, candidate, nil, nil, nil)
 }
 
 func (r *sqliteRepository) AutoMergeCandidateIntoAboveForSession(ctx context.Context, identity QueueSessionIdentity, candidate *QueuedMessage) (*QueuedMessage, bool, error) {
 	if candidate == nil || candidate.SessionID != identity.SessionID || candidate.TaskID != identity.TaskID {
 		return nil, false, ErrSessionIdentityMismatch
 	}
-	return r.autoMergeCandidateIntoAbove(ctx, &identity, candidate, nil, nil)
+	return r.autoMergeCandidateIntoAbove(ctx, &identity, candidate, nil, nil, nil)
 }
 
 func (r *sqliteRepository) AutoMergeCandidateIntoAboveForSessionWithClaim(ctx context.Context, identity QueueSessionIdentity, candidate *QueuedMessage, claim QueueAttachmentClaim) (*QueuedMessage, bool, error) {
 	if candidate == nil || candidate.SessionID != identity.SessionID || candidate.TaskID != identity.TaskID {
 		return nil, false, ErrSessionIdentityMismatch
 	}
-	return r.autoMergeCandidateIntoAbove(ctx, &identity, candidate, &claim, nil)
+	return r.autoMergeCandidateIntoAbove(ctx, &identity, candidate, &claim, nil, nil)
 }
 
 func (r *sqliteRepository) AutoMergeCandidateIntoAboveForSessionWithPolicy(
@@ -4704,7 +4784,21 @@ func (r *sqliteRepository) AutoMergeCandidateIntoAboveForSessionWithPolicy(
 	if candidate == nil || candidate.SessionID != identity.SessionID || candidate.TaskID != identity.TaskID {
 		return nil, false, ErrSessionIdentityMismatch
 	}
-	return r.autoMergeCandidateIntoAbove(ctx, &identity, candidate, claim, &policy)
+	return r.autoMergeCandidateIntoAbove(ctx, &identity, candidate, claim, &policy, nil)
+}
+
+func (r *sqliteRepository) AutoMergeCandidateIntoAboveForSessionWithWorkflowEntry(
+	ctx context.Context,
+	identity QueueSessionIdentity,
+	entry WorkflowEntryIdentity,
+	candidate *QueuedMessage,
+	claim *QueueAttachmentClaim,
+	policy *AutoMergePolicy,
+) (*QueuedMessage, bool, error) {
+	if candidate == nil || candidate.SessionID != identity.SessionID || candidate.TaskID != identity.TaskID {
+		return nil, false, ErrSessionIdentityMismatch
+	}
+	return r.autoMergeCandidateIntoAbove(ctx, &identity, candidate, claim, policy, &entry)
 }
 
 //nolint:cyclop,funlen // the transaction's guards intentionally surround the one durable fold.
@@ -4714,6 +4808,7 @@ func (r *sqliteRepository) autoMergeCandidateIntoAbove(
 	candidate *QueuedMessage,
 	claim *QueueAttachmentClaim,
 	policy *AutoMergePolicy,
+	workflowEntry *WorkflowEntryIdentity,
 ) (*QueuedMessage, bool, error) {
 	unlock := r.withSessionLock(candidate.SessionID)
 	defer unlock()
@@ -4731,6 +4826,9 @@ func (r *sqliteRepository) autoMergeCandidateIntoAbove(
 	if err := r.guardActiveTaskTx(ctx, tx, candidate.TaskID); err != nil {
 		return nil, false, err
 	}
+	if err := r.validateWorkflowEntryTx(ctx, tx, candidate.TaskID, workflowEntry); err != nil {
+		return nil, false, err
+	}
 	if err := r.lockSessionTx(ctx, tx, candidate.SessionID); err != nil {
 		return nil, false, err
 	}

apps/backend/internal/orchestrator/messagequeue/service.go

@@ -98,6 +98,30 @@ type autoMergeAdmissionRepository interface {
 	) (*QueuedMessage, bool, error)
 }
 
+// workflowEntryAdmissionRepository validates a captured workflow entry inside
+// the same task-row-guarded transaction that inserts or folds a queue entry.
+// Repositories without this optional capability retain the legacy admission
+// path for callers that do not provide a workflow entry.
+type workflowEntryAdmissionRepository interface {
+	InsertForSessionWithWorkflowEntry(
+		context.Context,
+		QueueSessionIdentity,
+		WorkflowEntryIdentity,
+		*QueuedMessage,
+		*QueueAttachmentClaim,
+		int,
+		*AutoMergePolicy,
+	) error
+	AutoMergeCandidateIntoAboveForSessionWithWorkflowEntry(
+		context.Context,
+		QueueSessionIdentity,
+		WorkflowEntryIdentity,
+		*QueuedMessage,
+		*QueueAttachmentClaim,
+		*AutoMergePolicy,
+	) (*QueuedMessage, bool, error)
+}
+
 // NewService creates a Service backed by the supplied repository. maxPerSession
 // is the per-session cap (entries beyond this return ErrQueueFull on insert);
 // pass 0 to disable the cap.
@@ -1058,6 +1082,28 @@ func (s *Service) QueueMessageWithMetadataForSession(
 	)
 }
 
+// QueueMessageWithMetadataForSessionAtWorkflowEntry admits a workflow prompt
+// only while the task still owns the immutable entry captured at launch. The
+// repository performs that check in the queue insertion transaction, after it
+// has acquired the task-row guard used by workflow moves.
+func (s *Service) QueueMessageWithMetadataForSessionAtWorkflowEntry(
+	ctx context.Context,
+	identity QueueSessionIdentity,
+	entry WorkflowEntryIdentity,
+	content, model, userID string,
+	planMode bool,
+	attachments []MessageAttachment,
+	metadata map[string]interface{},
+) (*QueuedMessage, error) {
+	if err := s.validateSessionIdentity(ctx, identity); err != nil {
+		return nil, err
+	}
+	return s.queueMessageWithMetadataAdmissionAtWorkflowEntry(
+		ctx, &identity, &entry, identity.SessionID, identity.TaskID, content, model, userID,
+		planMode, attachments, metadata, nil, nil,
+	)
+}
+
 // QueueMessageWithMetadataForSessionWithClientQueueID admits an ordinary
 // browser queue message with a durable caller-owned replay identity. The
 // receipt and any queue insertion, fold, or attachment claim share one
@@ -1181,13 +1227,24 @@ func (s *Service) QueueMessageWithMetadataForSessionAfterInsert(
 //
 //nolint:gocognit // retry boundaries keep policy validation before each mutation.
 func (s *Service) queueMessageWithMetadataAdmission(ctx context.Context, identity *QueueSessionIdentity, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, claim *QueueAttachmentClaim, afterInsert func(context.Context, *QueuedMessage) error) (*QueuedMessage, error) {
+	return s.queueMessageWithMetadataAdmissionAtWorkflowEntry(
+		ctx, identity, nil, sessionID, taskID, content, model, userID, planMode,
+		attachments, metadata, claim, afterInsert,
+	)
+}
+
+// queueMessageWithMetadataAdmissionAtWorkflowEntry is the admission core for
+// both ordinary queue writes and workflow-start recovery writes.
+//
+//nolint:gocognit // retry boundaries keep policy and workflow-entry validation before each mutation.
+func (s *Service) queueMessageWithMetadataAdmissionAtWorkflowEntry(ctx context.Context, identity *QueueSessionIdentity, workflowEntry *WorkflowEntryIdentity, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, claim *QueueAttachmentClaim, afterInsert func(context.Context, *QueuedMessage) error) (*QueuedMessage, error) {
 	var queued *QueuedMessage
 	err := s.WithSessionAdmission(ctx, sessionID, func(admittedCtx context.Context) error {
 		for {
 			policy := s.resolveAdmissionAutoMergePolicy(admittedCtx, identity, sessionID)
-			source, insertErr := s.insertQueueMessageWithMetadata(
+			source, insertErr := s.insertQueueMessageWithMetadataAtWorkflowEntry(
 				admittedCtx, identity, sessionID, taskID, content, model, userID, planMode,
-				attachments, metadata, claim, s.MaxPerSession(), policy,
+				attachments, metadata, claim, s.MaxPerSession(), policy, workflowEntry,
 			)
 			if errors.Is(insertErr, ErrAutoMergePolicyChanged) {
 				continue
@@ -1199,9 +1256,9 @@ func (s *Service) queueMessageWithMetadataAdmission(ctx context.Context, identit
 					return insertErr
 				}
 				var merged *QueuedMessage
-				source, merged, insertErr = s.admitQueueFullMessage(
+				source, merged, insertErr = s.admitQueueFullMessageAtWorkflowEntry(
 					admittedCtx, identity, insertErr, sessionID, taskID, content, model, userID, planMode,
-					attachments, metadata, claim, s.MaxPerSession(), policy,
+					attachments, metadata, claim, s.MaxPerSession(), policy, workflowEntry,
 				)
 				if errors.Is(insertErr, ErrAutoMergePolicyChanged) {
 					continue
@@ -1252,9 +1309,32 @@ func (s *Service) autoMergeCandidate(
 	claim *QueueAttachmentClaim,
 	policy *AutoMergePolicy,
 ) (*QueuedMessage, bool, error) {
+	return s.autoMergeCandidateAtWorkflowEntry(ctx, identity, nil, candidate, claim, policy)
+}
+
+func (s *Service) autoMergeCandidateAtWorkflowEntry(
+	ctx context.Context,
+	identity *QueueSessionIdentity,
+	workflowEntry *WorkflowEntryIdentity,
+	candidate *QueuedMessage,
+	claim *QueueAttachmentClaim,
+	policy *AutoMergePolicy,
+) (*QueuedMessage, bool, error) {
+	if workflowEntry != nil && identity == nil {
+		return nil, false, ErrQueueAdmissionUnavailable
+	}
 	if identity == nil {
 		return s.repo.AutoMergeCandidateIntoAbove(ctx, candidate)
 	}
+	if workflowEntry != nil {
+		workflowRepo, ok := s.repo.(workflowEntryAdmissionRepository)
+		if !ok {
+			return nil, false, ErrQueueAdmissionUnavailable
+		}
+		return workflowRepo.AutoMergeCandidateIntoAboveForSessionWithWorkflowEntry(
+			ctx, *identity, *workflowEntry, candidate, claim, policy,
+		)
+	}
 	if policyRepo, ok := s.repo.(autoMergeAdmissionRepository); ok && policy != nil {
 		return policyRepo.AutoMergeCandidateIntoAboveForSessionWithPolicy(ctx, *identity, candidate, claim, *policy)
 	}
@@ -1281,6 +1361,25 @@ func (s *Service) admitQueueFullMessage(
 	claim *QueueAttachmentClaim,
 	maxPerSession int,
 	policy *AutoMergePolicy,
+) (*QueuedMessage, *QueuedMessage, error) {
+	return s.admitQueueFullMessageAtWorkflowEntry(
+		ctx, identity, queueFullErr, sessionID, taskID, content, model, userID, planMode,
+		attachments, metadata, claim, maxPerSession, policy, nil,
+	)
+}
+
+func (s *Service) admitQueueFullMessageAtWorkflowEntry(
+	ctx context.Context,
+	identity *QueueSessionIdentity,
+	queueFullErr error,
+	sessionID, taskID, content, model, userID string,
+	planMode bool,
+	attachments []MessageAttachment,
+	metadata map[string]interface{},
+	claim *QueueAttachmentClaim,
+	maxPerSession int,
+	policy *AutoMergePolicy,
+	workflowEntry *WorkflowEntryIdentity,
 ) (*QueuedMessage, *QueuedMessage, error) {
 	candidate := &QueuedMessage{
 		SessionID:   sessionID,
@@ -1293,12 +1392,14 @@ func (s *Service) admitQueueFullMessage(
 		QueuedAt:    time.Now().UTC(),
 		QueuedBy:    userID,
 	}
-	merged, didMerge, err := s.autoMergeCandidate(ctx, identity, candidate, claim, policy)
+	merged, didMerge, err := s.autoMergeCandidateAtWorkflowEntry(ctx, identity, workflowEntry, candidate, claim, policy)
 	if errors.Is(err, ErrAutoMergePolicyChanged) {
 		return nil, nil, err
 	}
 	if err != nil {
-		if errors.Is(err, ErrTaskInactive) {
+		if errors.Is(err, ErrTaskInactive) ||
+			errors.Is(err, ErrWorkflowEntryMismatch) ||
+			errors.Is(err, ErrLifecycleCancelled) {
 			return nil, nil, err
 		}
 		s.logger.Error("automatic merge into full queue failed; preserving queue full rejection",
@@ -1312,9 +1413,9 @@ func (s *Service) admitQueueFullMessage(
 			zap.String("surviving_entry_id", merged.ID))
 		return nil, merged, nil
 	}
-	source, err := s.insertQueueMessageWithMetadata(
+	source, err := s.insertQueueMessageWithMetadataAtWorkflowEntry(
 		ctx, identity, sessionID, taskID, content, model, userID, planMode,
-		attachments, metadata, claim, maxPerSession, policy,
+		attachments, metadata, claim, maxPerSession, policy, workflowEntry,
 	)
 	return source, nil, err
 }
@@ -1393,9 +1494,33 @@ func (s *Service) insertQueueMessage(
 	maxPerSession int,
 	policy *AutoMergePolicy,
 ) error {
+	return s.insertQueueMessageAtWorkflowEntry(ctx, identity, nil, msg, claim, maxPerSession, policy)
+}
+
+func (s *Service) insertQueueMessageAtWorkflowEntry(
+	ctx context.Context,
+	identity *QueueSessionIdentity,
+	workflowEntry *WorkflowEntryIdentity,
+	msg *QueuedMessage,
+	claim *QueueAttachmentClaim,
+	maxPerSession int,
+	policy *AutoMergePolicy,
+) error {
+	if workflowEntry != nil && identity == nil {
+		return ErrQueueAdmissionUnavailable
+	}
 	if identity == nil {
 		return s.repo.Insert(ctx, msg, maxPerSession)
 	}
+	if workflowEntry != nil {
+		workflowRepo, ok := s.repo.(workflowEntryAdmissionRepository)
+		if !ok {
+			return ErrQueueAdmissionUnavailable
+		}
+		return workflowRepo.InsertForSessionWithWorkflowEntry(
+			ctx, *identity, *workflowEntry, msg, claim, maxPerSession, policy,
+		)
+	}
 	if policyRepo, ok := s.repo.(autoMergeAdmissionRepository); ok && policy != nil {
 		return policyRepo.InsertForSessionWithPolicy(ctx, *identity, msg, claim, maxPerSession, *policy)
 	}
@@ -1411,6 +1536,13 @@ func (s *Service) insertQueueMessage(
 
 // insertQueueMessageWithMetadata inserts a message with metadata under the per-session admission lock.
 func (s *Service) insertQueueMessageWithMetadata(ctx context.Context, identity *QueueSessionIdentity, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, claim *QueueAttachmentClaim, maxPerSession int, policy *AutoMergePolicy) (*QueuedMessage, error) {
+	return s.insertQueueMessageWithMetadataAtWorkflowEntry(
+		ctx, identity, sessionID, taskID, content, model, userID, planMode, attachments,
+		metadata, claim, maxPerSession, policy, nil,
+	)
+}
+
+func (s *Service) insertQueueMessageWithMetadataAtWorkflowEntry(ctx context.Context, identity *QueueSessionIdentity, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, claim *QueueAttachmentClaim, maxPerSession int, policy *AutoMergePolicy, workflowEntry *WorkflowEntryIdentity) (*QueuedMessage, error) {
 	metadataCopy := copyMessageMetadata(metadata, 0)
 	msg := &QueuedMessage{
 		SessionID:   sessionID,
@@ -1422,7 +1554,7 @@ func (s *Service) insertQueueMessageWithMetadata(ctx context.Context, identity *
 		Metadata:    metadataCopy,
 		QueuedBy:    userID,
 	}
-	err := s.insertQueueMessage(ctx, identity, msg, claim, maxPerSession, policy)
+	err := s.insertQueueMessageAtWorkflowEntry(ctx, identity, workflowEntry, msg, claim, maxPerSession, policy)
 	if err != nil {
 		if errors.Is(err, ErrQueueFull) {
 			s.logger.Info("queue full",

apps/backend/internal/orchestrator/messagequeue/types.go

@@ -158,6 +158,9 @@ var (
 	// ErrSessionIdentityMismatch means the supplied immutable session identity
 	// no longer names the authoritative task-session row.
 	ErrSessionIdentityMismatch = errors.New("queue session identity mismatch")
+	// ErrWorkflowEntryMismatch means a workflow prompt was admitted after its
+	// captured task-step entry had been superseded.
+	ErrWorkflowEntryMismatch = errors.New("queue workflow entry identity mismatch")
 	// ErrLifecycleCancelled means an archive/delete purge invalidated a
 	// previously accepted lifecycle entry before it could be retried.
 	ErrLifecycleCancelled = errors.New("lifecycle queue entry cancelled")
@@ -213,6 +216,17 @@ type QueueSessionIdentity struct {
 	SessionIncarnationID string `json:"session_incarnation_id"`
 }
 
+// WorkflowEntryIdentity is the immutable task workflow entry captured when a
+// workflow auto-start is launched. TransitionID fences leave-and-return
+// re-entry to the same step; LifecycleGeneration fences archive/delete purge
+// work that was captured before the destructive mutation.
+type WorkflowEntryIdentity struct {
+	WorkflowID          string
+	WorkflowStepID      string
+	TransitionID        int64
+	LifecycleGeneration int64
+}
+
 // QueueAttachmentClaim carries authenticated staged-attachment ownership into
 // the queue repository transaction.
 type QueueAttachmentClaim struct {

apps/backend/internal/orchestrator/task_operations.go

@@ -897,6 +897,7 @@ func (s *Service) startCreatedSession(
 		mcpMode = executor.McpModeOffice
 	}
 	initialTurnID, initialTurnCreated := s.startTurnForSessionWithOwnership(ctx, sessionID)
+	ctx = bindWorkflowStartPromptAttemptTurn(ctx, initialTurnID)
 	if err := s.validateClaimedCeilingBinding(ctx, taskID, options.ceilingEntryBinding); err != nil {
 		if initialTurnCreated {
 			s.completeTurnIfCurrent(ctx, sessionID, initialTurnID)
@@ -5629,6 +5630,9 @@ type promptTaskOptions struct {
 	// plumbing and is never passed to a caller.
 	promptAccepted          *atomic.Bool
 	expectedSessionIdentity *messagequeue.QueueSessionIdentity
+	// configModeOverride preserves the launch-time mode for a deferred
+	// workflow prompt whose raw queue content is intentionally empty.
+	configModeOverride *bool
 	// promptAlreadyComposed and fallbackRetryPrompt mirror the composed-prompt
 	// seam autoStartStepPrompt's own ErrExecutionNotFound branch uses (see
 	// fallbackFreshLaunchOnMissingExecution). When promptAlreadyComposed is
@@ -5789,6 +5793,7 @@ func (s *Service) promptTask(ctx context.Context, taskID, sessionID string, prom
 	// AgentExecutionID, then begin the foreground dispatch on it.
 	session, effectivePrompt, foregroundDispatch, err := s.beginForegroundDispatchForPrompt(
 		resumePromptCtx, taskID, sessionID, prompt, planMode, session, foregroundClaim,
+		options.configModeOverride,
 	)
 	if err != nil {
 		return nil, err
@@ -6401,9 +6406,12 @@ func (s *Service) beginForegroundDispatchForPrompt(
 	planMode bool,
 	session *models.TaskSession,
 	foregroundClaim *foregroundClaim,
+	configModeOverride *bool,
 ) (*models.TaskSession, string, *foregroundDispatch, error) {
 	session = s.reloadPromptSession(ctx, sessionID, session)
-	effectivePrompt := s.effectivePromptForSession(sessionID, prompt, planMode, session)
+	effectivePrompt := s.effectivePromptForSessionWithConfigMode(
+		sessionID, prompt, planMode, session, configModeOverride,
+	)
 	activityExecutionID, _ := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
 	foregroundDispatch := s.beginForegroundDispatch(sessionID, foregroundClaim, activityExecutionID)
 	if foregroundDispatch == nil {
@@ -7210,8 +7218,24 @@ func (s *Service) handlePromptDispatchFailure(
 // follows ensureSessionRunning, in case metadata changed in between —
 // without double-prepending either transform's system-prompt wrapper.
 func (s *Service) effectivePromptForSession(sessionID, prompt string, planMode bool, session *models.TaskSession) string {
+	return s.effectivePromptForSessionWithConfigMode(sessionID, prompt, planMode, session, nil)
+}
+
+func (s *Service) effectivePromptForSessionWithConfigMode(
+	sessionID, prompt string,
+	planMode bool,
+	session *models.TaskSession,
+	configModeOverride *bool,
+) string {
 	effectivePrompt := prompt
-	if cm, ok := session.Metadata["config_mode"].(bool); ok && cm {
+	configMode := false
+	if session != nil {
+		configMode, _ = session.Metadata["config_mode"].(bool)
+	}
+	if configModeOverride != nil {
+		configMode = *configModeOverride
+	}
+	if configMode {
 		effectivePrompt = sysprompt.InjectConfigContext(sessionID, prompt)
 	}
 	if planMode {

apps/backend/internal/orchestrator/workflow_start_prompt.go

@@ -0,0 +1,484 @@
+package orchestrator
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync/atomic"
+
+	"github.com/google/uuid"
+	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
+	"github.com/kandev/kandev/internal/task/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+	"go.uber.org/zap"
+)
+
+type workflowStartPromptAttemptContextKey struct{}
+
+type workflowStartPromptAttemptState uint32
+
+const (
+	workflowStartPromptAttemptActive workflowStartPromptAttemptState = iota + 1
+	workflowStartPromptAttemptRetired
+	workflowStartPromptAttemptPreservationClaimed
+)
+
+type workflowStartPromptAttemptClaim struct {
+	state atomic.Uint32
+}
+
+// workflowStartPromptAttempt is the immutable input snapshot for one workflow
+// start. claim is shared by context copies made when the initial turn is bound,
+// so a callback can settle the logical launch exactly once.
+type workflowStartPromptAttempt struct {
+	claim *workflowStartPromptAttemptClaim
+
+	taskID                string
+	sessionID             string
+	workflowStep          workflowMessageOrigin
+	launchToken           string
+	turnID                string
+	queueIdentity         messagequeue.QueueSessionIdentity
+	workflowEntry         messagequeue.WorkflowEntryIdentity
+	workflowEntryRequired bool
+	workflowEntryCaptured bool
+	configMode            bool
+
+	prompt               string
+	planMode             bool
+	dispatchInputPresent bool
+	attachments          []v1.MessageAttachment
+	references           []v1.EntityReference
+	handoffText          string
+	userMessageRecorded  bool
+}
+
+func newWorkflowStartPromptAttempt(
+	taskID, sessionID string,
+	origin workflowMessageOrigin,
+	prompt string,
+	planMode bool,
+	attachments []v1.MessageAttachment,
+	references []v1.EntityReference,
+	handoffText string,
+	userMessageRecorded bool,
+) *workflowStartPromptAttempt {
+	attempt := newWorkflowStartPromptAttemptWithAdmission(
+		taskID, sessionID, origin, prompt, planMode, attachments, references,
+		handoffText, userMessageRecorded,
+		strings.TrimSpace(prompt) != "" || planMode || len(attachments) > 0 || len(references) > 0 || strings.TrimSpace(handoffText) != "",
+		messagequeue.QueueSessionIdentity{}, messagequeue.WorkflowEntryIdentity{}, false,
+	)
+	attempt.workflowEntryRequired = false
+	return attempt
+}
+
+func newWorkflowStartPromptAttemptWithAdmission(
+	taskID, sessionID string,
+	origin workflowMessageOrigin,
+	prompt string,
+	planMode bool,
+	attachments []v1.MessageAttachment,
+	references []v1.EntityReference,
+	handoffText string,
+	userMessageRecorded bool,
+	dispatchInputPresent bool,
+	queueIdentity messagequeue.QueueSessionIdentity,
+	workflowEntry messagequeue.WorkflowEntryIdentity,
+	workflowEntryCaptured bool,
+) *workflowStartPromptAttempt {
+	attempt := &workflowStartPromptAttempt{
+		claim:                 &workflowStartPromptAttemptClaim{},
+		taskID:                taskID,
+		sessionID:             sessionID,
+		workflowStep:          origin,
+		launchToken:           uuid.NewString(),
+		queueIdentity:         queueIdentity,
+		workflowEntry:         workflowEntry,
+		workflowEntryRequired: true,
+		workflowEntryCaptured: workflowEntryCaptured,
+		prompt:                prompt,
+		planMode:              planMode,
+		dispatchInputPresent:  dispatchInputPresent,
+		attachments:           append([]v1.MessageAttachment(nil), attachments...),
+		references:            append([]v1.EntityReference(nil), references...),
+		handoffText:           handoffText,
+		userMessageRecorded:   userMessageRecorded,
+	}
+	attempt.claim.state.Store(uint32(workflowStartPromptAttemptActive))
+	return attempt
+}
+
+func withWorkflowStartPromptAttempt(ctx context.Context, attempt *workflowStartPromptAttempt) context.Context {
+	if ctx == nil || attempt == nil {
+		return ctx
+	}
+	return context.WithValue(ctx, workflowStartPromptAttemptContextKey{}, attempt)
+}
+
+func workflowStartPromptAttemptFromContext(ctx context.Context) *workflowStartPromptAttempt {
+	if ctx == nil {
+		return nil
+	}
+	attempt, _ := ctx.Value(workflowStartPromptAttemptContextKey{}).(*workflowStartPromptAttempt)
+	return attempt
+}
+
+func bindWorkflowStartPromptAttemptTurn(ctx context.Context, turnID string) context.Context {
+	attempt := workflowStartPromptAttemptFromContext(ctx)
+	if attempt == nil || attempt.turnID == turnID {
+		return ctx
+	}
+	bound := *attempt
+	bound.turnID = turnID
+	return withWorkflowStartPromptAttempt(ctx, &bound)
+}
+
+func (a *workflowStartPromptAttempt) retire() {
+	if a == nil || a.claim == nil {
+		return
+	}
+	a.claim.state.CompareAndSwap(
+		uint32(workflowStartPromptAttemptActive),
+		uint32(workflowStartPromptAttemptRetired),
+	)
+}
+
+func (a *workflowStartPromptAttempt) claimPreservation() bool {
+	if a == nil || a.claim == nil {
+		return false
+	}
+	return a.claim.state.CompareAndSwap(
+		uint32(workflowStartPromptAttemptActive),
+		uint32(workflowStartPromptAttemptPreservationClaimed),
+	)
+}
+
+func (a *workflowStartPromptAttempt) hasInput() bool {
+	return a != nil && (a.dispatchInputPresent || len(a.attachments) > 0 ||
+		len(a.references) > 0 || strings.TrimSpace(a.handoffText) != "")
+}
+
+type workflowStartPromptTransitionReader interface {
+	GetLatestTaskStepTransitionID(context.Context, string) (int64, error)
+}
+
+type workflowStartPromptTransitionEnsurer interface {
+	EnsureCurrentTaskStepTransition(context.Context, string) (int64, error)
+}
+
+type workflowStartPromptInputSnapshot struct {
+	dispatchInputPresent      bool
+	dispatchInputPresentKnown bool
+	configMode                bool
+	configModeKnown           bool
+}
+
+func (s workflowStartPromptInputSnapshot) dispatchInputPresentPtr() *bool {
+	if !s.dispatchInputPresentKnown {
+		return nil
+	}
+	present := s.dispatchInputPresent
+	return &present
+}
+
+func (s workflowStartPromptInputSnapshot) configModePtr() *bool {
+	if !s.configModeKnown {
+		return nil
+	}
+	configMode := s.configMode
+	return &configMode
+}
+
+// workflowStartPromptInputSnapshot records actionability at queue admission.
+// A raw prompt can be empty while plan mode, config mode, references, or a
+// handoff still produces a real dispatch. When session metadata cannot be
+// read, only independently known raw input is stamped; the queue drain then
+// retains its retryable read-error behavior instead of guessing empty input.
+func (s *Service) workflowStartPromptInputSnapshot(
+	ctx context.Context,
+	sessionID, prompt string,
+	planMode bool,
+	attachments []v1.MessageAttachment,
+	references []v1.EntityReference,
+	handoffText string,
+) workflowStartPromptInputSnapshot {
+	snapshot := workflowStartPromptInputSnapshot{
+		dispatchInputPresent: strings.TrimSpace(prompt) != "" || planMode ||
+			len(attachments) > 0 || len(references) > 0 || strings.TrimSpace(handoffText) != "",
+		dispatchInputPresentKnown: false,
+	}
+	if snapshot.dispatchInputPresent {
+		snapshot.dispatchInputPresentKnown = true
+	}
+	if s.repo == nil {
+		return snapshot
+	}
+	session, err := s.repo.GetTaskSession(ctx, sessionID)
+	if err != nil || session == nil {
+		return snapshot
+	}
+	snapshot.configMode, _ = session.Metadata["config_mode"].(bool)
+	snapshot.configModeKnown = true
+	snapshot.dispatchInputPresent = snapshot.dispatchInputPresent || snapshot.configMode
+	snapshot.dispatchInputPresentKnown = true
+	return snapshot
+}
+
+// captureWorkflowStartPromptAdmission records the task-step transition,
+// session incarnation, and purge generation observed before the launch. The
+// queue repository rechecks all three inside its insertion transaction.
+func (s *Service) captureWorkflowStartPromptAdmission(
+	ctx context.Context,
+	taskID string,
+	session *models.TaskSession,
+) (messagequeue.QueueSessionIdentity, messagequeue.WorkflowEntryIdentity, bool, error) {
+	if s.messageQueue == nil || !s.messageQueue.SupportsAtomicDeferredMoveTransition() {
+		// Focused and legacy backends without the shared SQL queue/task
+		// transaction intentionally use the existing non-fenced recovery path.
+		return messagequeue.QueueSessionIdentity{}, messagequeue.WorkflowEntryIdentity{}, false, nil
+	}
+	if session == nil || taskID == "" || session.ID == "" || session.QueueIncarnationID == "" {
+		return messagequeue.QueueSessionIdentity{}, messagequeue.WorkflowEntryIdentity{}, true,
+			fmt.Errorf("workflow launch admission identity is incomplete")
+	}
+	reader, ok := s.repo.(workflowStartPromptTransitionReader)
+	if !ok {
+		return messagequeue.QueueSessionIdentity{}, messagequeue.WorkflowEntryIdentity{}, true,
+			fmt.Errorf("workflow launch admission transition reader is unavailable")
+	}
+	task, transitionID, err := s.captureWorkflowStartPromptTaskAndEntry(ctx, taskID, reader)
+	if err != nil {
+		return messagequeue.QueueSessionIdentity{}, messagequeue.WorkflowEntryIdentity{}, true, err
+	}
+	generation := int64(0)
+	if s.messageQueue != nil {
+		generation, err = s.messageQueue.LifecycleGeneration(ctx, taskID)
+		if err != nil {
+			return messagequeue.QueueSessionIdentity{}, messagequeue.WorkflowEntryIdentity{}, true,
+				fmt.Errorf("read lifecycle generation for workflow launch admission: %w", err)
+		}
+	}
+	return messagequeue.QueueSessionIdentity{
+			TaskID: taskID, SessionID: session.ID, SessionIncarnationID: session.QueueIncarnationID,
+		}, messagequeue.WorkflowEntryIdentity{
+			WorkflowID: task.WorkflowID, WorkflowStepID: task.WorkflowStepID,
+			TransitionID: transitionID, LifecycleGeneration: generation,
+		}, true, nil
+}
+
+func (s *Service) captureWorkflowStartPromptTaskAndEntry(
+	ctx context.Context,
+	taskID string,
+	reader workflowStartPromptTransitionReader,
+) (*models.Task, int64, error) {
+	task, err := s.repo.GetTask(ctx, taskID)
+	if err != nil {
+		return nil, 0, fmt.Errorf("read task for workflow launch admission: %w", err)
+	}
+	if task == nil {
+		return nil, 0, fmt.Errorf("task %q is missing for workflow launch admission", taskID)
+	}
+	transitionID, err := reader.GetLatestTaskStepTransitionID(ctx, taskID)
+	if err != nil {
+		return nil, 0, fmt.Errorf("read workflow launch entry for task %q: %w", taskID, err)
+	}
+	if transitionID <= 0 {
+		ensurer, ok := s.repo.(workflowStartPromptTransitionEnsurer)
+		if !ok {
+			return nil, 0, fmt.Errorf("workflow launch admission cannot establish a durable entry for task %q", taskID)
+		}
+		transitionID, err = ensurer.EnsureCurrentTaskStepTransition(ctx, taskID)
+		if err != nil {
+			return nil, 0, fmt.Errorf("establish workflow launch entry for task %q: %w", taskID, err)
+		}
+		if transitionID <= 0 {
+			return nil, 0, fmt.Errorf("task %q has no current workflow entry", taskID)
+		}
+	}
+	// The entry read and task projection must describe one durable state. A
+	// move may have committed between the first task read and the transition
+	// read, so reload after either the normal read or a legacy backfill.
+	task, err = s.repo.GetTask(ctx, taskID)
+	if err != nil {
+		return nil, 0, fmt.Errorf("reload task after reading workflow launch entry: %w", err)
+	}
+	if task == nil {
+		return nil, 0, fmt.Errorf("task %q disappeared after reading workflow launch entry", taskID)
+	}
+	latestTransitionID, err := reader.GetLatestTaskStepTransitionID(ctx, taskID)
+	if err != nil {
+		return nil, 0, fmt.Errorf("re-read workflow launch entry for task %q: %w", taskID, err)
+	}
+	if latestTransitionID <= 0 {
+		return nil, 0, fmt.Errorf("task %q lost its current workflow entry", taskID)
+	}
+	return task, latestTransitionID, nil
+}
+
+func (s *Service) retireWorkflowStartPromptAttempt(ctx context.Context, taskID, sessionID, executionID string) {
+	attempt := workflowStartPromptAttemptFromContext(ctx)
+	if attempt == nil || attempt.taskID != taskID || attempt.sessionID != sessionID {
+		return
+	}
+	if executionID != "" && s.agentManager != nil {
+		liveExecutionID, err := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
+		if err == nil && liveExecutionID != "" && liveExecutionID != executionID {
+			return
+		}
+	}
+	attempt.retire()
+}
+
+// preserveWorkflowStartPromptAfterFailure is called while the session's
+// cancellation guard is held and before the executor projects the startup
+// failure. It validates the captured ownership boundary, claims the attempt,
+// and writes the existing queue form without scheduling a new resume.
+func (s *Service) preserveWorkflowStartPromptAfterFailure(
+	ctx context.Context,
+	taskID, sessionID, executionID string,
+) {
+	attempt := workflowStartPromptAttemptFromContext(ctx)
+	if attempt == nil || attempt.taskID != taskID || attempt.sessionID != sessionID || !attempt.hasInput() {
+		return
+	}
+	if attempt.workflowEntryRequired && !attempt.workflowEntryCaptured {
+		return
+	}
+	if !s.workflowStartPromptAttemptIsCurrent(ctx, attempt, executionID) {
+		return
+	}
+	if !attempt.claimPreservation() {
+		return
+	}
+	var err error
+	if attempt.workflowEntryCaptured {
+		_, err = s.persistAutoStartPromptAtWorkflowEntry(
+			ctx,
+			attempt.taskID,
+			attempt.sessionID,
+			attempt.prompt,
+			attempt.planMode,
+			attempt.attachments,
+			attempt.workflowStep,
+			attempt.userMessageRecorded,
+			attempt.references,
+			attempt.handoffText,
+			attempt.dispatchInputPresent,
+			attempt.configMode,
+			attempt.queueIdentity,
+			attempt.workflowEntry,
+		)
+	} else {
+		_, err = s.persistAutoStartPrompt(
+			ctx,
+			attempt.taskID,
+			attempt.sessionID,
+			attempt.prompt,
+			attempt.planMode,
+			attempt.attachments,
+			attempt.workflowStep,
+			attempt.userMessageRecorded,
+			attempt.references,
+			attempt.handoffText,
+		)
+	}
+	if err != nil {
+		// Keep the startup error as the user-visible recovery signal. This
+		// diagnostic carries only ownership identifiers, never prompt content.
+		s.logger.Warn("failed to preserve workflow prompt after asynchronous startup failure",
+			zap.String("task_id", attempt.taskID),
+			zap.String("session_id", attempt.sessionID),
+			zap.String("agent_execution_id", executionID),
+			zap.String("launch_token", attempt.launchToken),
+			zap.Error(err))
+	}
+}
+
+func (s *Service) workflowStartPromptAttemptIsCurrent(
+	ctx context.Context,
+	attempt *workflowStartPromptAttempt,
+	executionID string,
+) bool {
+	if attempt == nil || executionID == "" || s.repo == nil {
+		return false
+	}
+	if !s.workflowStartPromptSessionIsCurrent(ctx, attempt, executionID) {
+		return false
+	}
+	if !s.workflowStartPromptExecutionIsCurrent(ctx, attempt.sessionID, executionID) {
+		return false
+	}
+	if !s.workflowStartPromptTaskIsCurrent(ctx, attempt) {
+		return false
+	}
+	if !s.workflowStartPromptEntryIsCurrent(ctx, attempt) {
+		return false
+	}
+	return s.workflowStartPromptTurnIsCurrent(ctx, attempt)
+}
+
+func (s *Service) workflowStartPromptSessionIsCurrent(
+	ctx context.Context,
+	attempt *workflowStartPromptAttempt,
+	executionID string,
+) bool {
+	session, err := s.repo.GetTaskSession(ctx, attempt.sessionID)
+	if err != nil || session == nil || session.TaskID != attempt.taskID {
+		return false
+	}
+	if session.State != models.TaskSessionStateStarting {
+		return false
+	}
+	return session.AgentExecutionID == "" || session.AgentExecutionID == executionID
+}
+
+func (s *Service) workflowStartPromptExecutionIsCurrent(ctx context.Context, sessionID, executionID string) bool {
+	if s.agentManager == nil {
+		return false
+	}
+	liveExecutionID, err := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
+	return err == nil && liveExecutionID != "" && liveExecutionID == executionID
+}
+
+func (s *Service) workflowStartPromptTaskIsCurrent(ctx context.Context, attempt *workflowStartPromptAttempt) bool {
+	task, err := s.repo.GetTask(ctx, attempt.taskID)
+	if err != nil || task == nil || taskArchived(task) {
+		return false
+	}
+	if attempt.workflowStep.StepID != "" && task.WorkflowStepID != attempt.workflowStep.StepID {
+		return false
+	}
+	if attempt.workflowEntryCaptured && (task.WorkflowID != attempt.workflowEntry.WorkflowID ||
+		task.WorkflowStepID != attempt.workflowEntry.WorkflowStepID) {
+		return false
+	}
+	return true
+}
+
+func (s *Service) workflowStartPromptEntryIsCurrent(ctx context.Context, attempt *workflowStartPromptAttempt) bool {
+	if !attempt.workflowEntryCaptured {
+		return true
+	}
+	reader, ok := s.repo.(workflowStartPromptTransitionReader)
+	if !ok {
+		return false
+	}
+	transitionID, err := reader.GetLatestTaskStepTransitionID(ctx, attempt.taskID)
+	if err != nil || transitionID != attempt.workflowEntry.TransitionID {
+		return false
+	}
+	if s.messageQueue == nil {
+		return true
+	}
+	generation, err := s.messageQueue.LifecycleGeneration(ctx, attempt.taskID)
+	return err == nil && generation == attempt.workflowEntry.LifecycleGeneration
+}
+
+func (s *Service) workflowStartPromptTurnIsCurrent(ctx context.Context, attempt *workflowStartPromptAttempt) bool {
+	if attempt.turnID == "" || s.turnService == nil {
+		return false
+	}
+	activeTurnID, err := s.peekActiveTurnID(ctx, attempt.sessionID)
+	return err == nil && activeTurnID == attempt.turnID
+}

apps/backend/internal/orchestrator/workflow_start_prompt_admission_test.go

@@ -0,0 +1,126 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
+	"github.com/kandev/kandev/internal/task/models"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+)
+
+func TestCaptureWorkflowStartPromptAdmission_BackfillsLegacyEntry(t *testing.T) {
+	svc, repo, ctx, _, _ := newWorkflowStartPromptAttemptFixture(
+		t, models.TaskSessionStateStarting, "workflow-fence-execution",
+	)
+	if _, err := repo.DB().ExecContext(ctx,
+		`DELETE FROM task_step_transitions WHERE task_id = ?`, "workflow-fence-task"); err != nil {
+		t.Fatalf("remove legacy task transition rows: %v", err)
+	}
+	session, err := repo.GetTaskSession(ctx, "workflow-fence-session")
+	if err != nil {
+		t.Fatalf("get session: %v", err)
+	}
+	_, entry, required, err := svc.captureWorkflowStartPromptAdmission(
+		ctx, "workflow-fence-task", session,
+	)
+	if err != nil {
+		t.Fatalf("capture legacy workflow entry: %v", err)
+	}
+	if !required {
+		t.Fatal("workflow entry admission = unsupported, want strict admission")
+	}
+	if entry.TransitionID <= 0 {
+		t.Fatalf("backfilled transition id = %d, want positive identity", entry.TransitionID)
+	}
+	var rows int
+	if err := repo.DB().QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM task_step_transitions WHERE task_id = ?`,
+		"workflow-fence-task").Scan(&rows); err != nil {
+		t.Fatalf("count backfilled transition rows: %v", err)
+	}
+	if rows != 1 {
+		t.Fatalf("backfilled transition rows = %d, want 1", rows)
+	}
+}
+
+type workflowStartPromptTransitionReadFailureRepository struct {
+	*sqliterepo.Repository
+	err error
+}
+
+func (r *workflowStartPromptTransitionReadFailureRepository) GetLatestTaskStepTransitionID(context.Context, string) (int64, error) {
+	return 0, r.err
+}
+
+func TestAutoStartCreatedLaunch_AbortsWhenAdmissionCaptureFails(t *testing.T) {
+	fixture := newWorkflowAsyncStartFailureFixture(t, errors.New("provider startup failed"))
+	fixture.svc.repo = &workflowStartPromptTransitionReadFailureRepository{
+		Repository: fixture.repo,
+		err:        errors.New("transition ledger unavailable"),
+	}
+
+	err := fixture.svc.autoStartStepPrompt(
+		context.Background(), fixture.taskID, fixture.session, fixture.step,
+		fixture.prompt, false, true, nil,
+	)
+	if err == nil || !strings.Contains(err.Error(), "read workflow launch entry") {
+		t.Fatalf("auto-start error = %v, want admission capture failure", err)
+	}
+	select {
+	case <-fixture.startEntered:
+		t.Fatal("agent launch started after admission capture failed")
+	case <-time.After(100 * time.Millisecond):
+	}
+	if got := fixture.svc.messageQueue.GetStatus(context.Background(), fixture.sessionID).Count; got != 0 {
+		t.Fatalf("queue count after admission capture failure = %d, want 0", got)
+	}
+}
+
+type workflowStartPromptSessionReadFailureRepository struct {
+	*sqliterepo.Repository
+	err error
+}
+
+func (r *workflowStartPromptSessionReadFailureRepository) GetTaskSession(context.Context, string) (*models.TaskSession, error) {
+	return nil, r.err
+}
+
+func TestDispatchTakenQueuedMessage_RetainsEntryWhenInputReadFails(t *testing.T) {
+	svc, repo, ctx, _, _ := newWorkflowStartPromptAttemptFixture(
+		t, models.TaskSessionStateStarting, "workflow-fence-execution",
+	)
+	svc.repo = &workflowStartPromptSessionReadFailureRepository{
+		Repository: repo,
+		err:        errors.New("session metadata temporarily unavailable"),
+	}
+	if _, err := svc.messageQueue.QueueMessageWithMetadata(
+		ctx, "workflow-fence-session", "workflow-fence-task", "", "",
+		messagequeue.QueuedByWorkflow, false, nil, nil,
+	); err != nil {
+		t.Fatalf("queue empty raw entry: %v", err)
+	}
+	identity, err := svc.messageQueue.ResolveSessionIdentity(
+		ctx, "workflow-fence-task", "workflow-fence-session",
+	)
+	if err != nil {
+		t.Fatalf("resolve session identity: %v", err)
+	}
+	queued, exists, autoRun, err := svc.messageQueue.
+		ReserveQueuedWithAutoRunForSession(ctx, identity)
+	if err != nil {
+		t.Fatalf("reserve queued entry: %v", err)
+	}
+	if !exists || !autoRun || queued == nil {
+		t.Fatalf("reservation = (%#v, %t, %t), want one auto-run entry", queued, exists, autoRun)
+	}
+	if dispatched := svc.dispatchTakenQueuedMessageForSession(ctx, identity, queued, true); dispatched {
+		t.Fatal("dispatch returned success after input inspection failed")
+	}
+	if got := svc.messageQueue.GetStatus(ctx, identity.SessionID).Count; got != 1 {
+		t.Fatalf("queue count after input inspection failure = %d, want retained entry", got)
+	}
+}

apps/backend/internal/orchestrator/workflow_start_prompt_effective_input_test.go

@@ -0,0 +1,265 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/orchestrator/watcher"
+	"github.com/kandev/kandev/internal/sysprompt"
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+func TestWorkflowAsyncStartFailure_PreservesPlanOnlyInputThroughRecovery(t *testing.T) {
+	fixture := newWorkflowAsyncStartFailureFixture(t, errors.New("provider startup failed"))
+	ctx := context.Background()
+	task, err := fixture.repo.GetTask(ctx, fixture.taskID)
+	if err != nil {
+		t.Fatalf("get task: %v", err)
+	}
+	task.Description = ""
+	if err := fixture.repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("clear task description: %v", err)
+	}
+
+	if err := fixture.svc.autoStartStepPrompt(
+		ctx, fixture.taskID, fixture.session, fixture.step, "", true, true, nil,
+	); err != nil {
+		t.Fatalf("autoStartStepPrompt returned launch error before asynchronous failure: %v", err)
+	}
+	select {
+	case <-fixture.startEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for asynchronous startup")
+	}
+	close(fixture.releaseStart)
+	select {
+	case <-fixture.startReturned:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for startup failure callback")
+	}
+	waitForSessionState(t, fixture.repo, fixture.sessionID, models.TaskSessionStateFailed)
+
+	status := fixture.svc.messageQueue.GetStatus(ctx, fixture.sessionID)
+	if status.Count != 1 {
+		t.Fatalf("queue count after plan-only startup failure = %d, want 1", status.Count)
+	}
+	if status.Entries[0].Content != "" {
+		t.Fatalf("queued raw content = %q, want empty raw content", status.Entries[0].Content)
+	}
+	if !status.Entries[0].PlanMode {
+		t.Fatal("queued plan mode = false, want true")
+	}
+	if present, _ := status.Entries[0].Metadata[metaKeyWorkflowDispatchInputPresent].(bool); !present {
+		t.Fatal("queued dispatch input marker = false, want true")
+	}
+
+	recoveryDone := make(chan error, 1)
+	go func() {
+		_, recoverErr := fixture.svc.RecoverSession(ctx, fixture.taskID, fixture.sessionID, "fresh_start")
+		recoveryDone <- recoverErr
+	}()
+	select {
+	case <-fixture.secondStartEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for explicit recovery startup")
+	}
+	resumeAttempt, ok := fixture.svc.resumeAttemptStore().current(fixture.sessionID)
+	if !ok || resumeAttempt == nil {
+		t.Fatal("explicit recovery did not retain a live resume attempt")
+	}
+	fixture.svc.handleAgentBootReady(ctx, watcher.AgentEventData{
+		TaskID:           fixture.taskID,
+		SessionID:        fixture.sessionID,
+		AgentExecutionID: "workflow-async-execution",
+		AttemptID:        resumeAttempt.identity(),
+	})
+	select {
+	case recoverErr := <-recoveryDone:
+		if recoverErr != nil {
+			t.Fatalf("explicit recovery failed: %v", recoverErr)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for explicit recovery")
+	}
+	select {
+	case <-fixture.agentMgr.promptDone:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for plan-only prompt delivery")
+	}
+
+	if len(fixture.agentMgr.capturedPrompts) != 1 {
+		t.Fatalf("provider prompt count = %d, want one effective prompt", len(fixture.agentMgr.capturedPrompts))
+	}
+	if strings.Count(fixture.agentMgr.capturedPrompts[0], sysprompt.PlanMode()) != 1 {
+		t.Fatalf("provider plan instructions were duplicated: %q", fixture.agentMgr.capturedPrompts[0])
+	}
+	messageCreator, ok := fixture.svc.messageCreator.(*mockMessageCreator)
+	if !ok {
+		t.Fatal("workflow service message creator has unexpected type")
+	}
+	if len(messageCreator.userMessages) != 1 {
+		t.Fatalf("user transcript rows = %d, want one", len(messageCreator.userMessages))
+	}
+	if strings.Count(messageCreator.userMessages[0].content, sysprompt.PlanMode()) != 1 {
+		t.Fatalf("transcript plan instructions were duplicated: %q", messageCreator.userMessages[0].content)
+	}
+}
+
+func TestWorkflowAsyncStartFailure_PreservesConfigOnlyInputThroughRecovery(t *testing.T) {
+	fixture := newWorkflowAsyncStartFailureFixture(t, errors.New("provider startup failed"))
+	ctx := context.Background()
+	task, err := fixture.repo.GetTask(ctx, fixture.taskID)
+	if err != nil {
+		t.Fatalf("get task: %v", err)
+	}
+	task.Description = ""
+	if err := fixture.repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("clear task description: %v", err)
+	}
+	session, err := fixture.repo.GetTaskSession(ctx, fixture.sessionID)
+	if err != nil {
+		t.Fatalf("get config session: %v", err)
+	}
+	if session.Metadata == nil {
+		session.Metadata = make(map[string]interface{})
+	}
+	session.Metadata["config_mode"] = true
+	if err := fixture.repo.UpdateSessionMetadata(ctx, fixture.sessionID, session.Metadata); err != nil {
+		t.Fatalf("enable config mode: %v", err)
+	}
+	session, err = fixture.repo.GetTaskSession(ctx, fixture.sessionID)
+	if err != nil {
+		t.Fatalf("reload config session: %v", err)
+	}
+	fixture.session = session
+
+	if err := fixture.svc.autoStartStepPrompt(
+		ctx, fixture.taskID, fixture.session, fixture.step, "", false, true, nil,
+	); err != nil {
+		t.Fatalf("autoStartStepPrompt returned launch error before asynchronous failure: %v", err)
+	}
+	select {
+	case <-fixture.startEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for asynchronous startup")
+	}
+	close(fixture.releaseStart)
+	select {
+	case <-fixture.startReturned:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for startup failure callback")
+	}
+	waitForSessionState(t, fixture.repo, fixture.sessionID, models.TaskSessionStateFailed)
+
+	status := fixture.svc.messageQueue.GetStatus(ctx, fixture.sessionID)
+	if status.Count != 1 {
+		t.Fatalf("queue count after config-only startup failure = %d, want 1", status.Count)
+	}
+	if status.Entries[0].Content != "" {
+		t.Fatalf("queued raw content = %q, want empty raw content", status.Entries[0].Content)
+	}
+	if status.Entries[0].PlanMode {
+		t.Fatal("queued plan mode = true, want false")
+	}
+	if present, _ := status.Entries[0].Metadata[metaKeyWorkflowDispatchInputPresent].(bool); !present {
+		t.Fatal("queued config-only dispatch input marker = false, want true")
+	}
+	if configMode, _ := status.Entries[0].Metadata[metaKeyWorkflowConfigMode].(bool); !configMode {
+		t.Fatal("queued launch config-mode marker = false, want true")
+	}
+	// The queue entry owns the mode that made the empty raw prompt actionable.
+	// Recovery must not discard or turn it into an empty provider call if the
+	// mutable session projection changes before the drain.
+	if err := fixture.repo.UpdateSessionMetadata(ctx, fixture.sessionID, map[string]interface{}{"config_mode": false}); err != nil {
+		t.Fatalf("disable current config mode: %v", err)
+	}
+
+	recoveryDone := make(chan error, 1)
+	go func() {
+		_, recoverErr := fixture.svc.RecoverSession(ctx, fixture.taskID, fixture.sessionID, "fresh_start")
+		recoveryDone <- recoverErr
+	}()
+	select {
+	case <-fixture.secondStartEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for explicit recovery startup")
+	}
+	resumeAttempt, ok := fixture.svc.resumeAttemptStore().current(fixture.sessionID)
+	if !ok || resumeAttempt == nil {
+		t.Fatal("explicit recovery did not retain a live resume attempt")
+	}
+	fixture.svc.handleAgentBootReady(ctx, watcher.AgentEventData{
+		TaskID:           fixture.taskID,
+		SessionID:        fixture.sessionID,
+		AgentExecutionID: "workflow-async-execution",
+		AttemptID:        resumeAttempt.identity(),
+	})
+	select {
+	case recoverErr := <-recoveryDone:
+		if recoverErr != nil {
+			t.Fatalf("explicit recovery failed: %v", recoverErr)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for explicit recovery")
+	}
+	select {
+	case <-fixture.agentMgr.promptDone:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for config-only prompt delivery")
+	}
+
+	if len(fixture.agentMgr.capturedPrompts) != 1 {
+		t.Fatalf("provider prompt count = %d, want one effective prompt", len(fixture.agentMgr.capturedPrompts))
+	}
+	configContext := sysprompt.FormatConfigContext(fixture.sessionID)
+	if strings.Count(fixture.agentMgr.capturedPrompts[0], configContext) != 1 {
+		t.Fatalf("provider config instructions were duplicated or lost: %q", fixture.agentMgr.capturedPrompts[0])
+	}
+	messageCreator, ok := fixture.svc.messageCreator.(*mockMessageCreator)
+	if !ok {
+		t.Fatal("workflow service message creator has unexpected type")
+	}
+	if len(messageCreator.userMessages) != 1 {
+		t.Fatalf("user transcript rows = %d, want one", len(messageCreator.userMessages))
+	}
+	if strings.Count(messageCreator.userMessages[0].content, configContext) != 1 {
+		t.Fatalf("transcript config instructions were duplicated or lost: %q", messageCreator.userMessages[0].content)
+	}
+}
+
+func TestWorkflowAsyncStartFailure_ExcludesTrulyEmptyInput(t *testing.T) {
+	fixture := newWorkflowAsyncStartFailureFixture(t, errors.New("provider startup failed"))
+	ctx := context.Background()
+
+	if err := fixture.svc.autoStartStepPrompt(
+		ctx, fixture.taskID, fixture.session, fixture.step, "", false, true, nil,
+	); err != nil {
+		t.Fatalf("autoStartStepPrompt returned launch error before asynchronous failure: %v", err)
+	}
+	select {
+	case <-fixture.startEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for asynchronous startup")
+	}
+	close(fixture.releaseStart)
+	select {
+	case <-fixture.startReturned:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for startup failure callback")
+	}
+	waitForSessionState(t, fixture.repo, fixture.sessionID, models.TaskSessionStateFailed)
+
+	if got := fixture.svc.messageQueue.GetStatus(ctx, fixture.sessionID).Count; got != 0 {
+		t.Fatalf("truly empty startup failure queued %d entries, want 0", got)
+	}
+	messageCreator, ok := fixture.svc.messageCreator.(*mockMessageCreator)
+	if !ok {
+		t.Fatal("workflow service message creator has unexpected type")
+	}
+	if len(messageCreator.userMessages) != 0 {
+		t.Fatalf("truly empty startup failure recorded %d transcript rows, want 0", len(messageCreator.userMessages))
+	}
+}

apps/backend/internal/orchestrator/workflow_start_prompt_fencing_test.go

@@ -0,0 +1,279 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
+	"github.com/kandev/kandev/internal/task/models"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+)
+
+func TestWorkflowAsyncStartFailure_RejectsStaleAndDuplicateCallbacks(t *testing.T) {
+	t.Run("move committed between ownership read and queue insertion is fenced", func(t *testing.T) {
+		svc, repo, ctx, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+			t, models.TaskSessionStateStarting, "workflow-fence-execution",
+		)
+		moveBeforeInsert := &workflowStartPromptMoveBeforeInsertRepository{
+			Repository: newWorkflowStartPromptSQLiteRepository(t, repo),
+			beforeInsert: func() {
+				task, err := repo.GetTask(ctx, "workflow-fence-task")
+				if err != nil {
+					t.Fatalf("get task at admission barrier: %v", err)
+				}
+				task.WorkflowStepID = "successor-step"
+				if err := repo.UpdateTask(ctx, task); err != nil {
+					t.Fatalf("commit successor move at admission barrier: %v", err)
+				}
+			},
+		}
+		svc.messageQueue = messagequeue.NewService(
+			moveBeforeInsert, messagequeue.DefaultMaxPerSession, testLogger(),
+		)
+		svc.preserveWorkflowStartPromptAfterFailure(
+			attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+		)
+		if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 0 {
+			t.Fatalf("obsolete callback queued %d entries after the committed move", got)
+		}
+		agentMgr := svc.agentManager.(*mockAgentManager)
+		if len(agentMgr.capturedPrompts) != 0 {
+			t.Fatalf("obsolete callback dispatched %d prompts after the committed move", len(agentMgr.capturedPrompts))
+		}
+	})
+
+	t.Run("leave and return to the same step is fenced", func(t *testing.T) {
+		svc, repo, ctx, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+			t, models.TaskSessionStateStarting, "workflow-fence-execution",
+		)
+		svc.messageQueue = messagequeue.NewService(
+			newWorkflowStartPromptSQLiteRepository(t, repo), messagequeue.DefaultMaxPerSession, testLogger(),
+		)
+		task, err := repo.GetTask(ctx, "workflow-fence-task")
+		if err != nil {
+			t.Fatalf("get task: %v", err)
+		}
+		task.WorkflowStepID = "successor-step"
+		if err := repo.UpdateTask(ctx, task); err != nil {
+			t.Fatalf("leave original step: %v", err)
+		}
+		task.WorkflowStepID = "workflow-fence-step"
+		if err := repo.UpdateTask(ctx, task); err != nil {
+			t.Fatalf("return to original step: %v", err)
+		}
+
+		svc.preserveWorkflowStartPromptAfterFailure(
+			attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+		)
+		if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 0 {
+			t.Fatalf("re-entered step accepted %d obsolete entries", got)
+		}
+		agentMgr := svc.agentManager.(*mockAgentManager)
+		if len(agentMgr.capturedPrompts) != 0 {
+			t.Fatalf("re-entered step dispatched %d obsolete prompts", len(agentMgr.capturedPrompts))
+		}
+	})
+
+	t.Run("duplicate callback claims once", func(t *testing.T) {
+		svc, repo, ctx, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+			t, models.TaskSessionStateStarting, "workflow-fence-execution",
+		)
+		if handled := svc.handleAgentStartFailed(
+			attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+			errors.New("startup failed"), false,
+		); handled {
+			t.Fatal("generic startup failure should remain owned by executor projection")
+		}
+		if handled := svc.handleAgentStartFailed(
+			attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+			errors.New("duplicate startup failed"), false,
+		); handled {
+			t.Fatal("duplicate generic startup failure should remain owned by executor projection")
+		}
+		if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 1 {
+			t.Fatalf("queue count = %d, want 1 after duplicate callbacks", got)
+		}
+		if _, err := repo.GetTaskSession(ctx, "workflow-fence-session"); err != nil {
+			t.Fatalf("session was not retained after duplicate callbacks: %v", err)
+		}
+	})
+
+	t.Run("successor turn is fenced", func(t *testing.T) {
+		svc, _, ctx, attemptCtx, turnID := newWorkflowStartPromptAttemptFixture(
+			t, models.TaskSessionStateStarting, "workflow-fence-execution",
+		)
+		svc.completeTurnIfCurrent(ctx, "workflow-fence-session", turnID)
+		successorTurnID := svc.startTurnForSession(ctx, "workflow-fence-session")
+		if successorTurnID == "" || successorTurnID == turnID {
+			t.Fatalf("successor turn = %q, want a new turn after %q", successorTurnID, turnID)
+		}
+		svc.handleAgentStartFailed(
+			attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+			errors.New("stale startup failed"), false,
+		)
+		if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 0 {
+			t.Fatalf("stale turn callback queued %d entries", got)
+		}
+	})
+
+	t.Run("rotated execution is fenced", func(t *testing.T) {
+		svc, repo, ctx, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+			t, models.TaskSessionStateStarting, "workflow-fence-execution",
+		)
+		seedExecutorRunning(t, repo, "workflow-fence-session", "workflow-fence-task", "workflow-successor-execution")
+		svc.handleAgentStartFailed(
+			attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+			errors.New("superseded startup failed"), false,
+		)
+		if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 0 {
+			t.Fatalf("rotated execution callback queued %d entries", got)
+		}
+	})
+
+	for _, tc := range []struct {
+		name   string
+		mutate func(*testing.T, *Service, *sqliterepo.Repository, context.Context, *workflowStartPromptAttempt, string)
+	}{
+		{
+			name: "terminal session",
+			mutate: func(t *testing.T, _ *Service, repo *sqliterepo.Repository, ctx context.Context, _ *workflowStartPromptAttempt, _ string) {
+				if err := repo.UpdateTaskSessionState(ctx, "workflow-fence-session", models.TaskSessionStateCancelled, "cancelled"); err != nil {
+					t.Fatalf("cancel session: %v", err)
+				}
+			},
+		},
+		{
+			name: "archived task",
+			mutate: func(t *testing.T, _ *Service, repo *sqliterepo.Repository, ctx context.Context, _ *workflowStartPromptAttempt, _ string) {
+				if err := repo.ArchiveTask(ctx, "workflow-fence-task"); err != nil {
+					t.Fatalf("archive task: %v", err)
+				}
+			},
+		},
+		{
+			name: "superseded workflow entry",
+			mutate: func(t *testing.T, _ *Service, repo *sqliterepo.Repository, ctx context.Context, _ *workflowStartPromptAttempt, _ string) {
+				task, err := repo.GetTask(ctx, "workflow-fence-task")
+				if err != nil {
+					t.Fatalf("get task: %v", err)
+				}
+				task.WorkflowStepID = "successor-step"
+				if err := repo.UpdateTask(ctx, task); err != nil {
+					t.Fatalf("move task to successor step: %v", err)
+				}
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			svc, repo, ctx, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+				t, models.TaskSessionStateStarting, "workflow-fence-execution",
+			)
+			tc.mutate(t, svc, repo, ctx, workflowStartPromptAttemptFromContext(attemptCtx), "workflow-fence-execution")
+			svc.handleAgentStartFailed(
+				attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+				errors.New("stale startup failed"), false,
+			)
+			if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 0 {
+				t.Fatalf("%s callback queued %d entries", tc.name, got)
+			}
+		})
+	}
+
+	t.Run("deleted task or session is fenced", func(t *testing.T) {
+		svc, repo, ctx, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+			t, models.TaskSessionStateStarting, "workflow-fence-execution",
+		)
+		if err := repo.DeleteTask(ctx, "workflow-fence-task"); err != nil {
+			t.Fatalf("delete task: %v", err)
+		}
+		svc.handleAgentStartFailed(
+			attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+			errors.New("deleted startup failed"), false,
+		)
+		if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 0 {
+			t.Fatalf("deleted task callback queued %d entries", got)
+		}
+	})
+}
+
+type workflowStartPromptMoveBeforeInsertRepository struct {
+	messagequeue.Repository
+	beforeInsert func()
+	once         sync.Once
+}
+
+func newWorkflowStartPromptSQLiteRepository(t *testing.T, repo *sqliterepo.Repository) messagequeue.Repository {
+	t.Helper()
+	db := sqlx.NewDb(repo.DB(), "sqlite3")
+	queueRepo, err := messagequeue.NewSQLiteRepository(db, db)
+	if err != nil {
+		t.Fatalf("create shared queue repository: %v", err)
+	}
+	return queueRepo
+}
+
+func (r *workflowStartPromptMoveBeforeInsertRepository) InsertForSession(
+	ctx context.Context,
+	identity messagequeue.QueueSessionIdentity,
+	msg *messagequeue.QueuedMessage,
+	maxPerSession int,
+) error {
+	r.once.Do(r.beforeInsert)
+	return r.Repository.InsertForSession(ctx, identity, msg, maxPerSession)
+}
+
+func (r *workflowStartPromptMoveBeforeInsertRepository) InsertForSessionWithWorkflowEntry(
+	ctx context.Context,
+	identity messagequeue.QueueSessionIdentity,
+	entry messagequeue.WorkflowEntryIdentity,
+	msg *messagequeue.QueuedMessage,
+	claim *messagequeue.QueueAttachmentClaim,
+	maxPerSession int,
+	policy *messagequeue.AutoMergePolicy,
+) error {
+	r.once.Do(r.beforeInsert)
+	repo, ok := r.Repository.(interface {
+		InsertForSessionWithWorkflowEntry(
+			context.Context,
+			messagequeue.QueueSessionIdentity,
+			messagequeue.WorkflowEntryIdentity,
+			*messagequeue.QueuedMessage,
+			*messagequeue.QueueAttachmentClaim,
+			int,
+			*messagequeue.AutoMergePolicy,
+		) error
+	})
+	if !ok {
+		return r.Repository.InsertForSession(ctx, identity, msg, maxPerSession)
+	}
+	return repo.InsertForSessionWithWorkflowEntry(ctx, identity, entry, msg, claim, maxPerSession, policy)
+}
+
+func (r *workflowStartPromptMoveBeforeInsertRepository) AutoMergeCandidateIntoAboveForSessionWithWorkflowEntry(
+	ctx context.Context,
+	identity messagequeue.QueueSessionIdentity,
+	entry messagequeue.WorkflowEntryIdentity,
+	candidate *messagequeue.QueuedMessage,
+	claim *messagequeue.QueueAttachmentClaim,
+	policy *messagequeue.AutoMergePolicy,
+) (*messagequeue.QueuedMessage, bool, error) {
+	repo, ok := r.Repository.(interface {
+		AutoMergeCandidateIntoAboveForSessionWithWorkflowEntry(
+			context.Context,
+			messagequeue.QueueSessionIdentity,
+			messagequeue.WorkflowEntryIdentity,
+			*messagequeue.QueuedMessage,
+			*messagequeue.QueueAttachmentClaim,
+			*messagequeue.AutoMergePolicy,
+		) (*messagequeue.QueuedMessage, bool, error)
+	})
+	if !ok {
+		return nil, false, messagequeue.ErrNoMergeTarget
+	}
+	return repo.AutoMergeCandidateIntoAboveForSessionWithWorkflowEntry(
+		ctx, identity, entry, candidate, claim, policy,
+	)
+}

apps/backend/internal/orchestrator/workflow_start_prompt_metadata_test.go

@@ -0,0 +1,100 @@
+package orchestrator
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
+	"github.com/kandev/kandev/internal/task/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+func TestWorkflowAsyncStartFailure_PreservesInputMetadata(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID      = "workflow-metadata-task"
+		sessionID   = "workflow-metadata-session"
+		stepID      = "workflow-metadata-step"
+		executionID = "workflow-metadata-execution"
+	)
+
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting)
+	task, err := repo.GetTask(ctx, taskID)
+	if err != nil {
+		t.Fatalf("get task: %v", err)
+	}
+	task.WorkflowStepID = stepID
+	if err := repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("update task workflow step: %v", err)
+	}
+	seedExecutorRunning(t, repo, sessionID, taskID, executionID)
+
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+	svc.agentManager = agentMgr
+	svc.turnService = &repoTurnService{repo: repo}
+	turnID := svc.startTurnForSession(ctx, sessionID)
+	if turnID == "" {
+		t.Fatal("expected active turn for metadata preservation")
+	}
+
+	origin := workflowMessageOrigin{StepID: stepID, StepName: "Async metadata", StepColor: "#abc123"}
+	attachments := []v1.MessageAttachment{{
+		Type:         "image",
+		AttachmentID: "attachment-1",
+		MimeType:     "image/png",
+		Name:         "diagram.png",
+		SizeBytes:    42,
+		DeliveryMode: "path",
+	}}
+	references := []v1.EntityReference{{
+		Version:  1,
+		Ref:      "JIRA-123",
+		Provider: "jira",
+		Kind:     "issue",
+		ID:       "123",
+		Title:    "Preserve launch input",
+		URL:      "https://example.invalid/JIRA-123",
+		Scope:    "workspace",
+	}}
+	attempt := newWorkflowStartPromptAttempt(
+		taskID, sessionID, origin, "raw workflow prompt", true,
+		attachments, references, "handoff once", true,
+	)
+	serviceCtx := bindWorkflowStartPromptAttemptTurn(
+		withWorkflowStartPromptAttempt(ctx, attempt), turnID,
+	)
+	svc.preserveWorkflowStartPromptAfterFailure(serviceCtx, taskID, sessionID, executionID)
+
+	status := svc.messageQueue.GetStatus(ctx, sessionID)
+	if status.Count != 1 {
+		t.Fatalf("queue count = %d, want 1", status.Count)
+	}
+	queued := status.Entries[0]
+	if queued.Content != "raw workflow prompt" {
+		t.Fatalf("queued content = %q, want raw workflow prompt", queued.Content)
+	}
+	if !queued.PlanMode {
+		t.Fatal("queued plan mode = false, want true")
+	}
+	if !reflect.DeepEqual(queued.Attachments, toQueuedAttachments(attachments)) {
+		t.Fatalf("queued attachments = %#v, want %#v", queued.Attachments, toQueuedAttachments(attachments))
+	}
+	if got, _ := queued.Metadata["workflow_step_id"].(string); got != stepID {
+		t.Fatalf("workflow_step_id = %q, want %q", got, stepID)
+	}
+	if got, _ := queued.Metadata["workflow_step_name"].(string); got != "Async metadata" {
+		t.Fatalf("workflow_step_name = %q, want Async metadata", got)
+	}
+	if got, _ := queued.Metadata[messagequeue.MetadataStepHandoff].(string); got != "handoff once" {
+		t.Fatalf("step handoff = %q, want handoff once", got)
+	}
+	if recorded, _ := queued.Metadata[metaKeyUserMessageRecorded].(bool); !recorded {
+		t.Fatal("user_message_recorded = false, want true")
+	}
+	if got, ok := queued.Metadata["entity_references"].([]v1.EntityReference); !ok || !reflect.DeepEqual(got, references) {
+		t.Fatalf("entity references = %#v, want %#v", queued.Metadata["entity_references"], references)
+	}
+}

apps/backend/internal/orchestrator/workflow_start_prompt_test.go

@@ -0,0 +1,757 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	agentruntime "github.com/kandev/kandev/internal/agent/runtime"
+	dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic"
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/agent/runtime/routingerr"
+	"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"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+	wfmodels "github.com/kandev/kandev/internal/workflow/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+func TestWorkflowAsyncStartFailure_PreservesPromptThroughRecovery(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID      = "workflow-async-task"
+		sessionID   = "workflow-async-session"
+		stepID      = "workflow-async-step"
+		profileID   = "workflow-async-profile"
+		executionID = "workflow-async-execution"
+	)
+
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateCreated)
+	task, err := repo.GetTask(ctx, taskID)
+	if err != nil {
+		t.Fatalf("get task: %v", err)
+	}
+	task.WorkflowStepID = stepID
+	if err := repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("update task workflow step: %v", err)
+	}
+	// The mock message creator does not persist rows in SQLite. Keep the task
+	// description empty so recovery's unrelated initial-message backfill does
+	// not fabricate a second row while this test counts the queue marker's
+	// exactly-once transcript behavior.
+	task.Description = ""
+	if err := repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("clear task description for transcript assertion: %v", err)
+	}
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("get session: %v", err)
+	}
+	session.AgentProfileID = profileID
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("update session profile: %v", err)
+	}
+	seedExecutorRunning(t, repo, sessionID, taskID, executionID)
+
+	step := &wfmodels.WorkflowStep{ID: stepID, WorkflowID: "wf1", Name: "Async work"}
+	stepGetter := newMockStepGetter()
+	stepGetter.steps[stepID] = step
+	taskRepo := newMockTaskRepo()
+	taskRepo.tasks[taskID] = &v1.Task{
+		ID:          taskID,
+		WorkflowID:  "wf1",
+		Title:       "Async workflow task",
+		Description: "task description",
+		State:       v1.TaskStateInProgress,
+	}
+
+	startEntered := make(chan struct{})
+	releaseStart := make(chan struct{})
+	startReturned := make(chan struct{})
+	secondStartEntered := make(chan struct{})
+	startupErr := &lifecycle.BootstrapFailure{
+		Code:   models.AgentErrorCauseCodeTransportUnavailable,
+		Detail: "The agent could not start.",
+		Cause:  errors.New("provider startup failed"),
+	}
+	var svc *Service
+	var startCalls atomic.Int32
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup: repo,
+		isAgentRunning:         true,
+		promptDone:             make(chan struct{}),
+		launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
+			return &executor.LaunchAgentResponse{AgentExecutionID: executionID, WorktreePath: req.RepositoryPath}, nil
+		},
+		startAgentProcessFunc: func(_ context.Context, _ string) error {
+			call := startCalls.Add(1)
+			if call == 1 {
+				close(startEntered)
+				<-releaseStart
+				close(startReturned)
+				return startupErr
+			}
+			if call == 2 {
+				close(secondStartEntered)
+				return nil
+			}
+			t.Errorf("unexpected startAgentProcess call %d", call)
+			return nil
+		},
+	}
+
+	svc = createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
+	queueRepo := newWorkflowStartPromptSQLiteRepository(t, repo)
+	svc.messageQueue = messagequeue.NewService(queueRepo, messagequeue.DefaultMaxPerSession, testLogger())
+	svc.messageCreator = &mockMessageCreator{}
+	svc.turnService = &repoTurnService{repo: repo}
+	svc.executor.SetOnAgentStartFailed(svc.handleAgentStartFailed)
+	svc.executor.SetOnAgentProcessStarted(svc.handleAgentProcessStarted)
+	svc.executor.SetOnAgentProcessStartFailed(svc.handleAgentProcessStartFailed)
+
+	if err := svc.autoStartStepPrompt(ctx, taskID, session, step, "Run the async workflow step", false, true, nil); err != nil {
+		t.Fatalf("autoStartStepPrompt returned launch error before asynchronous failure: %v", err)
+	}
+	select {
+	case <-startEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for asynchronous startup")
+	}
+	if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 {
+		t.Fatalf("queue populated before startup failure: %d", got)
+	}
+
+	close(releaseStart)
+	select {
+	case <-startReturned:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for startup failure callback")
+	}
+
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		failedSession, getErr := repo.GetTaskSession(ctx, sessionID)
+		if getErr == nil && failedSession != nil && failedSession.State == models.TaskSessionStateFailed {
+			break
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	failedSession, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("reload failed session: %v", err)
+	}
+	if failedSession.State != models.TaskSessionStateFailed {
+		t.Fatalf("session state = %s, want FAILED after asynchronous startup failure", failedSession.State)
+	}
+	if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 1 {
+		t.Fatalf("expected one preserved workflow prompt after asynchronous failure, queue count = %d", got)
+	}
+
+	recoveryDone := make(chan error, 1)
+	go func() {
+		_, recoverErr := svc.RecoverSession(ctx, taskID, sessionID, "fresh_start")
+		recoveryDone <- recoverErr
+	}()
+	select {
+	case <-secondStartEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for explicit recovery startup")
+	}
+	resumeAttempt, ok := svc.resumeAttemptStore().current(sessionID)
+	if !ok || resumeAttempt == nil {
+		t.Fatal("explicit recovery did not retain a live resume attempt")
+	}
+	svc.handleAgentBootReady(ctx, watcher.AgentEventData{
+		TaskID:           taskID,
+		SessionID:        sessionID,
+		AgentExecutionID: executionID,
+		AttemptID:        resumeAttempt.identity(),
+	})
+	select {
+	case recoverErr := <-recoveryDone:
+		if recoverErr != nil {
+			t.Fatalf("explicit recovery failed: %v", recoverErr)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for explicit recovery")
+	}
+	select {
+	case <-agentMgr.promptDone:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for preserved prompt delivery")
+	}
+	deadline = time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) && svc.messageQueue.GetStatus(ctx, sessionID).Count != 0 {
+		time.Sleep(5 * time.Millisecond)
+	}
+	if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 {
+		t.Fatalf("queue count after recovery delivery = %d, want 0", got)
+	}
+	if len(agentMgr.capturedPrompts) != 1 {
+		t.Fatalf("provider prompt count = %d, want one preserved prompt", len(agentMgr.capturedPrompts))
+	}
+	if agentMgr.capturedPrompts[0] != "Run the async workflow step" {
+		t.Fatalf("provider prompt = %q, want preserved workflow input", agentMgr.capturedPrompts[0])
+	}
+	messageCreator, ok := svc.messageCreator.(*mockMessageCreator)
+	if !ok {
+		t.Fatal("workflow service message creator has unexpected type")
+	}
+	if len(messageCreator.userMessages) != 1 {
+		t.Fatalf("user transcript rows = %d, want one", len(messageCreator.userMessages))
+	}
+}
+
+func TestWorkflowAsyncStartFailure_QueueFailureKeepsLaunchError(t *testing.T) {
+	fixture := newWorkflowAsyncStartFailureFixture(t, &lifecycle.BootstrapFailure{
+		Code:   models.AgentErrorCauseCodeTransportUnavailable,
+		Detail: "The agent could not start.",
+		Cause:  errors.New("provider startup failed"),
+	})
+	queueErr := errors.New("queue unavailable")
+	fixture.svc.messageQueue = messagequeue.NewService(&workflowStartPromptQueueFailureRepository{
+		Repository: newAuthoritativeMemoryRepository(fixture.repo),
+		err:        queueErr,
+	}, messagequeue.DefaultMaxPerSession, testLogger())
+
+	if err := fixture.svc.autoStartStepPrompt(
+		context.Background(), fixture.taskID, fixture.session, fixture.step,
+		fixture.prompt, false, true, nil,
+	); err != nil {
+		t.Fatalf("autoStartStepPrompt returned launch error before asynchronous failure: %v", err)
+	}
+	select {
+	case <-fixture.startEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for asynchronous startup")
+	}
+	close(fixture.releaseStart)
+	select {
+	case <-fixture.startReturned:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for startup failure callback")
+	}
+	waitForSessionState(t, fixture.repo, fixture.sessionID, models.TaskSessionStateFailed)
+
+	if got := fixture.svc.messageQueue.GetStatus(context.Background(), fixture.sessionID).Count; got != 0 {
+		t.Fatalf("queue count after failed persistence = %d, want 0", got)
+	}
+	failedSession, err := fixture.repo.GetTaskSession(context.Background(), fixture.sessionID)
+	if err != nil {
+		t.Fatalf("reload failed session: %v", err)
+	}
+	lastError, ok := models.LoadLastAgentError(failedSession.Metadata)
+	if !ok {
+		t.Fatal("typed launch error was lost when queue persistence failed")
+	}
+	if lastError.Code != models.LaunchErrorCategoryGenericLaunchFailure {
+		t.Fatalf("last launch error code = %q, want generic launch failure", lastError.Code)
+	}
+	if lastError.Message != "The agent could not start." {
+		t.Fatalf("last launch error message = %q, want safe bootstrap detail", lastError.Message)
+	}
+	if strings.Contains(lastError.Details, "provider startup failed") {
+		t.Fatalf("last launch error details expose provider cause: %q", lastError.Details)
+	}
+}
+
+func TestWorkflowAsyncStartFailure_QueueFullKeepsLaunchError(t *testing.T) {
+	fixture := newWorkflowAsyncStartFailureFixture(t, errors.New("provider startup failed"))
+	ctx := context.Background()
+	fixture.svc.messageQueue.SetMaxPerSession(1)
+	fixture.svc.messageQueue.SetAutoMergeEnabled(false)
+	if err := fixture.svc.messageQueue.SetAutoRun(ctx, fixture.sessionID, false); err != nil {
+		t.Fatalf("pause queue auto-run: %v", err)
+	}
+	if _, err := fixture.svc.messageQueue.QueueMessage(
+		ctx, fixture.sessionID, fixture.taskID, "existing queued prompt", "",
+		messagequeue.QueuedByUser, false, nil,
+	); err != nil {
+		t.Fatalf("seed full queue: %v", err)
+	}
+
+	if err := fixture.svc.autoStartStepPrompt(
+		ctx, fixture.taskID, fixture.session, fixture.step,
+		fixture.prompt, false, true, nil,
+	); err != nil {
+		t.Fatalf("autoStartStepPrompt returned launch error before asynchronous failure: %v", err)
+	}
+	select {
+	case <-fixture.startEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for asynchronous startup")
+	}
+	close(fixture.releaseStart)
+	select {
+	case <-fixture.startReturned:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for startup failure callback")
+	}
+	waitForSessionState(t, fixture.repo, fixture.sessionID, models.TaskSessionStateFailed)
+
+	status := fixture.svc.messageQueue.GetStatus(ctx, fixture.sessionID)
+	if status.Count != 1 || status.Entries[0].Content != "existing queued prompt" {
+		t.Fatalf("queue after full admission = %#v, want original single entry", status)
+	}
+	failedSession, err := fixture.repo.GetTaskSession(ctx, fixture.sessionID)
+	if err != nil {
+		t.Fatalf("reload failed session: %v", err)
+	}
+	lastError, ok := models.LoadLastAgentError(failedSession.Metadata)
+	if !ok || lastError.Code != models.LaunchErrorCategoryGenericLaunchFailure {
+		t.Fatalf("last launch error = %#v, want generic launch failure", lastError)
+	}
+}
+
+func TestWorkflowAsyncStartFailure_PreservesPromptWhenTranscriptWriteFails(t *testing.T) {
+	fixture := newWorkflowAsyncStartFailureFixture(t, errors.New("provider startup failed"))
+	fixture.svc.messageCreator = &mockMessageCreator{userMessageErr: errors.New("transcript unavailable")}
+	ctx := context.Background()
+
+	if err := fixture.svc.autoStartStepPrompt(
+		ctx, fixture.taskID, fixture.session, fixture.step,
+		fixture.prompt, false, true, nil,
+	); err != nil {
+		t.Fatalf("autoStartStepPrompt returned launch error before asynchronous failure: %v", err)
+	}
+	select {
+	case <-fixture.startEntered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for asynchronous startup")
+	}
+	close(fixture.releaseStart)
+	select {
+	case <-fixture.startReturned:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for startup failure callback")
+	}
+	waitForSessionState(t, fixture.repo, fixture.sessionID, models.TaskSessionStateFailed)
+
+	status := fixture.svc.messageQueue.GetStatus(ctx, fixture.sessionID)
+	if status.Count != 1 {
+		t.Fatalf("queue count after transcript failure = %d, want 1", status.Count)
+	}
+	if recorded, _ := status.Entries[0].Metadata[metaKeyUserMessageRecorded].(bool); recorded {
+		t.Fatal("queue incorrectly marked a failed transcript write as recorded")
+	}
+}
+
+func TestWorkflowAsyncStartFailure_PreservesClassifiedBootstrapErrors(t *testing.T) {
+	tests := []struct {
+		name    string
+		err     error
+		handled bool
+	}{
+		{
+			name: "generic bootstrap",
+			err: &lifecycle.BootstrapFailure{
+				Code:   models.AgentErrorCauseCodeTransportUnavailable,
+				Detail: "The agent could not start.",
+				Cause:  errors.New("transport unavailable"),
+			},
+		},
+		{
+			name: "authentication bootstrap",
+			err: &lifecycle.BootstrapFailure{
+				Code:   models.AgentErrorCauseCodeAuthenticationRequired,
+				Detail: "Authentication is required.",
+				Cause:  errors.New("auth required"),
+			},
+			handled: true,
+		},
+		{
+			name: "managed runtime bootstrap",
+			err: &routingerr.ManagedRuntimeStartupError{
+				Code:    routingerr.CodeManagedRuntimeNpmResolution,
+				Details: "managed runtime could not be resolved",
+				Cause:   errors.New("runtime resolution failed"),
+			},
+			handled: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			svc, _, ctx, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+				t, models.TaskSessionStateStarting, "workflow-classified-execution",
+			)
+			handled := svc.handleAgentStartFailed(
+				attemptCtx, "workflow-fence-task", "workflow-fence-session",
+				"workflow-classified-execution", tc.err, false,
+			)
+			if handled != tc.handled {
+				t.Fatalf("handleAgentStartFailed = %v, want %v", handled, tc.handled)
+			}
+			if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 1 {
+				t.Fatalf("queue count = %d, want one preserved entry", got)
+			}
+		})
+	}
+}
+
+func TestWorkflowAsyncStartFailure_DynamicFallbackOwnership(t *testing.T) {
+	ctx := context.Background()
+	svc, repo, _, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+		t, models.TaskSessionStateStarting, "workflow-fence-execution",
+	)
+	engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo))
+	svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true))
+	decision := seedClaimedDynamicRoute(
+		t, ctx, repo, engine, "workflow-fence-session", "workflow-fence-execution",
+	)
+
+	// A classified synchronous candidate failure is advanced by the conductor;
+	// the successful candidate's process-start callback owns retirement of the
+	// logical workflow attempt. A late failure for that candidate must not
+	// replay the prompt after the fallback has succeeded.
+	svc.handleAgentProcessStarted(
+		attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+	)
+	state, err := repo.LoadRouteState(ctx, "workflow-fence-session")
+	if err != nil {
+		t.Fatalf("load dynamic route state: %v", err)
+	}
+	if state == nil || state.Generation != decision.Generation || state.Status != dynamicRouteStatusActive {
+		t.Fatalf("dynamic route state = %#v, want active generation %d", state, decision.Generation)
+	}
+
+	if handled := svc.handleAgentStartFailed(
+		attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-fence-execution",
+		errors.New("final candidate startup failed after success callback"), false,
+	); handled {
+		t.Fatal("generic final candidate failure should remain owned by executor projection")
+	}
+	if got := svc.messageQueue.GetStatus(ctx, "workflow-fence-session").Count; got != 0 {
+		t.Fatalf("late dynamic candidate failure queued %d entries", got)
+	}
+}
+
+func TestWorkflowAsyncStartFailure_PersistedQueueSurvivesRestart(t *testing.T) {
+	ctx := context.Background()
+	svc, _, _, attemptCtx, _ := newWorkflowStartPromptAttemptFixture(
+		t, models.TaskSessionStateStarting, "workflow-restart-execution",
+	)
+	dbPath := filepath.Join(t.TempDir(), "workflow-queue.db")
+	queue, queueDB := newWorkflowTransferQueue(t, dbPath)
+	svc.messageQueue = queue
+	// This fixture intentionally uses a queue-only database to verify durable
+	// restart persistence. Entry-fenced admission requires the production
+	// shared task/queue database, so exercise the ordinary queue path here.
+	workflowStartPromptAttemptFromContext(attemptCtx).workflowEntryCaptured = false
+	workflowStartPromptAttemptFromContext(attemptCtx).workflowEntryRequired = false
+
+	if err := queue.SetAutoRun(ctx, "workflow-fence-session", false); err != nil {
+		t.Fatalf("pause queue auto-run: %v", err)
+	}
+	svc.preserveWorkflowStartPromptAfterFailure(
+		attemptCtx, "workflow-fence-task", "workflow-fence-session", "workflow-restart-execution",
+	)
+	status := queue.GetStatus(ctx, "workflow-fence-session")
+	if status.Count != 1 || status.AutoRun {
+		t.Fatalf("queue before restart = %#v, want one paused entry", status)
+	}
+	if err := queueDB.Close(); err != nil {
+		t.Fatalf("close queue database: %v", err)
+	}
+
+	restartedQueue, restartedDB := newWorkflowTransferQueue(t, dbPath)
+	defer func() { _ = restartedDB.Close() }()
+	restarted := restartedQueue.GetStatus(ctx, "workflow-fence-session")
+	if restarted.Count != 1 || restarted.AutoRun {
+		t.Fatalf("queue after restart = %#v, want one paused entry", restarted)
+	}
+	if restarted.Entries[0].Content != "fenced prompt" {
+		t.Fatalf("restarted queue content = %q, want preserved prompt", restarted.Entries[0].Content)
+	}
+}
+
+func TestWorkflowAsyncStartFailure_SuccessAndSynchronousRejection(t *testing.T) {
+	t.Run("successful startup retires attempt", func(t *testing.T) {
+		fixture := newWorkflowAsyncStartFailureFixture(t, nil)
+		if err := fixture.svc.autoStartStepPrompt(
+			context.Background(), fixture.taskID, fixture.session, fixture.step,
+			fixture.prompt, false, true, nil,
+		); err != nil {
+			t.Fatalf("autoStartStepPrompt returned launch error: %v", err)
+		}
+		select {
+		case <-fixture.startEntered:
+		case <-time.After(3 * time.Second):
+			t.Fatal("timed out waiting for successful asynchronous startup")
+		}
+		close(fixture.releaseStart)
+		select {
+		case <-fixture.startReturned:
+		case <-time.After(3 * time.Second):
+			t.Fatal("timed out waiting for successful startup callback")
+		}
+		select {
+		case <-fixture.processStarted:
+		case <-time.After(3 * time.Second):
+			t.Fatal("timed out waiting for successful process-start callback")
+		}
+		if got := fixture.svc.messageQueue.GetStatus(context.Background(), fixture.sessionID).Count; got != 0 {
+			t.Fatalf("successful startup created %d recovery entries", got)
+		}
+	})
+
+	t.Run("synchronous permanent rejection does not queue", func(t *testing.T) {
+		ctx := context.Background()
+		const (
+			taskID    = "workflow-rejected-task"
+			sessionID = "workflow-rejected-session"
+			stepID    = "workflow-rejected-step"
+		)
+		repo := setupTestRepo(t)
+		seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateCreated)
+		dbTask, err := repo.GetTask(ctx, taskID)
+		if err != nil {
+			t.Fatalf("get rejected task: %v", err)
+		}
+		dbTask.ProjectID = "office-project"
+		dbTask.WorkflowStepID = stepID
+		if err := repo.UpdateTask(ctx, dbTask); err != nil {
+			t.Fatalf("mark rejected task as Office-owned: %v", err)
+		}
+		seedExecutorRunning(t, repo, sessionID, taskID, "workflow-rejected-execution")
+
+		taskRepo := newMockTaskRepo()
+		taskRepo.tasks[taskID] = &v1.Task{ID: taskID, Title: "Rejected task", State: v1.TaskStateInProgress}
+		step := &wfmodels.WorkflowStep{ID: stepID, WorkflowID: "wf1", Name: "Rejected step"}
+		stepGetter := newMockStepGetter()
+		stepGetter.steps[stepID] = step
+		svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, &mockAgentManager{
+			repoForExecutionLookup: repo,
+		})
+		svc.messageCreator = &mockMessageCreator{}
+		session, err := repo.GetTaskSession(ctx, sessionID)
+		if err != nil {
+			t.Fatalf("get rejected session: %v", err)
+		}
+		session.AgentProfileID = "workflow-rejected-profile"
+		if err := repo.UpdateTaskSession(ctx, session); err != nil {
+			t.Fatalf("set rejected session profile: %v", err)
+		}
+
+		err = svc.autoStartStepPrompt(ctx, taskID, session, step, "Do not retry this", false, true, nil)
+		if err == nil || !strings.Contains(err.Error(), "office tasks must be started through Office") {
+			t.Fatalf("rejection error = %v, want Office scheduler rejection", err)
+		}
+		if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 0 {
+			t.Fatalf("synchronous rejection created %d recovery entries", got)
+		}
+	})
+}
+
+type workflowStartPromptQueueFailureRepository struct {
+	messagequeue.Repository
+	err error
+}
+
+func (r *workflowStartPromptQueueFailureRepository) Insert(
+	context.Context, *messagequeue.QueuedMessage, int,
+) error {
+	return r.err
+}
+
+func (r *workflowStartPromptQueueFailureRepository) InsertForSession(
+	context.Context, messagequeue.QueueSessionIdentity, *messagequeue.QueuedMessage, int,
+) error {
+	return r.err
+}
+
+type workflowAsyncStartFailureFixture struct {
+	svc       *Service
+	repo      *sqliterepo.Repository
+	agentMgr  *mockAgentManager
+	session   *models.TaskSession
+	step      *wfmodels.WorkflowStep
+	taskID    string
+	sessionID string
+	prompt    string
+
+	startEntered       chan struct{}
+	releaseStart       chan struct{}
+	startReturned      chan struct{}
+	secondStartEntered chan struct{}
+	processStarted     chan struct{}
+}
+
+func newWorkflowAsyncStartFailureFixture(
+	t *testing.T,
+	startupErr error,
+) *workflowAsyncStartFailureFixture {
+	t.Helper()
+	ctx := context.Background()
+	const (
+		taskID      = "workflow-async-task"
+		sessionID   = "workflow-async-session"
+		stepID      = "workflow-async-step"
+		profileID   = "workflow-async-profile"
+		executionID = "workflow-async-execution"
+	)
+
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateCreated)
+	task, err := repo.GetTask(ctx, taskID)
+	if err != nil {
+		t.Fatalf("get task: %v", err)
+	}
+	task.WorkflowStepID = stepID
+	if err := repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("update task workflow step: %v", err)
+	}
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("get session: %v", err)
+	}
+	session.AgentProfileID = profileID
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("update session profile: %v", err)
+	}
+	seedExecutorRunning(t, repo, sessionID, taskID, executionID)
+
+	step := &wfmodels.WorkflowStep{ID: stepID, WorkflowID: "wf1", Name: "Async work"}
+	stepGetter := newMockStepGetter()
+	stepGetter.steps[stepID] = step
+	taskRepo := newMockTaskRepo()
+	taskRepo.tasks[taskID] = &v1.Task{
+		ID:          taskID,
+		WorkflowID:  "wf1",
+		Title:       "Async workflow task",
+		Description: "task description",
+		State:       v1.TaskStateInProgress,
+	}
+
+	startEntered := make(chan struct{})
+	releaseStart := make(chan struct{})
+	startReturned := make(chan struct{})
+	secondStartEntered := make(chan struct{})
+	processStarted := make(chan struct{})
+	var startCalls atomic.Int32
+	agentMgr := &mockAgentManager{
+		repoForExecutionLookup: repo,
+		isAgentRunning:         true,
+		promptDone:             make(chan struct{}),
+		launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
+			return &executor.LaunchAgentResponse{AgentExecutionID: executionID, WorktreePath: req.RepositoryPath}, nil
+		},
+		startAgentProcessFunc: func(_ context.Context, _ string) error {
+			call := startCalls.Add(1)
+			if call == 1 {
+				close(startEntered)
+				<-releaseStart
+				close(startReturned)
+				return startupErr
+			}
+			if call == 2 {
+				close(secondStartEntered)
+				return nil
+			}
+			t.Errorf("unexpected startAgentProcess call %d", call)
+			return nil
+		},
+	}
+
+	svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
+	queueRepo := newWorkflowStartPromptSQLiteRepository(t, repo)
+	svc.messageQueue = messagequeue.NewService(queueRepo, messagequeue.DefaultMaxPerSession, testLogger())
+	svc.messageCreator = &mockMessageCreator{}
+	svc.turnService = &repoTurnService{repo: repo}
+	svc.executor.SetOnAgentStartFailed(svc.handleAgentStartFailed)
+	var processStartedOnce sync.Once
+	svc.executor.SetOnAgentProcessStarted(func(
+		callbackCtx context.Context, callbackTaskID, callbackSessionID, callbackExecutionID string,
+	) {
+		svc.handleAgentProcessStarted(callbackCtx, callbackTaskID, callbackSessionID, callbackExecutionID)
+		processStartedOnce.Do(func() { close(processStarted) })
+	})
+	svc.executor.SetOnAgentProcessStartFailed(svc.handleAgentProcessStartFailed)
+
+	return &workflowAsyncStartFailureFixture{
+		svc:                svc,
+		repo:               repo,
+		agentMgr:           agentMgr,
+		session:            session,
+		step:               step,
+		taskID:             taskID,
+		sessionID:          sessionID,
+		prompt:             "Run the async workflow step",
+		startEntered:       startEntered,
+		releaseStart:       releaseStart,
+		startReturned:      startReturned,
+		secondStartEntered: secondStartEntered,
+		processStarted:     processStarted,
+	}
+}
+
+func newWorkflowStartPromptAttemptFixture(
+	t *testing.T,
+	sessionState models.TaskSessionState,
+	executionID string,
+) (*Service, *sqliterepo.Repository, context.Context, context.Context, string) {
+	t.Helper()
+	ctx := context.Background()
+	const (
+		taskID    = "workflow-fence-task"
+		sessionID = "workflow-fence-session"
+		stepID    = "workflow-fence-step"
+	)
+
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, sessionState)
+	task, err := repo.GetTask(ctx, taskID)
+	if err != nil {
+		t.Fatalf("get task: %v", err)
+	}
+	task.WorkflowStepID = stepID
+	if err := repo.UpdateTask(ctx, task); err != nil {
+		t.Fatalf("update task workflow step: %v", err)
+	}
+	seedExecutorRunning(t, repo, sessionID, taskID, executionID)
+
+	agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
+	stepGetter := newMockStepGetter()
+	stepGetter.steps[stepID] = &wfmodels.WorkflowStep{ID: stepID, WorkflowID: "wf1", Name: "Fence step"}
+	taskRepo := newMockTaskRepo()
+	taskRepo.tasks[taskID] = &v1.Task{
+		ID:          taskID,
+		WorkflowID:  "wf1",
+		Title:       "Fence task",
+		Description: "fenced task",
+		State:       v1.TaskStateInProgress,
+	}
+	svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
+	queueRepo := newWorkflowStartPromptSQLiteRepository(t, repo)
+	svc.messageQueue = messagequeue.NewService(queueRepo, messagequeue.DefaultMaxPerSession, testLogger())
+	svc.turnService = &repoTurnService{repo: repo}
+	turnID := svc.startTurnForSession(ctx, sessionID)
+	if turnID == "" {
+		t.Fatal("expected active turn for preservation fixture")
+	}
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("get preservation session: %v", err)
+	}
+	queueIdentity, workflowEntry, workflowEntryCaptured, err := svc.captureWorkflowStartPromptAdmission(ctx, taskID, session)
+	if err != nil {
+		t.Fatalf("capture workflow launch admission: %v", err)
+	}
+	attempt := newWorkflowStartPromptAttemptWithAdmission(
+		taskID, sessionID,
+		workflowMessageOrigin{StepID: stepID, StepName: "Fence step"},
+		"fenced prompt", false, nil, nil, "", true, true,
+		queueIdentity, workflowEntry, workflowEntryCaptured,
+	)
+	return svc, repo, ctx, bindWorkflowStartPromptAttemptTurn(
+		withWorkflowStartPromptAttempt(ctx, attempt), turnID,
+	), turnID
+}

apps/backend/internal/task/repository/sqlite/step_transitions.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/jmoiron/sqlx"
 	"github.com/kandev/kandev/internal/db/dialect"
 	"github.com/kandev/kandev/internal/steptelemetry"
 )
@@ -162,6 +163,92 @@ func (r *Repository) GetLatestTaskStepTransitionID(ctx context.Context, taskID s
 	return id, nil
 }
 
+// EnsureCurrentTaskStepTransition returns the latest workflow-entry identity,
+// creating a durable current-entry row when an older database predates the
+// transition ledger. The task-row write guard serializes this backfill with
+// workflow moves, so a launch never captures a synthetic identity for an
+// obsolete step.
+func (r *Repository) EnsureCurrentTaskStepTransition(ctx context.Context, taskID string) (int64, error) {
+	tx, err := r.db.BeginTxx(ctx, nil)
+	if err != nil {
+		return 0, fmt.Errorf("begin workflow entry backfill: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	entryID, err := r.ensureCurrentTaskStepTransitionTx(ctx, tx, taskID)
+	if err != nil {
+		return 0, err
+	}
+	if err := tx.Commit(); err != nil {
+		return 0, err
+	}
+	return entryID, nil
+}
+
+func (r *Repository) ensureCurrentTaskStepTransitionTx(
+	ctx context.Context,
+	tx *sqlx.Tx,
+	taskID string,
+) (int64, error) {
+	// Acquire the same task-row write boundary used by queue admission. SQLite
+	// otherwise allows a read transaction to observe the task before a
+	// concurrent writer commits its move, while PostgreSQL obtains the row lock
+	// through the same UPDATE and its FOR UPDATE read below.
+	guard, err := tx.ExecContext(ctx, r.db.Rebind(`
+		UPDATE tasks SET updated_at = updated_at WHERE id = ?
+	`), taskID)
+	if err != nil {
+		return 0, fmt.Errorf("guard task for workflow entry backfill: %w", err)
+	}
+	rows, err := guard.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("workflow entry backfill task rows affected: %w", err)
+	}
+	if rows == 0 {
+		return 0, fmt.Errorf("%w: %s", ErrTaskNotFound, taskID)
+	}
+
+	var latestID int64
+	err = tx.QueryRowContext(ctx, r.db.Rebind(`
+		SELECT id
+		FROM task_step_transitions
+		WHERE task_id = ?
+		ORDER BY id DESC
+		LIMIT 1
+	`), taskID).Scan(&latestID)
+	if err != nil && err != sql.ErrNoRows {
+		return 0, fmt.Errorf("read workflow entry during backfill: %w", err)
+	}
+	if err == nil && latestID > 0 {
+		if err := tx.Commit(); err != nil {
+			return 0, err
+		}
+		return latestID, nil
+	}
+
+	workflowID, workflowStepID, found, err := r.readTaskStepInTx(ctx, tx, taskID)
+	if err != nil {
+		return 0, err
+	}
+	if !found || workflowID == "" || workflowStepID == "" {
+		if err := tx.Commit(); err != nil {
+			return 0, err
+		}
+		return 0, nil
+	}
+
+	backfilledID, err := r.recordStepTransition(ctx, tx, stepTransitionInput{
+		taskID:           taskID,
+		toWorkflowID:     workflowID,
+		toWorkflowStepID: workflowStepID,
+		occurredAt:       time.Now().UTC(),
+	})
+	if err != nil {
+		return 0, fmt.Errorf("write workflow entry backfill: %w", err)
+	}
+	return backfilledID, nil
+}
+
 // CountStepEntries returns the number of committed task_step_transitions
 // rows whose to_workflow_step_id is stepID for taskID — the recorded entry
 // count REQ-TWS-001 floors at 1 to derive the step-entry number. Recorded