From bfdfcdef8ea39bc6aeede590c1b026628ef47047 Mon Sep 17 00:00:00 2001 From: Henry Date: Fri, 11 Sep 2026 04:30:10 +0800 Subject: [PATCH] fix(backend): prevent duplicate or stranded launches when tasks auto-start on create (#3529) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .../deferred_launch_consume_test.go | 2 +- .../event_handlers_dependencies.go | 31 +- .../orchestrator/event_handlers_github.go | 9 + .../event_handlers_github_review_test.go | 6 +- .../orchestrator/event_handlers_workflow.go | 505 ++++++++++++++-- ...dlers_workflow_auto_start_inflight_test.go | 131 +++++ ...dlers_workflow_auto_start_recovery_test.go | 552 ++++++++++++++++++ apps/backend/internal/orchestrator/service.go | 8 + .../internal/orchestrator/session_launch.go | 16 +- apps/backend/internal/task/models/models.go | 15 + 10 files changed, 1195 insertions(+), 80 deletions(-) create mode 100644 apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_inflight_test.go create mode 100644 apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_recovery_test.go diff --git a/apps/backend/internal/orchestrator/deferred_launch_consume_test.go b/apps/backend/internal/orchestrator/deferred_launch_consume_test.go index 92cdeacf4d..e16bc7ee9d 100644 --- a/apps/backend/internal/orchestrator/deferred_launch_consume_test.go +++ b/apps/backend/internal/orchestrator/deferred_launch_consume_test.go @@ -564,7 +564,7 @@ func TestQueuePromotionOfBlockedWIPOverflowDoesNotLaunch(t *testing.T) { // resume the still-pending promotion lifecycle rather than leave the token // stranded after launching directly. dependencies.blocked = false - svc.autoStartTaskForStep(ctx, deferredChainTaskID, "step-limited", "task.dependencies_resolved", 0) + svc.autoStartTaskForStep(ctx, deferredChainTaskID, "step-limited", "task.dependencies_resolved", 0, false) require.True(t, counter.awaitLaunch(0), "resolved dependency did not launch the promoted task") awaitLaunchedSession(t, repo, deferredChainTaskID) started, err := repo.GetTask(ctx, deferredChainTaskID) diff --git a/apps/backend/internal/orchestrator/event_handlers_dependencies.go b/apps/backend/internal/orchestrator/event_handlers_dependencies.go index 397af9dcdc..3199fb1924 100644 --- a/apps/backend/internal/orchestrator/event_handlers_dependencies.go +++ b/apps/backend/internal/orchestrator/event_handlers_dependencies.go @@ -31,26 +31,33 @@ func (s *Service) SetTaskDependencyReader(reader TaskDependencyReader) { } // dependencyBlocksAutoStart reports whether taskID has an unresolved dependency -// and therefore must not be launched by an automated path. +// and therefore must not be launched by an automated path. gateErrored reports +// whether the block came from a failed read rather than a genuine dependency +// block: a genuine block is already covered by dependency-resolution launch +// paths (evaluateDependentAfterPredecessorChange, reconcileDependencyLaunchesOnStartup), +// but a read failure is not, so a caller holding a one-shot launch token must +// restore it only in that case — restoring on every genuine block would burn +// recoverTaskLifecycleAttempt's bounded retry budget on every startup for as +// long as the task stays blocked. // -// Fails CLOSED: a read error returns true. Failing open would launch work whose -// predecessor may never have run, which is the single outcome task dependencies -// exist to prevent. -func (s *Service) dependencyBlocksAutoStart(ctx context.Context, taskID, eventName string) bool { +// Fails CLOSED: a read error returns blocked=true. Failing open would launch +// work whose predecessor may never have run, which is the single outcome task +// dependencies exist to prevent. +func (s *Service) dependencyBlocksAutoStart(ctx context.Context, taskID, eventName string) (blocked, gateErrored bool) { if s.dependencyReader == nil { - return false + return false, false } - blocked, reason, err := s.dependencyReader.DependencyGate(ctx, taskID) + isBlocked, reason, err := s.dependencyReader.DependencyGate(ctx, taskID) if err != nil { s.logger.Warn(eventName+": dependency lookup failed; skipping auto-start", zap.String("task_id", taskID), zap.Error(err)) - return true + return true, true } - if blocked { + if isBlocked { s.logger.Debug(eventName+": task has unresolved dependencies; skipping auto-start", zap.String("task_id", taskID), zap.String("blocked_reason", reason)) } - return blocked + return isBlocked, false } // handleTaskDependenciesForTerminalState reacts to a task reaching a terminal @@ -116,7 +123,7 @@ func (s *Service) evaluateDependentAfterPredecessorChange( zap.String("task_id", task.ID)) return } - s.autoStartTaskForStep(ctx, task.ID, task.WorkflowStepID, events.TaskDependenciesResolved, 0) + s.autoStartTaskForStep(ctx, task.ID, task.WorkflowStepID, events.TaskDependenciesResolved, 0, false) } // publishDependenciesResolved announces that a task's last unresolved @@ -185,6 +192,6 @@ func (s *Service) reconcileDependencyLaunchesOnStartup(ctx context.Context) { } s.logger.Info("startup: launching task whose dependencies resolved while down", zap.String("task_id", candidate.TaskID)) - s.autoStartTaskForStep(ctx, candidate.TaskID, candidate.WorkflowStepID, "startup.dependencies_resolved", 0) + s.autoStartTaskForStep(ctx, candidate.TaskID, candidate.WorkflowStepID, "startup.dependencies_resolved", 0, false) } } diff --git a/apps/backend/internal/orchestrator/event_handlers_github.go b/apps/backend/internal/orchestrator/event_handlers_github.go index 589e883c3b..2e0bbfdc0b 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github.go +++ b/apps/backend/internal/orchestrator/event_handlers_github.go @@ -644,6 +644,7 @@ func (s *Service) autoStartReviewTask( if s.shouldSkipTerminalPRAutoStart(ctx, task) { return } + createAutoStartOwned := s.ownsAutoStartOnCreateInFlight(task.ID) // Compete for the one-shot auto-start token set at task creation. // The promotion that ran inside CreateTask may have already triggered // autoStartTaskForStep (Path A) asynchronously; only the first claimer @@ -673,8 +674,16 @@ func (s *Service) autoStartReviewTask( zap.String("task_id", task.ID), zap.Error(err)) s.restoreAutoStartClaim(ctx, task.ID, "review.auto_start") + if createAutoStartOwned || s.ownsAutoStartOnCreateInFlight(task.ID) { + s.restoreAutoStartOnCreate(ctx, task.ID, "review.auto_start") + } return } + // The watcher path can win the permanent review guard while the + // create-time path is still preparing its detached launch. A successful + // watcher launch satisfies that intent too, so clear the durable marker + // owned by the other path. + s.completeAutoStartOnCreate(ctx, task.ID, "review.auto_start") s.logger.Info("auto-started review task", zap.String("task_id", task.ID), zap.Int("pr_number", evt.PR.Number)) diff --git a/apps/backend/internal/orchestrator/event_handlers_github_review_test.go b/apps/backend/internal/orchestrator/event_handlers_github_review_test.go index 28ec813a23..259b70e0f7 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github_review_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_github_review_test.go @@ -405,7 +405,7 @@ func TestAutoStart_BothPathsFireExactlyOnce(t *testing.T) { // Path A: promotion event-driven auto-start on the same task. // autoStartTaskForStep spawns a goroutine — we synchronise via the launched channel. - svc.autoStartTaskForStep(ctx, taskID, stepID, "task.queue_promoted", 0) + svc.autoStartTaskForStep(ctx, taskID, stepID, "task.queue_promoted", 0, false) // Wait for exactly one launch. Allow 2 seconds for the async goroutine. deadline := time.After(2 * time.Second) @@ -568,7 +568,7 @@ func TestAutoStart_NoTokenDoesNotBlock(t *testing.T) { } svc := createTestServiceWithScheduler(repo, sg, taskRepo, agentMgr) - svc.autoStartTaskForStep(ctx, taskID, stepID, "task.moved", 0) + svc.autoStartTaskForStep(ctx, taskID, stepID, "task.moved", 0, false) select { case <-launched: @@ -627,7 +627,7 @@ func TestAutoStart_FailedLaunchWithoutGuardDoesNotStampClaimed(t *testing.T) { } svc := createTestServiceWithScheduler(repo, sg, taskRepo, agentMgr) - svc.autoStartTaskForStep(ctx, taskID, stepID, "task.moved", 0) + svc.autoStartTaskForStep(ctx, taskID, stepID, "task.moved", 0, false) select { case <-attempted: diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index ed2b9cb233..dc7f4ab4c9 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -710,7 +710,7 @@ func (s *Service) handleTaskMoved(ctx context.Context, data watcher.TaskMovedEve return } if prerequisites != nil && prerequisites.targetStep != nil { - s.autoStartTaskForLoadedStep(ctx, task, prerequisites.targetStep, "task.moved", data.QueuePromotion, data.StepTransitionID) + s.autoStartTaskForLoadedStep(ctx, task, prerequisites.targetStep, "task.moved", data.QueuePromotion, data.StepTransitionID, false) } else { s.handleTaskMovedNoSession(ctx, data) } @@ -792,33 +792,66 @@ func (s *Service) loadTaskMovedSessionPrerequisites( // If the target step has auto_start_agent, it creates a session and starts the agent // using agent/executor profile IDs from the task's metadata. func (s *Service) handleTaskMovedNoSession(ctx context.Context, data watcher.TaskMovedEventData) { - s.autoStartTaskForStep(ctx, data.TaskID, data.ToStepID, "task.moved", data.StepTransitionID) + s.autoStartTaskForStep(ctx, data.TaskID, data.ToStepID, "task.moved", data.StepTransitionID, false) } func (s *Service) handleTaskQueuePromoted(ctx context.Context, data watcher.TaskEventData) { - task, err := s.repo.GetTask(ctx, data.TaskID) + s.handleTaskQueuePromotedWithAutoStartOnCreateClaimed(ctx, data, false) +} + +// loadQueuePromotedTaskAndTargetStep loads and validates the task and its +// current workflow step for a task.queue_promoted delivery, returning +// ok=false for every condition under which the event should be dropped +// (load failure, no longer queued/admitted, or a manual move still mid-flight). +func (s *Service) loadQueuePromotedTaskAndTargetStep(ctx context.Context, taskID string) (*models.Task, *wfmodels.WorkflowStep, bool) { + task, err := s.repo.GetTask(ctx, taskID) if err != nil { - s.logger.Warn("task.queue_promoted: failed to load task", zap.String("task_id", data.TaskID), zap.Error(err)) - return + s.logger.Warn("task.queue_promoted: failed to load task", zap.String("task_id", taskID), zap.Error(err)) + return nil, nil, false } if task.QueuedForStepID != "" || !task.WIPAdmitted { - return + return nil, nil, false } if queuedMoveExitPending(task) || manualMoveLifecyclePending(task) { // A manual move has not finished its lifecycle yet. // Keep the promotion token durable; source-exit completion will trigger // queue reconciliation and retry destination entry. - return + return nil, nil, false } targetStep, err := s.workflowStepGetter.GetStep(ctx, task.WorkflowStepID) if err != nil || targetStep == nil { s.logger.Warn("task.queue_promoted: failed to load target step", zap.String("task_id", task.ID), zap.String("step_id", task.WorkflowStepID), zap.Error(err)) + return nil, nil, false + } + return task, targetStep, true +} + +// handleTaskQueuePromotedWithAutoStartOnCreateClaimed is handleTaskQueuePromoted's +// implementation, parameterized by whether the caller already claimed +// MetaKeyAutoStartOnCreate for this launch attempt (see +// claimAutoStartOnCreateForLaunch). autoStartTaskForStep's queue-promotion +// redirect calls this directly so a create-time opt-in that handleTaskCreated +// already claimed carries through promotion instead of being silently +// dropped; every other caller goes through handleTaskQueuePromoted and passes +// false. +// +//nolint:cyclop // queue promotion has independent task, dependency, session, and token recovery states +func (s *Service) handleTaskQueuePromotedWithAutoStartOnCreateClaimed(ctx context.Context, data watcher.TaskEventData, autoStartOnCreateClaimed bool) { + restoreCreateIntent := func() { + if autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(ctx, data.TaskID, "task.queue_promoted") + } + } + task, targetStep, ok := s.loadQueuePromotedTaskAndTargetStep(ctx, data.TaskID) + if !ok { + restoreCreateIntent() return } if err := s.syncTaskStateForQueuePromotion(ctx, task, targetStep); err != nil { s.logger.Warn("task.queue_promoted: failed to synchronize task state", zap.String("task_id", task.ID), zap.Error(err)) + restoreCreateIntent() return } session, sessionErr := s.repo.GetActiveTaskSessionByTaskID(ctx, task.ID) @@ -831,13 +864,19 @@ func (s *Service) handleTaskQueuePromoted(ctx context.Context, data watcher.Task if sessionErr != nil { s.logger.Warn("task.queue_promoted: failed to load active session", zap.String("task_id", task.ID), zap.Error(sessionErr)) + restoreCreateIntent() return } - if session == nil && s.dependencyBlocksAutoStart(ctx, task.ID, "task.queue_promoted") { - // Promotion controls WIP admission, not dependency readiness. Keep both - // the promotion lifecycle token and deferred launch intent intact so the - // dependency-resolution path can start the task once its blockers clear. - return + if session == nil { + if blocked, _ := s.dependencyBlocksAutoStart(ctx, task.ID, "task.queue_promoted"); blocked { + // Promotion controls WIP admission, not dependency readiness. Keep both + // the promotion lifecycle token and deferred launch intent intact so the + // dependency-resolution path can start the task once its blockers clear. + if autoStartOnCreateClaimed { + s.discardAutoStartOnCreate(ctx, task.ID, "task.queue_promoted") + } + return + } } // Read the source descriptor before claiming the one-shot promotion token. // The claim removes the marker durably and a repository implementation may @@ -847,10 +886,12 @@ func (s *Service) handleTaskQueuePromoted(ctx context.Context, data watcher.Task if sourceErr != nil { s.logger.Warn("task.queue_promoted: failed to load source step", zap.String("task_id", task.ID), zap.Error(sourceErr)) + restoreCreateIntent() return } queuePromotionToken := queuePromotionLifecycleToken(task) if !s.claimTaskEventMetadata(ctx, task, models.MetaKeyQueuePromotionPending) { + restoreCreateIntent() return } if remover, ok := s.repo.(taskMetadataKeyRemover); ok { @@ -861,15 +902,21 @@ func (s *Service) handleTaskQueuePromoted(ctx context.Context, data watcher.Task go func() { if err := s.finalizeStepEnter(context.WithoutCancel(ctx), task.ID, session.ID, targetStep, task.Description, targetStep.HasOnEnterAction(wfmodels.OnEnterAutoStartAgent), sourceStep, data.StepTransitionID); err != nil { s.restoreTaskLifecycleToken(context.WithoutCancel(ctx), task.ID, models.MetaKeyQueuePromotionPending, queuePromotionToken, "task.queue_promoted") + if autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(context.WithoutCancel(ctx), task.ID, "task.queue_promoted") + } return } + if autoStartOnCreateClaimed { + s.completeAutoStartOnCreate(context.WithoutCancel(ctx), task.ID, "task.queue_promoted") + } if s.onTaskQueuePromotionEntryComplete != nil { s.onTaskQueuePromotionEntryComplete() } }() return } - s.autoStartTaskForLoadedStep(ctx, task, targetStep, "task.queue_promoted", true, data.StepTransitionID) + s.autoStartTaskForLoadedStep(ctx, task, targetStep, "task.queue_promoted", true, data.StepTransitionID, autoStartOnCreateClaimed) } // handleTaskCreated lets a task that is created directly onto a step whose @@ -907,11 +954,11 @@ func (s *Service) handleTaskQueuePromoted(ctx context.Context, data watcher.Task // anything with it. Either way the key can never change this handler's // outcome, so checking it here would be dead code. // -// The opt-in itself is claimed via claimTaskEventMetadata before launching, -// the same one-shot-token pattern used for MetaKeyQueuePromotionPending: a -// duplicate task.created delivery for the same task would otherwise see the -// key still present and launch a second time. Only the delivery that wins -// the atomic RemoveTaskMetadataKey proceeds. +// The opt-in itself is claimed before launching, the same one-shot-token +// pattern used for MetaKeyQueuePromotionPending. The claim also writes a +// durable in-flight marker before removing the intent, so a process exit +// cannot erase the last recovery signal. Only the delivery that wins the +// local and database ownership hand-off proceeds. func (s *Service) handleTaskCreated(ctx context.Context, data watcher.TaskEventData) { task, err := s.repo.GetTask(ctx, data.TaskID) if err != nil || task == nil { @@ -923,10 +970,22 @@ func (s *Service) handleTaskCreated(ctx context.Context, data watcher.TaskEventD if !autoStartOnCreateActionable(task) { return } - if !s.claimTaskEventMetadata(ctx, task, models.MetaKeyAutoStartOnCreate) { + sessions, err := s.repo.ListTaskSessions(ctx, task.ID) + if err != nil { + s.logger.Warn("task.created: failed to check existing sessions", zap.String("task_id", task.ID), zap.Error(err)) return } - s.autoStartTaskForStep(ctx, task.ID, task.WorkflowStepID, events.TaskCreated, data.StepTransitionID) + if len(sessions) > 0 { + s.completeAutoStartOnCreate(ctx, task.ID, events.TaskCreated) + return + } + if result := s.claimAutoStartOnCreateForLaunch(ctx, task, false); result != autoStartOnCreateClaimOwned { + return + } + // Carry ownership into the launch attempt so a StartTask failure before a + // durable session exists can restore the intent (see + // handleAutoStartFailure). + s.autoStartTaskForStep(ctx, task.ID, task.WorkflowStepID, events.TaskCreated, data.StepTransitionID, true) } func (s *Service) claimTaskEventMetadata(ctx context.Context, task *models.Task, key string) bool { @@ -1072,8 +1131,13 @@ func (s *Service) reconcileTaskLifecycleTokens(ctx context.Context) { s.logger.Warn("failed to list auto-start-on-create tokens for recovery", zap.Error(err)) return } - // autoStarts is the raw, pre-filter list; only entries that pass - // autoStartOnCreateActionable below are added to jobs, so it is left out + autoStartsInFlight, err := lister.ListTasksWithMetadataKey(ctx, models.MetaKeyAutoStartOnCreateInFlight) + if err != nil { + s.logger.Warn("failed to list in-flight auto-start-on-create tokens for recovery", zap.Error(err)) + return + } + // The auto-start lists are raw, pre-filter lists; only entries that pass + // autoStartOnCreateActionable below are added to jobs, so they are left out // of this capacity hint rather than causing routine over-allocation. jobs := make(map[string]struct{}, len(pending)+len(promotions)+len(manualPending)+len(manualCompleted)) for _, task := range pending { @@ -1096,7 +1160,7 @@ func (s *Service) reconcileTaskLifecycleTokens(ctx context.Context) { jobs[task.ID] = struct{}{} } } - for _, task := range autoStarts { + for _, task := range append(autoStarts, autoStartsInFlight...) { // ListTasksWithMetadataKey matches key EXISTENCE, which is broader // than what handleTaskCreated will act on. Rows it refuses keep their // key forever, so admitting them would schedule work that can never @@ -1275,9 +1339,30 @@ func (s *Service) recoverTaskLifecycleAttempt(ctx context.Context, taskID string if _, err := s.clearManualMoveLifecycleMarkersIfCompleted(ctx, taskID, completedAt); err != nil { return true } + task, err = s.repo.GetTask(ctx, taskID) + if err != nil || task == nil { + // A transient failure here only ends this attempt; it leaves any + // surviving lifecycle marker untouched, so the next startup sweep + // re-lists and retries the task. + return false + } } if _, pending := task.Metadata[models.MetaKeyQueuePromotionPending]; pending { s.handleTaskQueuePromoted(ctx, watcher.TaskEventData{TaskID: taskID}) + // handleTaskQueuePromoted's no-session branch schedules a launch via + // autoStartTaskForLoadedStep, which synchronously claims + // MetaKeyAutoStartOnCreate when the task still carries it (see + // claimAutoStartOnCreateForLaunch) before this call returns. Re-fetch + // so the actionability check below observes that claim instead of the + // stale in-memory task, which would otherwise report the token as + // still pending and schedule a second, redundant launch attempt. + task, err = s.repo.GetTask(ctx, taskID) + if err != nil || task == nil { + // A transient failure here only ends this attempt; it leaves any + // surviving lifecycle marker untouched, so the next startup sweep + // re-lists and retries the task. + return false + } } if autoStartOnCreateActionable(task) { s.recoverAutoStartOnCreate(ctx, task) @@ -1411,20 +1496,22 @@ func autoStartOnCreateActionable(task *models.Task) bool { if task == nil || task.IsFromOffice { return false } - return models.HasAutoStartOnCreateIntent(task.Metadata) + return models.HasAutoStartOnCreateIntent(task.Metadata) || + models.HasAutoStartOnCreateInFlight(task.Metadata) } // recoverAutoStartOnCreate replays a lost task.created delivery for a task that // still carries an actionable create-time opt-in. // -// The opt-in is NOT proof that no launch happened. handleTaskCreated is the -// only function that claims this key; a manual StartTask, a task.moved -// auto-start and handleTaskQueuePromoted all launch without touching it. So -// after a lost delivery — the very failure this sweep repairs — an operator -// starting the task by hand leaves the token behind, and replaying it here -// would launch a second agent onto a task that is already running or already -// finished. Neither autoStartTaskForStep nor startTask has an existing-session -// guard, so that launch would go all the way through. +// The opt-in is NOT proof that no launch happened. A manual StartTask launches +// without ever touching this key, so an operator starting the task by hand +// after a lost delivery leaves the token behind. Every automated auto-start +// path (task.moved, dependency resolution, handleTaskQueuePromoted) now claims +// this key itself via claimAutoStartOnCreateForLaunch before it launches. The +// durable in-flight marker remains until a session or run is created, while +// the local ownership map prevents this process from reclaiming it during the +// detached launch. Replaying here without the session check would launch a +// second agent onto a task that is already running or already finished. func (s *Service) recoverAutoStartOnCreate(ctx context.Context, task *models.Task) { sessions, err := s.repo.ListTaskSessions(ctx, task.ID) if err != nil { @@ -1435,13 +1522,13 @@ func (s *Service) recoverAutoStartOnCreate(ctx context.Context, task *models.Tas return } if len(sessions) > 0 { - // The task was started by one of those other paths, so this one-shot - // token is spent. Claim it rather than merely skipping, so the row - // converges instead of being re-scanned on every future startup. - if s.claimTaskEventMetadata(ctx, task, models.MetaKeyAutoStartOnCreate) { - s.logger.Info("discarded spent auto-start-on-create token: task already has a session", - zap.String("task_id", task.ID), zap.Int("session_count", len(sessions))) - } + // The task was started by one of those other paths, so both the + // intent and any in-flight hand-off are spent. Clear them rather than + // merely skipping, so the row converges instead of being re-scanned on + // every future startup. + s.completeAutoStartOnCreate(ctx, task.ID, "auto-start-on-create recovery") + s.logger.Info("discarded spent auto-start-on-create token: task already has a session", + zap.String("task_id", task.ID), zap.Int("session_count", len(sessions))) return } // Re-enter handleTaskCreated exactly like a live delivery would, so the @@ -1503,36 +1590,73 @@ func (s *Service) syncTaskStateForQueuePromotion(ctx context.Context, task *mode return nil } -func (s *Service) autoStartTaskForStep(ctx context.Context, taskID, stepID, eventName string, stepTransitionID int64) { +// autoStartTaskForStep evaluates a target step's on_enter auto-start action +// for a task with no session yet. autoStartOnCreateClaimed is true only when +// the caller (handleTaskCreated) already reserved the create-time launch +// marker before dispatching here. Every other caller passes false and lets +// autoStartTaskForLoadedStep claim the key itself if the task still carries +// it, so a concurrent launch attempt cannot observe the token as if nobody +// had scheduled a launch for it (see claimAutoStartOnCreateForLaunch). +// +//nolint:cyclop // the single auto-start chokepoint evaluates each lifecycle gate before dispatch +func (s *Service) autoStartTaskForStep(ctx context.Context, taskID, stepID, eventName string, stepTransitionID int64, autoStartOnCreateClaimed bool) { task, err := s.repo.GetTask(ctx, taskID) if err != nil { s.logger.Warn(eventName+": failed to load task for auto-start", zap.String("task_id", taskID), zap.Error(err)) + if autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(ctx, taskID, eventName) + } return } - if task == nil || task.QueuedForStepID != "" { + if task == nil { + if autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(ctx, taskID, eventName) + } + return + } + if task.QueuedForStepID != "" { + if autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(ctx, taskID, eventName) + } return } // Dependency gate. Sits here — the single automated-launch chokepoint — so // one check covers task.moved, task.queue_promoted, watcher auto-start and // dependency resolution itself. Placed BEFORE launchDeferredTask so a - // blocked task neither starts a session nor consumes its launch intent. - if s.dependencyBlocksAutoStart(ctx, taskID, eventName) { + // blocked task never starts a session. A genuine block still consumes the + // caller's launch intent: evaluateDependentAfterPredecessorChange and + // reconcileDependencyLaunchesOnStartup independently launch this task once + // its dependency resolves. A failed dependency read has no such fallback, + // so it restores the token instead of stranding it. + if blocked, gateErrored := s.dependencyBlocksAutoStart(ctx, taskID, eventName); blocked { + if gateErrored && autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(ctx, taskID, eventName) + } else if autoStartOnCreateClaimed { + s.discardAutoStartOnCreate(ctx, taskID, eventName) + } return } if hasQueuePromotionPending(task) { // A dependency may resolve after same-step WIP promotion was admitted. // Resume through the promotion handler so destination-entry lifecycle is // claimed and its token participates in deferred-launch failure recovery. - s.handleTaskQueuePromoted(ctx, watcher.TaskEventData{ + // autoStartOnCreateClaimed carries forward here: if handleTaskCreated + // already claimed MetaKeyAutoStartOnCreate for this attempt, the + // promotion path must inherit that ownership rather than starting a + // fresh, unclaimed one (see claimAutoStartOnCreateForLaunch). + s.handleTaskQueuePromotedWithAutoStartOnCreateClaimed(ctx, watcher.TaskEventData{ TaskID: task.ID, StepTransitionID: stepTransitionID, - }) + }, autoStartOnCreateClaimed) return } if task != nil && s.shouldSkipTerminalPRAutoStart(ctx, task) { + if autoStartOnCreateClaimed { + s.discardAutoStartOnCreate(ctx, taskID, eventName) + } return } - if s.launchDeferredTask(ctx, task, eventName, false) { + if s.launchDeferredTask(ctx, task, eventName, false, autoStartOnCreateClaimed) { return } @@ -1543,30 +1667,230 @@ func (s *Service) autoStartTaskForStep(ctx context.Context, taskID, stepID, even zap.String("task_id", taskID), zap.String("to_step_id", stepID), zap.Error(err)) + if autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(ctx, taskID, eventName) + } return } - s.autoStartTaskForLoadedStep(ctx, task, step, eventName, false, stepTransitionID) + s.autoStartTaskForLoadedStep(ctx, task, step, eventName, false, stepTransitionID, autoStartOnCreateClaimed) } -func (s *Service) autoStartTaskForLoadedStep(ctx context.Context, task *models.Task, step *wfmodels.WorkflowStep, eventName string, restoreQueuePromotion bool, stepTransitionID int64) { - if task == nil || task.QueuedForStepID != "" || step == nil { +// autoStartLaunchTokens carries the one-shot lifecycle tokens a launch +// attempt must restore if it fails before producing a durable session or +// run, so the next startup sweep retries instead of the task silently +// losing its opt-in. +type autoStartLaunchTokens struct { + hasGuard bool + restoreQueuePromotion bool + queuePromotionToken interface{} + restoreAutoStartOnCreate bool +} + +type autoStartOnCreateClaimResult uint8 + +const ( + autoStartOnCreateNoIntent autoStartOnCreateClaimResult = iota + autoStartOnCreateClaimOwned + autoStartOnCreateClaimLost + autoStartOnCreateClaimError +) + +// claimAutoStartOnCreateForLaunch reserves the create-time launch intent +// before a detached launch starts. The original intent is removed only after +// the durable in-flight marker is written. A second caller therefore sees a +// lost claim, while a process restart can recover the in-flight marker. +// +//nolint:cyclop // durable claim/recovery ordering has separate stale, write, and ownership outcomes +func (s *Service) claimAutoStartOnCreateForLaunch(ctx context.Context, task *models.Task, alreadyClaimed bool) autoStartOnCreateClaimResult { + if task == nil { + return autoStartOnCreateClaimError + } + if alreadyClaimed { + s.autoStartOnCreateMu.Lock() + if s.autoStartOnCreateInFlight == nil { + s.autoStartOnCreateInFlight = make(map[string]struct{}) + } + s.autoStartOnCreateInFlight[task.ID] = struct{}{} + s.autoStartOnCreateMu.Unlock() + return autoStartOnCreateClaimOwned + } + + s.autoStartOnCreateMu.Lock() + defer s.autoStartOnCreateMu.Unlock() + if s.autoStartOnCreateInFlight == nil { + s.autoStartOnCreateInFlight = make(map[string]struct{}) + } + if _, inFlight := s.autoStartOnCreateInFlight[task.ID]; inFlight { + return autoStartOnCreateClaimLost + } + + intent := models.HasAutoStartOnCreateIntent(task.Metadata) + staleInFlight := models.HasAutoStartOnCreateInFlight(task.Metadata) + if !intent && !staleInFlight { + return autoStartOnCreateNoIntent + } + setter, setterOK := s.repo.(taskMetadataKeySetter) + remover, removerOK := s.repo.(taskMetadataKeyRemover) + if !setterOK || !removerOK { + s.logger.Warn("auto-start-on-create claim unavailable: repository lacks metadata primitives", + zap.String("task_id", task.ID)) + return autoStartOnCreateClaimError + } + if staleInFlight { + // An in-flight marker without a local owner belongs to a previous + // process, or to a claim that crashed between its two writes. Rebuild + // the original intent before taking a fresh reservation. + if !intent { + if err := setter.SetTaskMetadataKey(ctx, task.ID, models.MetaKeyAutoStartOnCreate, true); err != nil { + s.logger.Warn("failed to restore stale auto-start-on-create intent", + zap.String("task_id", task.ID), zap.Error(err)) + return autoStartOnCreateClaimError + } + } + if _, err := remover.RemoveTaskMetadataKey(ctx, task.ID, models.MetaKeyAutoStartOnCreateInFlight); err != nil { + s.logger.Warn("failed to clear stale auto-start-on-create marker", + zap.String("task_id", task.ID), zap.Error(err)) + return autoStartOnCreateClaimError + } + } + if err := setter.SetTaskMetadataKey(ctx, task.ID, models.MetaKeyAutoStartOnCreateInFlight, true); err != nil { + s.logger.Warn("failed to persist auto-start-on-create in-flight marker", + zap.String("task_id", task.ID), zap.Error(err)) + return autoStartOnCreateClaimError + } + claimed, err := remover.RemoveTaskMetadataKey(ctx, task.ID, models.MetaKeyAutoStartOnCreate) + if err != nil { + s.logger.Warn("failed to claim auto-start-on-create intent", + zap.String("task_id", task.ID), zap.Error(err)) + return autoStartOnCreateClaimError + } + if !claimed { + _, _ = remover.RemoveTaskMetadataKey(ctx, task.ID, models.MetaKeyAutoStartOnCreateInFlight) + return autoStartOnCreateClaimLost + } + s.autoStartOnCreateInFlight[task.ID] = struct{}{} + return autoStartOnCreateClaimOwned +} + +func (s *Service) clearAutoStartOnCreateInFlight(taskID string) { + s.autoStartOnCreateMu.Lock() + delete(s.autoStartOnCreateInFlight, taskID) + s.autoStartOnCreateMu.Unlock() +} + +func (s *Service) ownsAutoStartOnCreateInFlight(taskID string) bool { + s.autoStartOnCreateMu.Lock() + defer s.autoStartOnCreateMu.Unlock() + _, owned := s.autoStartOnCreateInFlight[taskID] + return owned +} + +// restoreAutoStartOnCreate restores a failed reservation and leaves a +// durable intent for the next lifecycle sweep. The original intent is +// written before the in-flight marker is removed, so either write ordering +// still leaves recovery work after a process exit. +func (s *Service) restoreAutoStartOnCreate(ctx context.Context, taskID, eventName string) { + setter, setterOK := s.repo.(taskMetadataKeySetter) + remover, removerOK := s.repo.(taskMetadataKeyRemover) + if !setterOK || !removerOK { + s.logger.Warn(eventName+": repository cannot restore auto-start-on-create reservation", + zap.String("task_id", taskID)) + s.clearAutoStartOnCreateInFlight(taskID) return } + if err := setter.SetTaskMetadataKey(ctx, taskID, models.MetaKeyAutoStartOnCreate, true); err != nil { + s.logger.Warn(eventName+": failed to restore auto-start-on-create intent", + zap.String("task_id", taskID), zap.Error(err)) + s.clearAutoStartOnCreateInFlight(taskID) + return + } + if _, err := remover.RemoveTaskMetadataKey(ctx, taskID, models.MetaKeyAutoStartOnCreateInFlight); err != nil { + s.logger.Warn(eventName+": failed to clear auto-start-on-create in-flight marker", + zap.String("task_id", taskID), zap.Error(err)) + } + s.clearAutoStartOnCreateInFlight(taskID) +} + +// discardAutoStartOnCreate consumes a reservation when the task no longer +// needs a create-time launch (for example, a terminal PR or a step without +// auto-start). It is intentionally separate from restoreAutoStartOnCreate. +func (s *Service) discardAutoStartOnCreate(ctx context.Context, taskID, eventName string) { + if remover, ok := s.repo.(taskMetadataKeyRemover); ok { + if _, err := remover.RemoveTaskMetadataKey(ctx, taskID, models.MetaKeyAutoStartOnCreateInFlight); err != nil { + s.logger.Warn(eventName+": failed to clear auto-start-on-create in-flight marker", + zap.String("task_id", taskID), zap.Error(err)) + } + } + s.clearAutoStartOnCreateInFlight(taskID) +} + +// completeAutoStartOnCreate clears both lifecycle markers after a session or +// run is durable. A session-existence check in startup recovery also calls +// this helper, which makes cleanup idempotent after a process restart. +func (s *Service) completeAutoStartOnCreate(ctx context.Context, taskID, eventName string) { + if remover, ok := s.repo.(taskMetadataKeyRemover); ok { + for _, key := range []string{models.MetaKeyAutoStartOnCreate, models.MetaKeyAutoStartOnCreateInFlight} { + if _, err := remover.RemoveTaskMetadataKey(ctx, taskID, key); err != nil { + s.logger.Warn(eventName+": failed to clear auto-start-on-create marker", + zap.String("task_id", taskID), zap.String("metadata_key", key), zap.Error(err)) + } + } + } + s.clearAutoStartOnCreateInFlight(taskID) +} + +//nolint:cyclop,gocognit,funlen // launch dispatch combines existing workflow gates with durable token ownership +func (s *Service) autoStartTaskForLoadedStep(ctx context.Context, task *models.Task, step *wfmodels.WorkflowStep, eventName string, restoreQueuePromotion bool, stepTransitionID int64, autoStartOnCreateClaimed bool) { + if task == nil || task.QueuedForStepID != "" || step == nil { + if autoStartOnCreateClaimed && task != nil { + s.restoreAutoStartOnCreate(ctx, task.ID, eventName) + } + return + } + if models.HasAutoStartOnCreateIntent(task.Metadata) || models.HasAutoStartOnCreateInFlight(task.Metadata) || autoStartOnCreateClaimed { + sessions, err := s.repo.ListTaskSessions(ctx, task.ID) + if err != nil { + s.logger.Warn(eventName+": failed to check existing sessions before auto-start-on-create claim", + zap.String("task_id", task.ID), zap.Error(err)) + if autoStartOnCreateClaimed { + s.restoreAutoStartOnCreate(ctx, task.ID, eventName) + } + return + } + if len(sessions) > 0 { + s.completeAutoStartOnCreate(ctx, task.ID, eventName) + return + } + } if s.shouldSkipTerminalPRAutoStart(ctx, task) { + if autoStartOnCreateClaimed { + s.discardAutoStartOnCreate(ctx, task.ID, eventName) + } return } - if s.launchDeferredTask(ctx, task, eventName, restoreQueuePromotion) { + if s.launchDeferredTask(ctx, task, eventName, restoreQueuePromotion, autoStartOnCreateClaimed) { return } if step == nil || !step.HasOnEnterAction(wfmodels.OnEnterAutoStartAgent) { s.logger.Debug(eventName+": target step has no auto-start", zap.String("task_id", task.ID), zap.String("to_step_id", step.ID)) + if autoStartOnCreateClaimed { + s.discardAutoStartOnCreate(ctx, task.ID, eventName) + } return } + claimResult := s.claimAutoStartOnCreateForLaunch(ctx, task, autoStartOnCreateClaimed) + if claimResult == autoStartOnCreateClaimLost || claimResult == autoStartOnCreateClaimError { + s.logger.Debug(eventName+": auto-start-on-create claim did not win", + zap.String("task_id", task.ID), zap.Uint8("claim_result", uint8(claimResult))) + return + } + restoreAutoStartOnCreate := claimResult == autoStartOnCreateClaimOwned + if s.isOfficeTask(ctx, task.ID) { - s.autoStartOfficeTaskForLoadedStep(ctx, task, step, eventName, restoreQueuePromotion, stepTransitionID) + s.autoStartOfficeTaskForLoadedStep(ctx, task, step, eventName, restoreQueuePromotion, stepTransitionID, restoreAutoStartOnCreate) return } @@ -1615,7 +1939,16 @@ func (s *Service) autoStartTaskForLoadedStep(ctx context.Context, task *models.T s.logger.Error(eventName+": failed to auto-start task", zap.String("task_id", task.ID), zap.Error(err)) - s.handleAutoStartFailure(asyncCtx, task.ID, eventName, hasGuard, restoreQueuePromotion, queuePromotionToken) + s.handleAutoStartFailure(asyncCtx, task.ID, eventName, autoStartLaunchTokens{ + hasGuard: hasGuard, + restoreQueuePromotion: restoreQueuePromotion, + queuePromotionToken: queuePromotionToken, + restoreAutoStartOnCreate: restoreAutoStartOnCreate, + }) + return + } + if restoreAutoStartOnCreate { + s.completeAutoStartOnCreate(asyncCtx, task.ID, eventName) } }() } @@ -1627,7 +1960,7 @@ func (s *Service) autoStartTaskForLoadedStep(ctx context.Context, task *models.T // a run through the engine's Office adapters instead, the same mechanism the // scheduler's recovery sweep and the "assign task" flow already use to start // Office work (internal/office/service/scheduler_recovery.go). -func (s *Service) autoStartOfficeTaskForLoadedStep(ctx context.Context, task *models.Task, step *wfmodels.WorkflowStep, eventName string, restoreQueuePromotion bool, stepTransitionID int64) { +func (s *Service) autoStartOfficeTaskForLoadedStep(ctx context.Context, task *models.Task, step *wfmodels.WorkflowStep, eventName string, restoreQueuePromotion bool, stepTransitionID int64, restoreAutoStartOnCreate bool) { queuePromotionToken := queuePromotionLifecycleToken(task) // Async for the same reason as the kanban path above: the event bus // delivers synchronously and blocking here would stall the HTTP handler @@ -1649,7 +1982,12 @@ func (s *Service) autoStartOfficeTaskForLoadedStep(ctx context.Context, task *mo s.logger.Error(eventName+": failed to queue office auto-start run", zap.String("task_id", task.ID), zap.Error(err)) - s.handleAutoStartFailure(asyncCtx, task.ID, eventName, hasGuard, restoreQueuePromotion, queuePromotionToken) + s.handleAutoStartFailure(asyncCtx, task.ID, eventName, autoStartLaunchTokens{ + hasGuard: hasGuard, + restoreQueuePromotion: restoreQueuePromotion, + queuePromotionToken: queuePromotionToken, + restoreAutoStartOnCreate: restoreAutoStartOnCreate, + }) return } // Log the outcome only after the attempt actually resolves — the log @@ -1663,12 +2001,18 @@ func (s *Service) autoStartOfficeTaskForLoadedStep(ctx context.Context, task *mo s.logger.Info(eventName+": queued office run (no session, auto-start step)", zap.String("task_id", task.ID), zap.String("to_step_id", step.ID)) + if restoreAutoStartOnCreate { + s.completeAutoStartOnCreate(asyncCtx, task.ID, eventName) + } return } s.logger.Debug(eventName+": office auto-start run not queued", zap.String("task_id", task.ID), zap.String("to_step_id", step.ID), zap.String("outcome", string(outcome))) + if restoreAutoStartOnCreate { + s.completeAutoStartOnCreate(asyncCtx, task.ID, eventName) + } }() } @@ -1727,20 +2071,27 @@ func officeAutoStartIdempotencyKey(task *models.Task, agentProfileID, stepID str // only ever called (and the token only ever taken) when the task carries // MetaKeyAutoStartGuard, so restoring it unconditionally stamps // auto_start_claimed onto tasks that never carried the guard, corrupting the -// invariant documented at MetaKeyAutoStartGuard. It also stamps -// MetaKeyAutoStartFailed so the failure surfaces on the task card instead of -// only in backend logs. -func (s *Service) handleAutoStartFailure(ctx context.Context, taskID, eventName string, hasGuard, restoreQueuePromotion bool, queuePromotionToken interface{}) { - if hasGuard { +// invariant documented at MetaKeyAutoStartGuard. It also restores +// MetaKeyAutoStartOnCreate when this attempt claimed it, so a StartTask/queue +// failure before a durable session or run exists leaves the task retryable by +// the next startup sweep instead of stranded with no marker at all. It also +// stamps MetaKeyAutoStartFailed so the failure surfaces on the task card +// instead of only in backend logs. +func (s *Service) handleAutoStartFailure(ctx context.Context, taskID, eventName string, tokens autoStartLaunchTokens) { + if tokens.hasGuard { s.restoreAutoStartClaim(ctx, taskID, eventName) } - if restoreQueuePromotion { - s.restoreTaskLifecycleToken(ctx, taskID, models.MetaKeyQueuePromotionPending, queuePromotionToken, eventName) + if tokens.restoreQueuePromotion { + s.restoreTaskLifecycleToken(ctx, taskID, models.MetaKeyQueuePromotionPending, tokens.queuePromotionToken, eventName) + } + if tokens.restoreAutoStartOnCreate { + s.restoreAutoStartOnCreate(ctx, taskID, eventName) } s.setTaskAutoStartFailedMarker(ctx, taskID, eventName) } -func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eventName string, restoreQueuePromotion bool) bool { +//nolint:cyclop,gocognit,funlen // deferred launches coordinate two durable tokens and async success/failure cleanup +func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eventName string, restoreQueuePromotion bool, autoStartOnCreateClaimed ...bool) bool { if task.Metadata == nil { return false } @@ -1750,6 +2101,9 @@ func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eve } encoded, err := json.Marshal(raw) if err != nil { + if len(autoStartOnCreateClaimed) > 0 && autoStartOnCreateClaimed[0] { + s.restoreAutoStartOnCreate(ctx, task.ID, eventName) + } return false } var intent struct { @@ -1765,8 +2119,26 @@ func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eve } if err := json.Unmarshal(encoded, &intent); err != nil || intent.AgentProfileID == "" { s.logger.Warn(eventName+": invalid deferred launch intent", zap.String("task_id", task.ID), zap.Error(err)) + if len(autoStartOnCreateClaimed) > 0 && autoStartOnCreateClaimed[0] { + s.restoreAutoStartOnCreate(ctx, task.ID, eventName) + } return true } + createClaimed := false + requestedCreateClaim := len(autoStartOnCreateClaimed) > 0 && autoStartOnCreateClaimed[0] + if requestedCreateClaim || models.HasAutoStartOnCreateIntent(task.Metadata) || models.HasAutoStartOnCreateInFlight(task.Metadata) { + claimResult := s.claimAutoStartOnCreateForLaunch(ctx, task, requestedCreateClaim) + switch claimResult { + case autoStartOnCreateClaimOwned: + createClaimed = true + case autoStartOnCreateNoIntent: + if requestedCreateClaim { + return true + } + default: + return true + } + } launchIntent := IntentStart if intent.Intent == "prepare" { launchIntent = IntentPrepare @@ -1775,6 +2147,9 @@ func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eve launchCtx := context.WithoutCancel(ctx) metadataClaimed, claimOK := s.claimDeferredLaunch(launchCtx, task.ID, eventName) if !claimOK { + if createClaimed { + s.restoreAutoStartOnCreate(launchCtx, task.ID, eventName) + } return } launchResp, launchErr := s.LaunchSession(launchCtx, &LaunchSessionRequest{ @@ -1792,8 +2167,14 @@ func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eve if restoreQueuePromotion { s.restoreTaskLifecycleToken(launchCtx, task.ID, models.MetaKeyQueuePromotionPending, queuePromotionLifecycleToken(task), eventName) } + if createClaimed { + s.restoreAutoStartOnCreate(launchCtx, task.ID, eventName) + } return } + if createClaimed { + s.completeAutoStartOnCreate(launchCtx, task.ID, eventName) + } if intent.RecordRecentUse { s.recordSuccessfulDeferredTaskProfileAsync(launchCtx, intent.UserID, launchResp.AgentProfileID) } diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_inflight_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_inflight_test.go new file mode 100644 index 0000000000..dfca61ac20 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_inflight_test.go @@ -0,0 +1,131 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +type failAutoStartOnCreateClaimRepository struct { + sessionExecutorStore +} + +func (r *failAutoStartOnCreateClaimRepository) RemoveTaskMetadataKey(ctx context.Context, taskID, key string) (bool, error) { + if key == models.MetaKeyAutoStartOnCreate { + return false, errors.New("injected auto-start claim failure") + } + return r.sessionExecutorStore.RemoveTaskMetadataKey(ctx, taskID, key) +} + +func TestAutoStartOnCreateClaimLeavesDurableInFlightMarker(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-inflight", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-inflight", WorkspaceID: "ws-inflight", Name: "WF", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "task-inflight", WorkspaceID: "ws-inflight", WorkflowID: "wf-inflight", + WorkflowStepID: "step-inflight", State: v1.TaskStateCreated, + Metadata: map[string]interface{}{models.MetaKeyAutoStartOnCreate: true}, + CreatedAt: now, UpdatedAt: now, + })) + + svc := &Service{repo: repo, logger: testLogger()} + task, err := repo.GetTask(ctx, "task-inflight") + requireNoError(t, err) + if got := svc.claimAutoStartOnCreateForLaunch(ctx, task, false); got != autoStartOnCreateClaimOwned { + t.Fatalf("claim result = %d, want owned", got) + } + + reloaded, err := repo.GetTask(ctx, task.ID) + requireNoError(t, err) + if models.HasAutoStartOnCreateIntent(reloaded.Metadata) { + t.Fatal("auto-start-on-create intent remained after claim") + } + if !models.HasAutoStartOnCreateInFlight(reloaded.Metadata) { + t.Fatal("in-flight marker was not persisted before launch") + } + if got := svc.claimAutoStartOnCreateForLaunch(ctx, reloaded, false); got != autoStartOnCreateClaimLost { + t.Fatalf("second claim result = %d, want lost", got) + } + + svc.completeAutoStartOnCreate(ctx, task.ID, "test") + reloaded, err = repo.GetTask(ctx, task.ID) + requireNoError(t, err) + if models.HasAutoStartOnCreateIntent(reloaded.Metadata) || models.HasAutoStartOnCreateInFlight(reloaded.Metadata) { + t.Fatalf("auto-start markers remained after completion: %#v", reloaded.Metadata) + } +} + +func TestAutoStartOnCreateClaimReclaimsStaleInFlightMarker(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-stale", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-stale", WorkspaceID: "ws-stale", Name: "WF", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "task-stale", WorkspaceID: "ws-stale", WorkflowID: "wf-stale", + WorkflowStepID: "step-stale", State: v1.TaskStateCreated, + Metadata: map[string]interface{}{models.MetaKeyAutoStartOnCreateInFlight: true}, + CreatedAt: now, UpdatedAt: now, + })) + + svc := &Service{repo: repo, logger: testLogger()} + task, err := repo.GetTask(ctx, "task-stale") + requireNoError(t, err) + if got := svc.claimAutoStartOnCreateForLaunch(ctx, task, false); got != autoStartOnCreateClaimOwned { + t.Fatalf("stale claim result = %d, want owned", got) + } + + reloaded, err := repo.GetTask(ctx, task.ID) + requireNoError(t, err) + if models.HasAutoStartOnCreateIntent(reloaded.Metadata) { + t.Fatal("stale recovery left the rebuilt intent in place") + } + if !models.HasAutoStartOnCreateInFlight(reloaded.Metadata) { + t.Fatal("stale recovery did not create a fresh in-flight marker") + } +} + +func TestAutoStartOnCreateClaimFailureSkipsLaunch(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-claim-error", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-claim-error", WorkspaceID: "ws-claim-error", Name: "WF", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "task-claim-error", WorkspaceID: "ws-claim-error", WorkflowID: "wf-claim-error", + WorkflowStepID: "step-claim-error", State: v1.TaskStateCreated, + Metadata: map[string]interface{}{models.MetaKeyAutoStartOnCreate: true}, + CreatedAt: now, UpdatedAt: now, + })) + + step := &wfmodels.WorkflowStep{ + ID: "step-claim-error", WorkflowID: "wf-claim-error", Name: "Start", Position: 0, + Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}}}, + } + stepGetter := newMockStepGetter() + stepGetter.steps[step.ID] = step + failingRepo := &failAutoStartOnCreateClaimRepository{sessionExecutorStore: repo} + svc := &Service{repo: failingRepo, workflowStepGetter: stepGetter, logger: testLogger()} + + task, err := repo.GetTask(ctx, "task-claim-error") + requireNoError(t, err) + // The claim must fail closed before any goroutine or executor call. The + // executor field is deliberately unset, so a launch would panic this test. + svc.autoStartTaskForLoadedStep(ctx, task, step, "test.claim_error", false, 0, false) + + reloaded, err := repo.GetTask(ctx, task.ID) + requireNoError(t, err) + if !models.HasAutoStartOnCreateIntent(reloaded.Metadata) { + t.Fatal("claim failure removed the original intent") + } + if !models.HasAutoStartOnCreateInFlight(reloaded.Metadata) { + t.Fatal("claim failure did not leave a recoverable in-flight marker") + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_recovery_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_recovery_test.go new file mode 100644 index 0000000000..695c882876 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_auto_start_recovery_test.go @@ -0,0 +1,552 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + taskservice "github.com/kandev/kandev/internal/task/service" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// This file covers the two latent auto-start recovery defects filed as a +// sibling of WO-36.1: a replay double-launch race between the queue-promotion +// lifecycle token and the create-time auto-start opt-in (Gap A), and a +// pre-session StartTask failure that stranded the create-time opt-in with no +// durable marker (Gap B). Both are fixed by the same mechanism: the opt-in +// is claimed synchronously by the launch attempt itself (autoStartTaskForLoadedStep, +// via claimAutoStartOnCreateForLaunch), not merely observed, and restored on +// failure (handleAutoStartFailure) exactly like the existing queue-promotion +// token. + +// TestRecoverTaskLifecycleAttemptConcurrentPromotionAndAutoStartOnCreateLaunchesOnce +// covers Gap A: recoverTaskLifecycleAttempt used to evaluate +// autoStartOnCreateActionable against the SAME in-memory task it loaded at +// the top of the function, without re-fetching after the queue-promotion +// branch ran. handleTaskQueuePromoted's no-session path schedules a launch +// through autoStartTaskForLoadedStep, but that launch's actual session +// creation happens in a detached goroutine — so a stale read could still see +// MetaKeyAutoStartOnCreate present and schedule a second, redundant launch via +// recoverAutoStartOnCreate. +// +// The fix makes the race structural rather than timing-dependent: +// autoStartTaskForLoadedStep claims MetaKeyAutoStartOnCreate synchronously, +// before spawning its launch goroutine, and recoverTaskLifecycleAttempt +// re-fetches the task after the promotion branch. So by the time +// autoStartOnCreateActionable evaluates, the token is already gone. +func TestRecoverTaskLifecycleAttemptConcurrentPromotionAndAutoStartOnCreateLaunchesOnce(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now})) + + metadata := map[string]interface{}{ + models.MetaKeyQueuePromotionPending: true, + models.MetaKeyAutoStartOnCreate: true, + models.MetaKeyAgentProfileID: "routine-assignee", + } + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t-race", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + WIPAdmitted: true, + Title: "Concurrent tokens", + Description: "prompt", + State: v1.TaskStateTODO, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Auto Start Step", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks["t-race"] = &v1.Task{ + ID: "t-race", + WorkspaceID: "ws1", + WorkflowID: "wf1", + Description: "prompt", + State: v1.TaskStateTODO, + Metadata: metadata, + } + launched := make(chan string, 4) + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentProfileID, _ := req.Metadata[models.MetaKeyAgentProfileID].(string) + launched <- agentProfileID + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr) + + svc.recoverTaskLifecycleAttempt(ctx, "t-race") + + select { + case got := <-launched: + if got != "routine-assignee" { + t.Fatalf("AgentProfileID = %q, want routine-assignee", got) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the auto-start launch") + } + // Settle window: prove no second launch follows the first, rather than + // just reading one value off the channel and declaring victory. + select { + case got := <-launched: + t.Fatalf("unexpected second launch for agent profile %q", got) + case <-time.After(300 * time.Millisecond): + } + + sessions, err := repo.ListTaskSessions(ctx, "t-race") + requireNoError(t, err) + if len(sessions) != 1 { + t.Fatalf("ListTaskSessions(t-race) = %d sessions, want 1", len(sessions)) + } +} + +// TestRecoverTaskLifecycleAttemptConcurrentPromotionLaunchesOnceWithoutWorkspaceBindingElection +// covers the same Gap A race, but on the session-create route that does NOT +// go through CreateTaskSessionWithWorkspaceBinding's election +// (executor_execute.go: bindWorkspace && !taskUsesDeferredEnvironmentInheritance). +// Today that election is what accidentally limits the race to one launch on +// the ordinary path: the loser's session-create call fails with "workspace is +// preparing", and handleAutoStartFailure restores the promotion token for the +// next sweep to converge. A task using workspace.mode=inherit_parent takes the +// CreateTaskSessionWithInitialRuntimeSeed / plain CreateTaskSession route +// instead, neither of which enforces per-task session uniqueness — so this is +// the test that proves the actual fix (the synchronous token claim), not the +// incidental mitigation. +func TestRecoverTaskLifecycleAttemptConcurrentPromotionLaunchesOnceWithoutWorkspaceBindingElection(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now})) + + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t-race-parent", WorkspaceID: "ws1", WorkflowID: "wf1", WorkflowStepID: "step1", + Title: "Parent", State: v1.TaskStateInProgress, CreatedAt: now, UpdatedAt: now, + })) + // ExecutorType intentionally left empty: the child launch below resolves + // no executor either (no MetaKeyExecutorID, no default configured in this + // test harness), and LaunchPreparedSession rejects an inherited + // environment whose non-empty ExecutorType disagrees with the launch's + // resolved one (executor_execute.go's workspace-reuse-unsafe guard). + requireNoError(t, repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{ + ID: "env-parent", TaskID: "t-race-parent", + Status: models.TaskEnvironmentStatusReady, + })) + requireNoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "ps1", TaskID: "t-race-parent", State: models.TaskSessionStateRunning, + IsPrimary: true, TaskEnvironmentID: "env-parent", StartedAt: now, UpdatedAt: now, + })) + + metadata := map[string]interface{}{ + models.MetaKeyQueuePromotionPending: true, + models.MetaKeyAutoStartOnCreate: true, + models.MetaKeyAgentProfileID: "routine-assignee", + "workspace": map[string]interface{}{"mode": "inherit_parent"}, + } + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t-race-child", + ParentID: "t-race-parent", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + WIPAdmitted: true, + Title: "Concurrent tokens, inherited workspace", + Description: "prompt", + State: v1.TaskStateTODO, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Auto Start Step", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks["t-race-child"] = &v1.Task{ + ID: "t-race-child", + ParentID: "t-race-parent", + WorkspaceID: "ws1", + WorkflowID: "wf1", + Description: "prompt", + State: v1.TaskStateTODO, + Metadata: metadata, + } + launched := make(chan string, 4) + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentProfileID, _ := req.Metadata[models.MetaKeyAgentProfileID].(string) + launched <- agentProfileID + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr) + + svc.recoverTaskLifecycleAttempt(ctx, "t-race-child") + + select { + case got := <-launched: + if got != "routine-assignee" { + t.Fatalf("AgentProfileID = %q, want routine-assignee", got) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the auto-start launch") + } + select { + case got := <-launched: + t.Fatalf("unexpected second launch for agent profile %q", got) + case <-time.After(300 * time.Millisecond): + } + + sessions, err := repo.ListTaskSessions(ctx, "t-race-child") + requireNoError(t, err) + if len(sessions) != 1 { + t.Fatalf("ListTaskSessions(t-race-child) = %d sessions, want 1", len(sessions)) + } + if sessions[0].TaskEnvironmentID != "env-parent" { + t.Fatalf("session TaskEnvironmentID = %q, want env-parent (inherited, non-bindWorkspace route confirms this isn't the workspace-binding election doing the work)", sessions[0].TaskEnvironmentID) + } +} + +// TestHandleTaskCreatedRestoresAutoStartOnCreateOnPreSessionFailure covers Gap +// B: handleTaskCreated claims (removes) MetaKeyAutoStartOnCreate synchronously +// before handing the launch to autoStartTaskForLoadedStep's detached +// goroutine. Before this fix, handleAutoStartFailure restored +// MetaKeyAutoStartGuard and MetaKeyQueuePromotionPending on failure but never +// MetaKeyAutoStartOnCreate, so a StartTask failure before a session exists +// stranded the task with no durable marker for the next startup sweep to +// find. +// +// getTaskErr fails scheduler.GetTask, which startTask calls after resolving +// the target step but before prepareSessionForStart creates any session row +// — a deterministic pre-session failure, no sleep-and-hope. +func TestHandleTaskCreatedRestoresAutoStartOnCreateOnPreSessionFailure(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now})) + + metadata := map[string]interface{}{ + models.MetaKeyAutoStartOnCreate: true, + models.MetaKeyAgentProfileID: "routine-assignee", + } + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t-pre-session-failure", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Title: "Routine run", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Routine Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + launched := make(chan string, 2) + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentProfileID, _ := req.Metadata[models.MetaKeyAgentProfileID].(string) + launched <- agentProfileID + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + taskRepo := newMockTaskRepo() + taskRepo.getTaskErr = errors.New("boom: transient task lookup failure") + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr) + + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t-pre-session-failure"}) + + waitForAutoStartFailedMarker(t, repo, ctx, "t-pre-session-failure") + + select { + case got := <-launched: + t.Fatalf("unexpected launch despite pre-session task lookup failure: %q", got) + default: + } + + reloaded, err := repo.GetTask(ctx, "t-pre-session-failure") + requireNoError(t, err) + if !models.HasAutoStartOnCreateIntent(reloaded.Metadata) { + t.Fatal("MetaKeyAutoStartOnCreate was not restored after a pre-session StartTask failure; the task is stranded with no durable marker for the next startup sweep") + } + sessions, err := repo.ListTaskSessions(ctx, "t-pre-session-failure") + requireNoError(t, err) + if len(sessions) != 0 { + t.Fatalf("ListTaskSessions = %d, want 0 (the injected failure must happen before session persistence)", len(sessions)) + } + + // Clear the injected failure and let a subsequent startup sweep retry + // using the restored token. Locked because the first launch attempt's + // detached goroutine (see autoStartTaskForLoadedStep) can still be + // finishing unrelated work on taskRepo after waitForAutoStartFailedMarker + // observes the marker on the separate repo above; GetTask takes the same + // lock on its read side. + taskRepo.mu.Lock() + taskRepo.getTaskErr = nil + taskRepo.tasks["t-pre-session-failure"] = &v1.Task{ + ID: "t-pre-session-failure", + WorkspaceID: "ws1", + WorkflowID: "wf1", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + } + taskRepo.mu.Unlock() + + svc.reconcileTaskLifecycleTokens(ctx) + + select { + case got := <-launched: + if got != "routine-assignee" { + t.Fatalf("AgentProfileID = %q, want routine-assignee", got) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the retried auto-start launch") + } +} + +// TestHandleTaskCreatedWithQueuePromotionPendingRestoresAutoStartOnCreateOnPreSessionFailure +// covers Review Round 1's F1: a task carrying BOTH MetaKeyAutoStartOnCreate and +// MetaKeyQueuePromotionPending reaches handleTaskCreated (as recoverAutoStartOnCreate's +// replay does), which claims MetaKeyAutoStartOnCreate and calls autoStartTaskForStep +// with autoStartOnCreateClaimed=true. autoStartTaskForStep sees the promotion +// token and redirects into the queue-promotion handler; before the fix that +// redirect dropped autoStartOnCreateClaimed, so claimAutoStartOnCreateForLaunch +// saw the (already-removed) key as unclaimed and a pre-session StartTask +// failure restored MetaKeyQueuePromotionPending but never +// MetaKeyAutoStartOnCreate, stranding the task on exactly the recovery path +// this card exists to close. +func TestHandleTaskCreatedWithQueuePromotionPendingRestoresAutoStartOnCreateOnPreSessionFailure(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now})) + + metadata := map[string]interface{}{ + models.MetaKeyAutoStartOnCreate: true, + models.MetaKeyQueuePromotionPending: true, + models.MetaKeyAgentProfileID: "routine-assignee", + } + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t-promotion-and-create", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + WIPAdmitted: true, + Title: "Promoted routine run", + Description: "prompt", + State: v1.TaskStateTODO, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Routine Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + launched := make(chan string, 2) + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentProfileID, _ := req.Metadata[models.MetaKeyAgentProfileID].(string) + launched <- agentProfileID + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + taskRepo := newMockTaskRepo() + taskRepo.getTaskErr = errors.New("boom: transient task lookup failure") + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr) + + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t-promotion-and-create"}) + + waitForAutoStartFailedMarker(t, repo, ctx, "t-promotion-and-create") + + select { + case got := <-launched: + t.Fatalf("unexpected launch despite pre-session task lookup failure: %q", got) + default: + } + + reloaded, err := repo.GetTask(ctx, "t-promotion-and-create") + requireNoError(t, err) + if !models.HasAutoStartOnCreateIntent(reloaded.Metadata) { + t.Fatal("MetaKeyAutoStartOnCreate was not restored after a pre-session StartTask failure reached through the queue-promotion redirect") + } + if _, pending := reloaded.Metadata[models.MetaKeyQueuePromotionPending]; !pending { + t.Fatal("MetaKeyQueuePromotionPending was not restored after the pre-session StartTask failure") + } + sessions, err := repo.ListTaskSessions(ctx, "t-promotion-and-create") + requireNoError(t, err) + if len(sessions) != 0 { + t.Fatalf("ListTaskSessions = %d, want 0 (the injected failure must happen before session persistence)", len(sessions)) + } +} + +// erroringDependencyReader models a transient DependencyGate read failure, +// distinct from a genuine dependency block: DependencyGate itself errors, so +// dependencyBlocksAutoStart fails closed with gateErrored=true regardless of +// whether the task has any dependency edges at all. +type erroringDependencyReader struct { + err error +} + +func (r *erroringDependencyReader) DependencyGate(context.Context, string) (bool, string, error) { + return false, "", r.err +} + +func (r *erroringDependencyReader) ListDependentTaskIDs(context.Context, string) ([]string, error) { + return nil, nil +} + +func (r *erroringDependencyReader) ListPendingDependencyLaunches(context.Context) ([]taskservice.PendingDependencyLaunch, error) { + return nil, nil +} + +// TestAutoStartTaskForStepRestoresAutoStartOnCreateOnDependencyGateError covers +// Review Round 2's F4: autoStartTaskForStep's dependency gate is its third +// early return, and unlike the other two (GetTask, GetStep failures, both +// already fixed), it did not restore MetaKeyAutoStartOnCreate when +// autoStartOnCreateClaimed was true. handleTaskCreated already removes the key +// before dispatching here, and the startup sweep discovers candidates only by +// that key's existence — so a DependencyGate read failure (not a genuine +// block) stranded the task permanently: it has no dependency edge for +// reconcileDependencyLaunchesOnStartup to find, and no other path re-lists it. +// +// A genuine block is deliberately NOT covered by this restore (see +// dependencyBlocksAutoStart's gateErrored split): it is already covered by +// evaluateDependentAfterPredecessorChange / reconcileDependencyLaunchesOnStartup +// once the dependency resolves, so restoring there would only burn +// recoverTaskLifecycleAttempt's bounded retry budget on every startup for as +// long as the task stays blocked. +func TestAutoStartTaskForStepRestoresAutoStartOnCreateOnDependencyGateError(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + requireNoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now})) + requireNoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "WF", CreatedAt: now, UpdatedAt: now})) + + metadata := map[string]interface{}{ + models.MetaKeyAutoStartOnCreate: true, + models.MetaKeyAgentProfileID: "routine-assignee", + } + requireNoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "t-dependency-gate-error", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Title: "Routine run", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + })) + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Routine Start", Position: 0, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{ + {Type: wfmodels.OnEnterAutoStartAgent}, + }, + }, + } + + launched := make(chan string, 2) + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + agentProfileID, _ := req.Metadata[models.MetaKeyAgentProfileID].(string) + launched <- agentProfileID + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + taskRepo := newMockTaskRepo() + taskRepo.tasks["t-dependency-gate-error"] = &v1.Task{ + ID: "t-dependency-gate-error", + WorkspaceID: "ws1", + WorkflowID: "wf1", + Description: "prompt", + State: v1.TaskStateCreated, + Metadata: metadata, + } + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr) + svc.SetTaskDependencyReader(&erroringDependencyReader{err: errors.New("boom: transient dependency gate failure")}) + + svc.handleTaskCreated(ctx, watcher.TaskEventData{TaskID: "t-dependency-gate-error"}) + + select { + case got := <-launched: + t.Fatalf("unexpected launch despite dependency gate error: %q", got) + default: + } + + reloaded, err := repo.GetTask(ctx, "t-dependency-gate-error") + requireNoError(t, err) + if !models.HasAutoStartOnCreateIntent(reloaded.Metadata) { + t.Fatal("MetaKeyAutoStartOnCreate was not restored after a DependencyGate error; the task is stranded with no dependency edge for reconcileDependencyLaunchesOnStartup to find and no other marker for the next sweep") + } + sessions, err := repo.ListTaskSessions(ctx, "t-dependency-gate-error") + requireNoError(t, err) + if len(sessions) != 0 { + t.Fatalf("ListTaskSessions = %d, want 0 (the dependency gate must block before any session is created)", len(sessions)) + } + + // Clear the injected failure and let a subsequent startup sweep retry using + // the restored token. + svc.SetTaskDependencyReader(&resolvedDependencyReader{}) + + svc.reconcileTaskLifecycleTokens(ctx) + + select { + case got := <-launched: + if got != "routine-assignee" { + t.Fatalf("AgentProfileID = %q, want routine-assignee", got) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the retried auto-start launch") + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index e822b9f298..818b066e6a 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -628,6 +628,13 @@ type Service struct { passthroughDispatchMu sync.Mutex passthroughDispatches map[string]map[*passthroughDispatchToken]struct{} + // autoStartOnCreateMu serializes the local ownership hand-off for the + // durable auto-start-on-create marker. The database marker survives a + // process restart; this map prevents recovery and event delivery from + // reclaiming the same marker while its detached launch is still running. + autoStartOnCreateMu sync.Mutex + autoStartOnCreateInFlight map[string]struct{} + // Message creator for saving agent responses messageCreator MessageCreator @@ -1546,6 +1553,7 @@ func NewService( executor: exec, scheduler: sched, messageQueue: msgQueue, + autoStartOnCreateInFlight: make(map[string]struct{}), taskLaunchRecoveryRepo: taskLaunchRecoveryRepo, clarificationWatchdogTimeout: 15 * time.Second, gitSnapshotCache: newGitSnapshotCache(), diff --git a/apps/backend/internal/orchestrator/session_launch.go b/apps/backend/internal/orchestrator/session_launch.go index e1befdf90a..ddee442939 100644 --- a/apps/backend/internal/orchestrator/session_launch.go +++ b/apps/backend/internal/orchestrator/session_launch.go @@ -276,14 +276,26 @@ func (s *Service) isPassthroughProfile(ctx context.Context, profileID string) bo return info.CLIPassthrough } +// blocksAutoStartLaunch reports whether an auto-start request must be +// downgraded to a prepare, either because the task's current step does not +// allow it or because it has an unresolved dependency. The dependency gate's +// launch-token restore concern does not apply here: this path owns no +// lifecycle token to restore. +func (s *Service) blocksAutoStartLaunch(ctx context.Context, req *LaunchSessionRequest) bool { + if s.shouldBlockAutoStart(ctx, req) { + return true + } + blocked, _ := s.dependencyBlocksAutoStart(ctx, req.TaskID, "session.launch") + return blocked +} + // launchStart creates a new session and launches the agent. // If the request is an auto-start and the task's current workflow step does not // have auto_start_agent, or the task has unresolved dependencies, the request // is downgraded to a prepare (workspace-only, no agent) to prevent unwanted // auto-starts from the frontend's useAutoStartSession hook. func (s *Service) launchStart(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { - if req.AutoStart && (s.shouldBlockAutoStart(ctx, req) || - s.dependencyBlocksAutoStart(ctx, req.TaskID, "session.launch")) { + if req.AutoStart && s.blocksAutoStartLaunch(ctx, req) { req.LaunchWorkspace = true return s.launchPrepare(ctx, req) } diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 91c38e2853..d67ba1b338 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -221,6 +221,12 @@ const ( // whose Routine workflow start step has no other transition to carry it // into an auto_start_agent evaluation. MetaKeyAutoStartOnCreate = "auto_start_on_create" + // MetaKeyAutoStartOnCreateInFlight is a durable hand-off marker for an + // auto-start-on-create launch. The original intent is consumed before the + // detached launch starts, but this marker remains until a session or run + // is durable. That lets startup recovery retry a process that exits in the + // gap instead of losing the last recovery signal. + MetaKeyAutoStartOnCreateInFlight = "auto_start_on_create_in_flight" // MetaKeyStepHandoffCarry is a single-slot, task-scoped token carrying one // consuming transition's completion handoff exactly one hop, to the next // step's first dispatched prompt. Its value is a StepHandoffCarryToken. @@ -346,6 +352,15 @@ func HasAutoStartOnCreateIntent(metadata map[string]interface{}) bool { return ok && intent } +// HasAutoStartOnCreateInFlight reports whether a launch attempt still owns +// the durable hand-off marker for a create-time auto-start. Only an explicit +// true value counts; JSON rehydration and in-process callers use the same +// representation as the other lifecycle markers. +func HasAutoStartOnCreateInFlight(metadata map[string]interface{}) bool { + inFlight, ok := metadata[MetaKeyAutoStartOnCreateInFlight].(bool) + return ok && inFlight +} + // TaskSession.Metadata key that records how the session came into existence. // workflow_switch means the session profile was selected by workflow routing // rather than direct user selection.