From 0b41329dc352b7e9b5b847287d101c737f1b41a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Fri, 18 Sep 2026 00:21:57 +0100 Subject: [PATCH] fix: preserve workflow prompts after async startup failure (#3753) (#3768) --- .../internal/orchestrator/dynamic_launch.go | 3 +- .../orchestrator/event_handlers_agent.go | 38 + .../orchestrator/event_handlers_workflow.go | 174 +++- .../messagequeue/repository_memory.go | 30 + .../messagequeue/repository_sqlite.go | 118 ++- .../orchestrator/messagequeue/service.go | 150 +++- .../orchestrator/messagequeue/types.go | 14 + .../internal/orchestrator/task_operations.go | 28 +- .../orchestrator/workflow_start_prompt.go | 484 +++++++++++ .../workflow_start_prompt_admission_test.go | 126 +++ ...kflow_start_prompt_effective_input_test.go | 265 ++++++ .../workflow_start_prompt_fencing_test.go | 279 +++++++ .../workflow_start_prompt_metadata_test.go | 100 +++ .../workflow_start_prompt_test.go | 757 ++++++++++++++++++ .../repository/sqlite/step_transitions.go | 87 ++ .../plan.md | 147 ++++ .../task-01-preserve-workflow-launch-input.md | 157 ++++ .../plan.md | 9 +- .../workflow-step-agent-start-ownership.md | 45 +- .../workflow-step-agent-start-ownership.md | 73 ++ 20 files changed, 3024 insertions(+), 60 deletions(-) create mode 100644 apps/backend/internal/orchestrator/workflow_start_prompt.go create mode 100644 apps/backend/internal/orchestrator/workflow_start_prompt_admission_test.go create mode 100644 apps/backend/internal/orchestrator/workflow_start_prompt_effective_input_test.go create mode 100644 apps/backend/internal/orchestrator/workflow_start_prompt_fencing_test.go create mode 100644 apps/backend/internal/orchestrator/workflow_start_prompt_metadata_test.go create mode 100644 apps/backend/internal/orchestrator/workflow_start_prompt_test.go create mode 100644 docs/plans/workflow-async-start-prompt-preservation/plan.md create mode 100644 docs/plans/workflow-async-start-prompt-preservation/task-01-preserve-workflow-launch-input.md diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index 979f33bffe..8d142b5508 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index 6dec8f6c33..7bcb27ad66 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/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", diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 10bf4df0fd..870ca1d3f0 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go index 5efd1187c8..29f2d22a9d 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go index 6d065a0c7e..3aae5c2ced 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go +++ b/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 } diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index 370fb308fa..06d08c93d3 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/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", diff --git a/apps/backend/internal/orchestrator/messagequeue/types.go b/apps/backend/internal/orchestrator/messagequeue/types.go index c616dd5156..1625406414 100644 --- a/apps/backend/internal/orchestrator/messagequeue/types.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 4b7564e196..de616f3b83 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/workflow_start_prompt.go b/apps/backend/internal/orchestrator/workflow_start_prompt.go new file mode 100644 index 0000000000..7d69913624 --- /dev/null +++ b/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 +} diff --git a/apps/backend/internal/orchestrator/workflow_start_prompt_admission_test.go b/apps/backend/internal/orchestrator/workflow_start_prompt_admission_test.go new file mode 100644 index 0000000000..e8c4050f6a --- /dev/null +++ b/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) + } +} diff --git a/apps/backend/internal/orchestrator/workflow_start_prompt_effective_input_test.go b/apps/backend/internal/orchestrator/workflow_start_prompt_effective_input_test.go new file mode 100644 index 0000000000..336151e773 --- /dev/null +++ b/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)) + } +} diff --git a/apps/backend/internal/orchestrator/workflow_start_prompt_fencing_test.go b/apps/backend/internal/orchestrator/workflow_start_prompt_fencing_test.go new file mode 100644 index 0000000000..7c6626f862 --- /dev/null +++ b/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, + ) +} diff --git a/apps/backend/internal/orchestrator/workflow_start_prompt_metadata_test.go b/apps/backend/internal/orchestrator/workflow_start_prompt_metadata_test.go new file mode 100644 index 0000000000..afe6dc1081 --- /dev/null +++ b/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) + } +} diff --git a/apps/backend/internal/orchestrator/workflow_start_prompt_test.go b/apps/backend/internal/orchestrator/workflow_start_prompt_test.go new file mode 100644 index 0000000000..56cda0cfe2 --- /dev/null +++ b/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 +} diff --git a/apps/backend/internal/task/repository/sqlite/step_transitions.go b/apps/backend/internal/task/repository/sqlite/step_transitions.go index 55bcc057ce..510f3da0c1 100644 --- a/apps/backend/internal/task/repository/sqlite/step_transitions.go +++ b/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 diff --git a/docs/plans/workflow-async-start-prompt-preservation/plan.md b/docs/plans/workflow-async-start-prompt-preservation/plan.md new file mode 100644 index 0000000000..3f76f1d9d1 --- /dev/null +++ b/docs/plans/workflow-async-start-prompt-preservation/plan.md @@ -0,0 +1,147 @@ +--- +created: 2026-09-17 +status: complete +requirements: + - REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005 + - REQ-TASKS-TASK-LAUNCH-FAILURE-RECOVERY-001 +system_design: + - ../../specs/tasks/system-design/workflow-step-agent-start-ownership.md + - ../../specs/tasks/system-design/task-launch-failure-recovery.md +legacy_specs: [] +--- + +# Implementation Plan: Workflow asynchronous start prompt preservation + +## Overview + +Preserve an automatic workflow prompt after asynchronous startup failure, then deliver it through existing recovery and queue admission. +One implementation work order owns the complete backend path and its regression evidence. +The backend implementation and regression evidence are complete in this workspace. + +## Evidence and root cause + +Source: [issue #3753](https://github.com/kdlbs/kandev/issues/3753), with no comments or image attachments at investigation time. +Investigated checkout: `f32eb373781bcec12583c37e4a9d52b1ee12072f`. + +The read-only control-flow trace confirms the prompt-loss path: + +1. `autoStartStepPrompt` records the workflow prompt and calls `startCreatedSessionWithComposedPrompt` for a `CREATED` session. +2. `startAgentOnExistingWorkspaceWithRequest` starts the prepared execution asynchronously and returns a successful `TaskExecution`. +3. The caller therefore skips `handleCreatedAutoStartLaunchFailure`, which owns synchronous prompt preservation. +4. A later `StartAgentProcess` error reaches `handleAgentProcessStartFailure` and `handleAgentStartFailed` without the workflow prompt envelope. +5. Those handlers classify the error and settle state, but neither persists that workflow input in the queue. +6. `RecoverTaskLaunch` calls `RecoverSession(..., "fresh_start")`, which resumes the same session without reconstructing this missing queue entry. + +The original [start-ownership package](../workflow-step-agent-start-ownership/plan.md) explicitly excluded this asynchronous case. +Its completed Task 02 covers synchronous busy errors only. +Current launch-recovery code already persists typed bootstrap errors and guards successor executions. +Thus, the source confirms prompt loss, but does not establish the reporter's exact idle-state sequence on this newer checkout. +No access to the reporter's runtime or private logs was required. + +## Scope + +### In scope + +- Extend the existing workflow owner with `REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005`. +- Capture the exact pending workflow input before asynchronous launch admission. +- Preserve it once after an eligible failure, without starting another launch from that callback. +- Prove explicit recovery, boot-ready drain, metadata preservation, and stale-event rejection. + +### Out of scope + +- Provider start failures themselves, ACP reset support, and the agentctl running-process guard. +- New UI, automatic retry policy, historical orphan repair, and prompt replay after ambiguous provider acceptance. +- A crash before callback persistence, a new prompt ledger, schema migrations, and changes to Office scheduling. + +## Technical approach + +The [owning design](../../specs/tasks/system-design/workflow-step-agent-start-ownership.md#asynchronous-launch-prompt-preservation) +defines attempt capture, guarded queue persistence, and recovery. + +Add the private envelope helper in `apps/backend/internal/orchestrator/workflow_start_prompt.go`. +Wire capture through `event_handlers_workflow.go` and `task_operations.go`. +Consume it through the accepted path in `event_handlers_agent.go`. +Separate queue persistence from automatic resume scheduling in `event_handlers_workflow.go`. +Preserve existing executor callbacks and dynamic route/ceiling bookkeeping. +Change `executor/executor_execute.go` only if propagation or final-attempt ordering needs correction. + +The existing queue owns durable input after preservation. Context holds only the in-flight launch snapshot. +Existing queue ownership and launch-error decisions remain authoritative, so this package introduces no separate architectural decision. + +## Tests + +New file: `apps/backend/internal/orchestrator/workflow_start_prompt_test.go`. + +| Acceptance | Planned test | +| --- | --- | +| 005.1, 005.2 | `TestWorkflowAsyncStartFailure_PreservesPromptThroughRecovery` | +| 005.1, 005.2 | `TestWorkflowAsyncStartFailure_PreservesPlanOnlyInputThroughRecovery`, `TestWorkflowAsyncStartFailure_PreservesConfigOnlyInputThroughRecovery` | +| 005.1, 005.2 | `TestWorkflowAsyncStartFailure_ExcludesTrulyEmptyInput` | +| 005.1, 005.2 | `TestWorkflowAsyncStartFailure_PreservesInputMetadata` | +| 005.3 | `TestWorkflowAsyncStartFailure_RejectsStaleAndDuplicateCallbacks`, including a committed move between the ownership read and queue insertion and leave-and-return to the same step | +| 005.3, 005.4 | `TestWorkflowAsyncStartFailure_DynamicFallbackOwnership` | +| 005.4, 005.5 | `TestWorkflowAsyncStartFailure_QueueFailureKeepsLaunchError` | +| 005.2, 005.5 | `TestWorkflowAsyncStartFailure_PersistedQueueSurvivesRestart` | +| 005.6 | `TestWorkflowAsyncStartFailure_SuccessAndSynchronousRejection` | + +Before implementation, the first test failed because the queue remained empty +after asynchronous failure. The completed implementation now preserves the +entry and exercises delivery through recovery. +Existing executor bootstrap tests cover `AC-TASKS-TASK-LAUNCH-FAILURE-RECOVERY-001.11` and `.12`. +Retain those checks for safe error projection and successor fencing. + +## End-to-end evidence + +The first regression uses real orchestrator and queue services with a controllable fake provider. +It drives workflow entry, delayed startup failure, the existing recovery action, boot-ready delivery, and the captured provider prompt. +This backend service flow covers the user outcome without a new browser interface or artificial browser fixture. +It must assert that no manual prompt is necessary after explicit launch recovery. + +## Work orders + +- [x] [Task 01: Preserve asynchronous workflow launch input](task-01-preserve-workflow-launch-input.md) + +Execution is sequential. No subagents are authorized. + +## Verification results + +Planning evidence: read-only source trace confirmed the missing ownership transfer. +The existing synchronous regression passed: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/orchestrator -run '^TestAutoStartCreatedLaunch_QueuesPromptWhenAgentAlreadyRunning$' -count=1) +``` + +This test covers the earlier synchronous repair, not the proposed asynchronous fix. +Implementation regressions pass, including asynchronous recovery delivery, +effective plan-only and config-only input, true-empty exclusion, metadata +preservation, stale-callback fencing, the atomic move-admission barrier, +same-step re-entry fencing, queue failure handling, dynamic ownership, restart +persistence, and synchronous rejection behavior. Workflow entry admission +captures the launch-time task transition, session incarnation, and lifecycle +generation, then validates them inside the shared task/queue transaction. The +queue retains the raw prompt and mode metadata, so recovery composes each +effective prompt exactly once. + +Package checks: + +```bash +python3 scripts/list-docs.py validate +python3 scripts/lint-spec-files.test.py +python3 scripts/lint-spec-files.py --all +git diff --check -- docs/specs docs/plans/workflow-async-start-prompt-preservation +git status --short -- docs/plans/workflow-async-start-prompt-preservation +``` + +Results on 2026-09-17: catalog validation passed for 285 decisions and 986 specifications. +All 36 linter tests passed. All specification files passed, and the diff whitespace check passed. +The implementation package is included in this pull request. +GitHub confirmed `carlosflorencio` as the issue assignee. + +## Risks + +- Process-start failure callbacks can precede launch return or arrive after a successor starts. +- Dynamic profile fallback must not queue input while another candidate owns the same launch. +- Calling the existing queue helper unchanged can schedule recovery before failed-process cleanup. +- A full queue or storage error prevents preservation. The existing launch error must remain visible. +- Restart protection begins after queue persistence. This package does not claim crash-safe admission before that boundary. diff --git a/docs/plans/workflow-async-start-prompt-preservation/task-01-preserve-workflow-launch-input.md b/docs/plans/workflow-async-start-prompt-preservation/task-01-preserve-workflow-launch-input.md new file mode 100644 index 0000000000..aeba2e86fb --- /dev/null +++ b/docs/plans/workflow-async-start-prompt-preservation/task-01-preserve-workflow-launch-input.md @@ -0,0 +1,157 @@ +--- +id: "01-preserve-workflow-launch-input" +title: "Preserve asynchronous workflow launch input" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005 + - REQ-TASKS-TASK-LAUNCH-FAILURE-RECOVERY-001 +acceptance_criteria: + - AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.1 + - AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.2 + - AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.3 + - AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.4 + - AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.5 + - AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.6 +system_design: + - ../../specs/tasks/system-design/workflow-step-agent-start-ownership.md + - ../../specs/tasks/system-design/task-launch-failure-recovery.md +--- + +# Task 01: Preserve asynchronous workflow launch input + +## Summary + +Carry immutable workflow input across launch admission and persist it when that asynchronous launch fails before delivery. +Keep existing recovery actions, queue policy, failure projection, and executor cleanup authoritative. + +## In scope + +- Add the private attempt envelope and retire it on success or synchronous rejection. +- Bind task, session, execution evidence, turn, workflow entry, and final dynamic attempt ownership. +- Persist raw queue-form input once through the accepted failure path, without immediate auto-resume scheduling. +- Retain recorded-message identity, attachment ownership, references, handoff, origin, and plan mode. +- Add all tests named in the plan, including failure-to-recovery service coverage. + +## Out of scope + +New UI, public APIs, migrations, provider retries, and the existing post-start transient error mechanism. + +## Acceptance + +1. An asynchronous failure after successful launch return preserves the exact input and recovery delivers it once with one transcript row. +2. Stale, duplicate, terminal, archived, deleted, and superseded-entry callbacks cannot enqueue input or affect successor work. +3. Error projection, paused queues, synchronous failures, dynamic fallback, and success retain their established behavior. + +## Regression procedure + +1. Seed a `CREATED` session with a prepared execution and a step with reset plus automatic start. +2. Block the fake `StartAgentProcess` call with channels. +3. Drive workflow entry through the real service and prove the launch call returned without error. +4. Release the provider with a typed pre-delivery startup failure. +5. Await failure handling through explicit synchronization, then assert one preserved queue entry. +6. Before implementation, observe the expected red result: zero preserved + entries. +7. Implement the ownership transfer, then recover through `RecoverTaskLaunch` and the normal boot-ready path. +8. Assert one provider prompt, one user message, and removal of the consumed queue entry. + +Add table cases for generic, authentication, and managed-runtime bootstrap errors. +Exercise duplicate callback delivery and a failure before launch return. +Race replacement execution, same-execution successor turn, cancellation, completion, archive, deletion, and a later workflow entry. +Exercise a dynamic fallback candidate followed by success and a final failed candidate. +Cover combined handoff, attachment-only input, entity references, plan mode, and failed original message persistence. +Cover queue-full/write errors, a paused queue, and reopening the queue repository after preservation. +Verify no replay from successful startup or post-start prompt failure. + +## Verification + +Run these commands from the repository root after the TDD cycle: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/orchestrator -race -run 'TestWorkflowAsyncStartFailure|TestAutoStartCreatedLaunch|TestAutoStartTransientError_BootReadyDrainsOrphanedQueue|TestProcessOnEnterResetAgentContext' -count=1) +(cd apps/backend && go test -tags fts5 ./internal/orchestrator/executor -race -run 'TestStartAgentProcessAsync|TestBootstrapFailure|TestBuildBootstrapLastAgentError' -count=1) +(cd apps/backend && go test -tags fts5 ./internal/orchestrator/messagequeue -race -count=1) +python3 scripts/list-docs.py validate +python3 scripts/lint-spec-files.py --all +git diff --check +``` + +## Files likely touched + +- `apps/backend/internal/orchestrator/workflow_start_prompt.go` (new) +- `apps/backend/internal/orchestrator/workflow_start_prompt_test.go` (new) +- `apps/backend/internal/orchestrator/event_handlers_workflow.go` +- `apps/backend/internal/orchestrator/task_operations.go` +- `apps/backend/internal/orchestrator/event_handlers_agent.go` +- `apps/backend/internal/orchestrator/executor/executor_execute.go` (only if callback propagation requires correction) +- `apps/backend/internal/orchestrator/executor/workflow_start_prompt_test.go` (new, if executor behavior changes) + +## Dependencies + +None. Existing synchronous preservation and typed bootstrap errors are present in the investigated checkout. + +## Risks + +The accepted failure path must preserve input before recoverable failure routing can expose a promptable session. +Guard acquisition must not deadlock the launch caller, callback, or queue drain. +Do not infer current ownership from mutable session-only caches. +Do not replace existing dynamic-route or capacity callbacks with the prompt-preservation callback. + +## Parallelism + +`sequential` + +## Inputs + +- [Requirement 005](../../specs/tasks/requirements/workflow-step-agent-start-ownership.md) +- [Asynchronous preservation design](../../specs/tasks/system-design/workflow-step-agent-start-ownership.md#asynchronous-launch-prompt-preservation) +- [Launch-error design](../../specs/tasks/system-design/task-launch-failure-recovery.md) +- Existing patterns: `event_handlers_duplicate_autostart_test.go`, `executor/executor_launch_failure_classification_test.go`, and `task_launch_recovery_test.go`. +- Existing queue rules: [server-owned auto-run](../../decisions/2026-08-16-server-owned-queue-auto-run.md). + +## Results + +Implemented the private immutable workflow-start attempt envelope and bound it +through initial-turn creation, prepared launch, dynamic process-start success, +and accepted asynchronous startup failure callbacks. Preserved input uses the +existing queue metadata and storage path, claims once, leaves auto-run policy +unchanged, and does not schedule a replacement from the failure callback. + +Added service regressions for recovery delivery, effective plan-only and +config-only input, true-empty exclusion, duplicate and stale callback fencing, +metadata and failed-transcript preservation, classified bootstrap errors, +queue write and capacity failures, dynamic route ownership, paused queue +restart persistence, successful startup, and synchronous permanent rejection. +The stale-callback suite also commits a task move between the pre-admission +ownership read and queue insertion, and leaves and returns to the same step; +both cases assert that no obsolete queue row or dispatch is produced. The +production queue path captures the launch-time transition, session +incarnation, and lifecycle generation and validates them inside the shared +task/queue transaction. In-memory repositories fail closed for this strict +admission instead of claiming an atomic fence they cannot provide. + +The effective-input regressions preserve empty raw queue content with mode +metadata, deliver one provider prompt, and assert one matching transcript +row/system block after recovery. A launch with no raw content, attachments, +references, handoff, plan mode, or session configuration remains excluded. +Admission capture failures now abort a prepared launch before the provider is +started, while legacy tasks without a transition row receive a durable current +entry before capture. Queue drains retain reserved entries when session +metadata reads fail, and recovery uses launch-time effective-input and +config-mode metadata rather than reclassifying a mode-only prompt from the +mutable session projection. + +Verification passed: + +```text +(cd apps/backend && go test -tags fts5 ./internal/orchestrator -race -run 'TestWorkflowAsyncStartFailure|TestAutoStartCreatedLaunch|TestAutoStartTransientError_BootReadyDrainsOrphanedQueue|TestProcessOnEnterResetAgentContext' -count=1) +(cd apps/backend && go test -tags fts5 ./internal/orchestrator -race -run 'TestCaptureWorkflowStartPromptAdmission|TestDispatchTakenQueuedMessage_RetainsEntryWhenInputReadFails' -count=1) +(cd apps/backend && go test -tags fts5 ./internal/orchestrator/executor -race -run 'TestStartAgentProcessAsync|TestBootstrapFailure|TestBuildBootstrapLastAgentError' -count=1) +(cd apps/backend && go test -tags fts5 ./internal/orchestrator/messagequeue -race -count=1) +make -C apps/backend build +python3 scripts/list-docs.py validate +python3 scripts/lint-spec-files.py --all +git diff --check +``` diff --git a/docs/plans/workflow-step-agent-start-ownership/plan.md b/docs/plans/workflow-step-agent-start-ownership/plan.md index 7999eb048d..232033dc5d 100644 --- a/docs/plans/workflow-step-agent-start-ownership/plan.md +++ b/docs/plans/workflow-step-agent-start-ownership/plan.md @@ -6,6 +6,10 @@ status: complete # Implementation Plan: Workflow Step Agent Start Ownership +The asynchronous gap excluded from this completed package now has an implemented +[follow-up package](../workflow-async-start-prompt-preservation/plan.md) for issue #3753. +The original implementation scope and results remain unchanged. + ## Overview Two small, ordered changes in `internal/orchestrator/event_handlers_workflow.go`. @@ -90,9 +94,8 @@ rejections are not queued. The queued prompt is drained by the existing `TestAutoStartTransientError_BootReadyDrainsOrphanedQueue` already pins, and the error is still returned so `FAILED` state and surfacing are unchanged. -The async path remains uncovered — see the spec's **Known gap**. Task 01 -removes the only known trigger for it on a first-turn launch, so the incident -cannot recur through it. +The async path is implemented and verified in the issue #3753 follow-up package +linked above. The original synchronous scope and its evidence remain unchanged. ## Frontend diff --git a/docs/specs/tasks/requirements/workflow-step-agent-start-ownership.md b/docs/specs/tasks/requirements/workflow-step-agent-start-ownership.md index 9139357e41..ac63a8c7c3 100644 --- a/docs/specs/tasks/requirements/workflow-step-agent-start-ownership.md +++ b/docs/specs/tasks/requirements/workflow-step-agent-start-ownership.md @@ -2,7 +2,7 @@ status: draft system: tasks created: 2026-08-05 -updated: 2026-09-12 +updated: 2026-09-17 owners: - Kandev --- @@ -97,6 +97,29 @@ intent-derived destination. - **AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-004.4:** Desktop and mobile task creation shall apply the same immediate-launch placement rule. +### REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005: Automatic prompt preservation after startup failure + +**Intent:** Retain an automatic workflow prompt when its agent launch fails before prompt delivery. +This draft extension addresses [issue #3753](https://github.com/kdlbs/kandev/issues/3753). + +#### Acceptance criteria + +- **AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.1:** When an admitted automatic launch fails asynchronously before prompt delivery, the system shall preserve its prompt for recovery. + The preserved input shall include handoff text, attachments, references, plan mode, and workflow origin. +- **AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.2:** After successful recovery, the system shall deliver the preserved prompt once through normal queue admission. + It shall retain one existing user message and respect a paused queue. +- **AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.3:** Duplicate or superseded launch failures shall not duplicate queued input or affect a successor turn. + Cancellation, completion, archive, deletion, and a superseding workflow entry shall prevent stale prompt recovery. +- **AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.4:** Prompt preservation shall retain the existing launch-error classification and recovery actions. + It shall not initiate another launch from the failure callback or turn permanent rejection into an automatic retry loop. +- **AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.5:** After successful queue persistence, the preserved prompt shall survive a backend restart. + If persistence fails, the launch error shall remain visible and diagnostics shall identify the preservation failure without exposing prompt content. +- **AC-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005.6:** Successful starts and synchronous permanent rejections shall not create asynchronous recovery entries. + Existing synchronous busy-error recovery shall retain its current behavior. + +This extension excludes a backend crash before the asynchronous failure callback persists the prompt. +It also excludes replay after ambiguous provider acceptance, automatic repair of historical orphaned messages, and new recovery controls. + ## Migrated source detail ## Why @@ -188,21 +211,13 @@ and unprompted. **WHEN** the failure is handled, **THEN** the error is returned and no message is queued. -## Known gap +## Asynchronous preservation delivery status -Prompt preservation across an **asynchronous** start failure is not covered -here. `startAgentOnExistingWorkspace` ends with `startAgentProcessAsync` and -returns `nil`, so a session with a prepared execution — the incident's own shape -— never surfaces a synchronous error for the clause above to catch. That failure -is handled entirely by `handleAgentProcessStartFailure` → -`Service.handleAgentStartFailed`, which has no access to the prompt the launch -was carrying and therefore cannot re-queue it. - -Task 01 removes the only known trigger for that path on a first-turn launch, so -the incident cannot recur through it. The residual gap — any *other* async start -failure on a first-turn launch drops the step prompt — is a distinct pre-existing -defect. Closing it needs a pending-prompt handle the async failure path can -read, which is a design change beyond this repair. +The original [start-ownership package](../../../plans/workflow-step-agent-start-ownership/plan.md) +closed the synchronous failure path. Requirement +`REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005` defines the asynchronous +recovery contract. The [asynchronous prompt preservation package](../../../plans/workflow-async-start-prompt-preservation/plan.md) +delivers it through its completed work order. ## Out of scope diff --git a/docs/specs/tasks/system-design/workflow-step-agent-start-ownership.md b/docs/specs/tasks/system-design/workflow-step-agent-start-ownership.md index 05159c88c1..d71650dc19 100644 --- a/docs/specs/tasks/system-design/workflow-step-agent-start-ownership.md +++ b/docs/specs/tasks/system-design/workflow-step-agent-start-ownership.md @@ -6,6 +6,7 @@ requirements: - REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-002 - REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-003 - REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-004 + - REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005 --- # Workflow Step Agent Start Ownership System Design @@ -26,6 +27,7 @@ The design preserves runtime configuration through the existing reset contract. | `REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-002` | [Active-turn reset flow](#active-turn-reset-flow), [Bounded predecessor wait](#bounded-predecessor-wait), [Reset failure containment](#reset-failure-containment) | | `REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-003` | [Prompt fallback ownership](#prompt-fallback-ownership), [Prompt-history contract](#prompt-history-contract), [Workflow-entry prompt flow](#workflow-entry-prompt-flow) | | `REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-004` | [Creation destination routing](#creation-destination-routing) | +| `REQ-TASKS-WORKFLOW-STEP-AGENT-START-OWNERSHIP-005` | [Asynchronous launch prompt preservation](#asynchronous-launch-prompt-preservation) | ## Components and responsibilities @@ -269,6 +271,77 @@ The prompt counter and fallback claim are durable across backend restarts. A restart cannot make an earlier task description eligible for another fallback dispatch, and deleting/recreating a session ID starts a new prompt boundary. +## Asynchronous launch prompt preservation + +This extension belongs to the task system because workflow entry owns the pending input. +The [implementation package](../../../plans/workflow-async-start-prompt-preservation/plan.md) +delivers the design and its regression matrix. + +### Capture before asynchronous admission + +`autoStartStepPrompt` already separates recorded content, agent content, references, attachments, and completion handoff text. +Capture the queue-form input before `startCreatedSessionWithComposedPrompt` admits the launch. +Use a private, immutable attempt envelope in the orchestrator, carried through the launch context. +Existing executor context propagation preserves values through `context.WithoutCancel` and dynamic launch paths. +The executor need not interpret the prompt envelope or import workflow composition types. + +The envelope includes task, session, workflow step, unique launch token, initial turn identity, and the queue-form input. +Bind the turn identity in `startCreatedSession` before `launchPreparedSessionWithDynamicFallback` starts asynchronous work. +Copy mutable slices and metadata. Preserve the actual `userMsgRecorded` result rather than assuming that transcript persistence succeeded. +Do not recover from the mutable `lastTurnPrompt` cache or select the latest transcript row. +Neither source binds all workflow metadata to the failed attempt. + +An attempt-local completion claim permits one preservation operation. +Synchronous rejection retires the envelope and retains `handleCreatedAutoStartLaunchFailure` as its sole queue owner. +Success retires the envelope without queue insertion. Context lifetime bounds retention after the launch callback returns. +Dynamic fallback attempts must not independently preserve the same logical prompt while another candidate can still start. +Only the final accepted startup failure can consume the preservation claim. + +### Preserve through the accepted failure path + +Extend `handleAgentStartFailed` after its cancellation, terminal-session, and execution checks. +Under the existing cancellation guard, validate the captured turn and workflow entry before queue persistence. +An execution ID alone is insufficient when successive attempts reuse a prepared execution. +Reject superseded attempts, completed or cancelled sessions, archived tasks, and missing sessions. +Queue admission must retain the existing session-incarnation and purge-generation protections. +Do not treat the unconditional `onAgentProcessStartFailed` notification as evidence that a failure passed these guards. + +Persist the envelope using the existing message-queue service and workflow metadata format. +Extract persistence from `queueAutoStartPrompt` so this call does not run `scheduleAutoResumeForWorkflowQueue`. +Preservation must not start a replacement before failure cleanup finishes. +Leave the user's `auto_run` policy unchanged. +After explicit recovery, existing boot-ready admission can drain the entry when all eligibility checks pass. + +The queue contains the composed visible workflow input, with entity references and completion handoff in their existing metadata fields. +It must not contain a second injected Kandev system block or independently restored copy of an already-merged handoff. +`executeQueuedMessage` continues to honor `user_message_recorded` and compose dispatch context through the existing path. +The queue uses its existing storage, attachment ownership, capacity limits, transfer rules, and restart recovery. +No new table, public payload, or provider API is required. + +### Failure and recovery boundaries + +The original error continues through auth, managed-runtime, or generic bootstrap handling. +The [launch recovery design](task-launch-failure-recovery.md) remains authoritative for safe errors, stamps, history, and recovery authorization. +`RecoverTaskLaunch` already routes session-owned retry through `RecoverSession` on the same session. +Recovery boots the session and lets the existing queue drain send its preserved input. +Do not send the same prompt as both a launch description and a queue entry. + +A queue write failure does not mask the original startup failure. +Log a separate bounded preservation diagnostic with task, session, execution, and attempt identity, without input text or attachment contents. +The existing persistent launch error remains the visible recovery signal. +No frontend layout or copy changes are required by this package. + +The guarantee starts when a live failure callback accepts ownership and persists the queue entry. +A process crash before that point is excluded. Successful process startup is not proof of exactly-once provider execution after an ambiguous prompt error. +Post-start prompt failures keep their separate correlated terminal path and do not use this replay mechanism. + +### Verification + +Use controlled startup barriers to prove preservation after the launch call already returned success. +Cover duplicate callbacks, replacement execution, same-execution successor turn, terminal races, and dynamic fallback ownership. +Exercise real queue persistence and the service recovery-to-boot-ready-to-dispatch path. +Verify one transcript row, one delivered prompt, preserved metadata, paused queues, and restart after queue persistence. + ## Observability The workflow reset logs the task, session, workflow step, and execution identifiers. Internal cancellation uses the existing cancellation and escalation logs.