diff --git a/apps/backend/internal/events/bus/memory.go b/apps/backend/internal/events/bus/memory.go index 45f93cba89..5c3d293865 100644 --- a/apps/backend/internal/events/bus/memory.go +++ b/apps/backend/internal/events/bus/memory.go @@ -33,6 +33,13 @@ type memorySubscription struct { mu sync.Mutex } +type memoryDelivery struct { + pattern string + subscription *memorySubscription + queue *queueGroup + queueKey string +} + // queueGroup manages load balancing for queue subscriptions type queueGroup struct { subscribers []*memorySubscription @@ -100,9 +107,8 @@ func NewMemoryEventBus(log *logger.Logger) *MemoryEventBus { // not enough for the per-session/per-run suffixed subjects (see Event.Subject). func (b *MemoryEventBus) Publish(ctx context.Context, subject string, event *Event) error { b.mu.RLock() - defer b.mu.RUnlock() - if b.closed { + b.mu.RUnlock() return fmt.Errorf("event bus is closed") } @@ -110,10 +116,12 @@ func (b *MemoryEventBus) Publish(ctx context.Context, subject string, event *Eve event.Subject = subject } - // Track which queue groups we've already delivered to + // Snapshot matching subscriptions before invoking handlers. A handler may + // publish another event or register a subscription. Holding the bus read + // lock while invoking it would block that nested operation and can deadlock + // when a subscriber-registration writer is waiting for the same lock. + deliveries := make([]memoryDelivery, 0) deliveredQueues := make(map[string]bool) - - // Find all matching subscriptions for pattern, subs := range b.subscriptions { for _, sub := range subs { sub.mu.Lock() @@ -133,17 +141,40 @@ func (b *MemoryEventBus) Publish(ctx context.Context, subject string, event *Eve queueKey := sub.queue + ":" + pattern if !deliveredQueues[queueKey] { deliveredQueues[queueKey] = true - b.publishToQueue(ctx, queueKey, subject, event) + deliveries = append(deliveries, memoryDelivery{ + pattern: pattern, + subscription: sub, + queue: b.queues[queueKey], + queueKey: queueKey, + }) } continue } - // Regular subscription - deliver to all synchronously to preserve ordering - if err := invokeHandler(ctx, b.logger, "regular", subject, pattern, event, sub.handler); err != nil { - b.logger.Error("Event handler error", - zap.String("subject", subject), - zap.Error(err)) - } + deliveries = append(deliveries, memoryDelivery{ + pattern: pattern, + subscription: sub, + }) + } + } + b.mu.RUnlock() + + for _, delivery := range deliveries { + if delivery.subscription.queue != "" { + // Queue subscriptions select their target under the queue lock, then + // invoke the selected handler without holding the bus lock. + b.publishToQueue(ctx, delivery.queue, delivery.queueKey, subject, event) + continue + } + if !delivery.subscription.IsValid() { + continue + } + + // Regular subscriptions are delivered synchronously to preserve ordering. + if err := invokeHandler(ctx, b.logger, "regular", subject, delivery.pattern, event, delivery.subscription.handler); err != nil { + b.logger.Error("Event handler error", + zap.String("subject", subject), + zap.Error(err)) } } @@ -348,9 +379,8 @@ func compilePattern(pattern string) *regexp.Regexp { } // publishToQueue delivers to one subscriber in the queue group (round-robin) -func (b *MemoryEventBus) publishToQueue(ctx context.Context, queueKey, subject string, event *Event) { - qg, ok := b.queues[queueKey] - if !ok { +func (b *MemoryEventBus) publishToQueue(ctx context.Context, qg *queueGroup, queueKey, subject string, event *Event) { + if qg == nil { return } diff --git a/apps/backend/internal/events/bus/memory_test.go b/apps/backend/internal/events/bus/memory_test.go index 14793293f4..933eae4a00 100644 --- a/apps/backend/internal/events/bus/memory_test.go +++ b/apps/backend/internal/events/bus/memory_test.go @@ -453,6 +453,32 @@ func TestMemoryEventBus_Request(t *testing.T) { } } +func TestMemoryEventBusHandlerCanSubscribeDuringPublish(t *testing.T) { + bus := NewMemoryEventBus(newTestLogger(t)) + + if _, err := bus.Subscribe("outer", func(context.Context, *Event) error { + _, err := bus.Subscribe("inner", func(context.Context, *Event) error { return nil }) + return err + }); err != nil { + t.Fatalf("Subscribe outer: %v", err) + } + + done := make(chan error, 1) + go func() { + done <- bus.Publish(context.Background(), "outer", NewEvent("outer", "test", nil)) + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Publish: %v", err) + } + bus.Close() + case <-time.After(time.Second): + t.Fatal("event handler could not subscribe while Publish was dispatching") + } +} + func TestMemoryEventBus_RequestTimeout(t *testing.T) { synctest.Test(t, func(t *testing.T) { log := newTestLogger(t) diff --git a/apps/backend/internal/orchestrator/event_handlers_feeder_lifecycle_test.go b/apps/backend/internal/orchestrator/event_handlers_feeder_lifecycle_test.go index baf6068f41..83e012fdb6 100644 --- a/apps/backend/internal/orchestrator/event_handlers_feeder_lifecycle_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_feeder_lifecycle_test.go @@ -20,7 +20,7 @@ type manualMoveFeederPullRecorder struct { once sync.Once } -func (r *manualMoveFeederPullRecorder) ReconcileFeederPulls(ctx context.Context, _, feederStepID string) { +func (r *manualMoveFeederPullRecorder) ReconcileFeederPulls(ctx context.Context, _, feederStepID string) error { task, err := r.repo.GetTask(ctx, "manual-feeder-barrier") if err != nil { r.err <- err @@ -28,6 +28,7 @@ func (r *manualMoveFeederPullRecorder) ReconcileFeederPulls(ctx context.Context, r.err <- &unexpectedStepError{got: task.WorkflowStepID, want: feederStepID} } r.once.Do(func() { close(r.called) }) + return nil } type unexpectedStepError struct { diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 82dc6fdc7d..59ce3b3920 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -69,6 +69,10 @@ type taskMetadataKeyRemover interface { RemoveTaskMetadataKey(context.Context, string, string) (bool, error) } +type manualMoveLifecycleMarkerCleaner interface { + ClearManualMoveLifecycleMarkersIfCompleted(context.Context, string, time.Time) (bool, error) +} + type taskMetadataKeySetter interface { SetTaskMetadataKey(context.Context, string, string, interface{}) error } @@ -916,17 +920,71 @@ func queuePromotionLifecycleToken(task *models.Task) interface{} { return true } -// lifecycleSweepStuckWarningInterval controls how long -// reconcileTaskLifecycleTokens waits before logging a "still recovering" -// warning while the sweep is still in flight. It is a var so tests can -// shorten it. This is a log-only signal: it never cancels the sweep, because -// the resume calls it drives run under context.WithoutCancel +// lifecycleSweepStuckWarningInterval controls how often +// reconcileTaskLifecycleTokens logs a "still recovering" warning while the +// sweep is still in flight. It is a var so tests can shorten it. This is a +// log-only signal: it never cancels an individual recovery, because the +// resume calls it drives run under context.WithoutCancel // (task_operations.go), so cancellation would not work anyway. var lifecycleSweepStuckWarningInterval = 60 * time.Second +// lifecycleSweepOverallDeadline bounds how long reconcileTaskLifecycleTokens +// waits for its worker pool before giving up and returning. It is a var so +// tests can shorten it. +// +// This deadline bounds admission and reporting, not an in-flight resume: the +// resume path a worker is waiting on runs under context.WithoutCancel +// (task_operations.go), so once a worker is inside waitForSessionReady this +// deadline cannot abort it. What it does bound is the sweep function itself +// returning, so a caller (Service.Stop, via stopLifecycleSweepAsync) is never +// left joining a goroutine for as long as the interactive +// constants.AgentLaunchTimeout budget allows. Workers still in flight when +// the deadline fires keep running detached in the background; their eventual +// completion is unreported. +var lifecycleSweepOverallDeadline = 5 * time.Minute + +// lifecycleSweepInFlight tracks which task IDs a worker is currently +// recovering, so a stuck-sweep warning (or a deadline-exceeded log) can name +// them instead of reporting only a count. +type lifecycleSweepInFlight struct { + mu sync.Mutex + ids map[string]struct{} +} + +func newLifecycleSweepInFlight() *lifecycleSweepInFlight { + return &lifecycleSweepInFlight{ids: make(map[string]struct{})} +} + +func (f *lifecycleSweepInFlight) start(taskID string) { + f.mu.Lock() + f.ids[taskID] = struct{}{} + f.mu.Unlock() +} + +func (f *lifecycleSweepInFlight) finish(taskID string) { + f.mu.Lock() + delete(f.ids, taskID) + f.mu.Unlock() +} + +func (f *lifecycleSweepInFlight) snapshot() []string { + f.mu.Lock() + defer f.mu.Unlock() + ids := make([]string, 0, len(f.ids)) + for id := range f.ids { + ids = append(ids, id) + } + return ids +} + // reconcileTaskLifecycleTokens scans durable lifecycle markers at startup. // A small fixed worker pool and bounded per-task attempts prevent a corrupted -// or repeatedly failing row from creating an unbounded goroutine storm. +// or repeatedly failing row from creating an unbounded goroutine storm. Job +// dispatch and the worker pool run in their own goroutines so that even if +// every worker is parked on a single task's interactive resume budget, this +// function's own wait is still bounded by lifecycleSweepOverallDeadline +// (or ctx cancellation) rather than by however long the worker pool takes to +// drain every job. func (s *Service) reconcileTaskLifecycleTokens(ctx context.Context) { lister, ok := s.repo.(lifecycleTaskMetadataLister) if !ok { @@ -999,13 +1057,22 @@ func (s *Service) reconcileTaskLifecycleTokens(ctx context.Context) { s.logger.Info("startup lifecycle sweep starting", zap.Int("task_count", taskCount)) var processed atomic.Int64 + inFlight := newLifecycleSweepInFlight() sweepDone := make(chan struct{}) warnDone := make(chan struct{}) go func() { defer close(warnDone) - s.warnIfLifecycleSweepStuck(taskCount, &processed, start, sweepDone) + s.warnIfLifecycleSweepStuck(taskCount, &processed, start, inFlight, sweepDone) }() + // dispatchCtx is a locally cancellable child of ctx so the deadline branch + // below can stop the dispatcher without waiting on ctx itself (owned by the + // caller, not this function) to be cancelled. Accepted workers keep using + // ctx: the deadline is an admission/reporting bound, not a cancellation of + // work that already started. + dispatchCtx, dispatchCancel := context.WithCancel(ctx) + defer dispatchCancel() + jobIDs := make(chan string) workerCount := 4 if len(jobs) < workerCount { @@ -1017,39 +1084,85 @@ func (s *Service) reconcileTaskLifecycleTokens(ctx context.Context) { go func() { defer wg.Done() for taskID := range jobIDs { + inFlight.start(taskID) s.recoverTaskLifecycleToken(ctx, taskID) + inFlight.finish(taskID) processed.Add(1) } }() } - for taskID := range jobs { - jobIDs <- taskID - } - close(jobIDs) - wg.Wait() - close(sweepDone) - <-warnDone + // Dispatch in its own goroutine: if every worker is parked on one task's + // interactive resume budget, sending the remaining job IDs would block + // this function on the same wait workersDone below is trying to bound. + // The dispatchCtx.Done branch matters independently of that: without it, + // once the deadline/cancel select case below fires and this function + // returns, an undispatched send blocks forever on the unbuffered channel + // — the dispatcher (and, transitively, the workers and the wg.Wait + // goroutine) outlives both the sweep's own deadline and Stop(). + go func() { + defer close(jobIDs) + for taskID := range jobs { + select { + case jobIDs <- taskID: + case <-dispatchCtx.Done(): + return + } + } + }() - s.logger.Info("startup lifecycle sweep finished", - zap.Int("task_count", taskCount), - zap.Duration("elapsed", time.Since(start))) -} + workersDone := make(chan struct{}) + go func() { + wg.Wait() + close(workersDone) + }() -// warnIfLifecycleSweepStuck logs a warning if the startup lifecycle sweep is -// still running after lifecycleSweepStuckWarningInterval, so a slow sweep is -// diagnosable from logs instead of requiring a goroutine dump. It is -// log-only: it never cancels the sweep, and it always exits (either the -// timer fires once, or done closes first). -func (s *Service) warnIfLifecycleSweepStuck(taskCount int, processed *atomic.Int64, start time.Time, done <-chan struct{}) { - timer := time.NewTimer(lifecycleSweepStuckWarningInterval) - defer timer.Stop() select { - case <-done: - case <-timer.C: - s.logger.Warn("startup lifecycle sweep still running", + case <-workersDone: + close(sweepDone) + <-warnDone + s.logger.Info("startup lifecycle sweep finished", zap.Int("task_count", taskCount), zap.Int64("processed", processed.Load()), zap.Duration("elapsed", time.Since(start))) + case <-ctx.Done(): + dispatchCancel() + close(sweepDone) + <-warnDone + s.logger.Warn("startup lifecycle sweep cancelled; abandoning wait, in-flight tasks continue recovering in the background", + zap.Int("task_count", taskCount), + zap.Int64("processed", processed.Load()), + zap.Strings("in_flight_task_ids", inFlight.snapshot()), + zap.Duration("elapsed", time.Since(start))) + case <-time.After(lifecycleSweepOverallDeadline): + dispatchCancel() + close(sweepDone) + <-warnDone + s.logger.Warn("startup lifecycle sweep exceeded its deadline; abandoning wait, in-flight tasks continue recovering in the background", + zap.Int("task_count", taskCount), + zap.Int64("processed", processed.Load()), + zap.Strings("in_flight_task_ids", inFlight.snapshot()), + zap.Duration("elapsed", time.Since(start))) + } +} + +// warnIfLifecycleSweepStuck logs a warning every lifecycleSweepStuckWarningInterval +// while the startup lifecycle sweep is still running, so a slow sweep is +// diagnosable from logs instead of requiring a goroutine dump. It is +// log-only: it never cancels the sweep, and it always exits once done closes. +func (s *Service) warnIfLifecycleSweepStuck(taskCount int, processed *atomic.Int64, start time.Time, inFlight *lifecycleSweepInFlight, done <-chan struct{}) { + ticker := time.NewTicker(lifecycleSweepStuckWarningInterval) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + s.logger.Warn("startup lifecycle sweep still running", + zap.Int("task_count", taskCount), + zap.Int64("processed", processed.Load()), + zap.Strings("in_flight_task_ids", inFlight.snapshot()), + zap.Duration("elapsed", time.Since(start))) + } } } @@ -1092,7 +1205,19 @@ func (s *Service) recoverTaskLifecycleAttempt(ctx context.Context, taskID string } } if manualMoveLifecycleCompleted(task) { - s.continueManualMoveLifecycle(ctx, taskID) + completedAt := task.UpdatedAt + if err := s.continueManualMoveLifecycle(ctx, taskID); err != nil { + return true + } + // Clear both markers in one compare-and-clear operation. A new manual + // move removes the old completed marker and writes its own pending + // marker in one task update, so an unconditional pending-key remove here + // could erase live recovery work. The update-generation predicate also + // prevents an already-completed newer move from being cleared before its + // feeder continuation runs. + if _, err := s.clearManualMoveLifecycleMarkersIfCompleted(ctx, taskID, completedAt); err != nil { + return true + } } if _, pending := task.Metadata[models.MetaKeyQueuePromotionPending]; pending { s.handleTaskQueuePromoted(ctx, watcher.TaskEventData{TaskID: taskID}) @@ -1992,20 +2117,47 @@ func (s *Service) persistManualMoveLifecycleCompletion(ctx context.Context, task return true } -func (s *Service) continueManualMoveLifecycle(ctx context.Context, taskID string) { +func (s *Service) continueManualMoveLifecycle(ctx context.Context, taskID string) error { latest, err := s.repo.GetTask(ctx, taskID) if err != nil || latest == nil { - return + if err != nil { + return err + } + return errors.New("manual move lifecycle task not found") } if !manualMoveLifecycleCompleted(latest) { - return + return nil } if s.feederPulls == nil { s.logger.Warn("manual move lifecycle has no feeder reconciler", zap.String("task_id", taskID)) - return + return errors.New("manual move lifecycle feeder reconciler is unavailable") } - s.feederPulls.ReconcileFeederPulls(ctx, latest.WorkflowID, latest.WorkflowStepID) + if err := s.feederPulls.ReconcileFeederPulls(ctx, latest.WorkflowID, latest.WorkflowStepID); err != nil { + s.logger.Warn("manual move lifecycle feeder reconciliation failed", + zap.String("task_id", taskID), zap.Error(err)) + return err + } + return nil +} + +func (s *Service) clearManualMoveLifecycleMarkersIfCompleted(ctx context.Context, taskID string, completedAt time.Time) (bool, error) { + cleaner, ok := s.repo.(manualMoveLifecycleMarkerCleaner) + if !ok { + return false, errors.New("repository cannot atomically clear manual move lifecycle markers") + } + cleared, err := cleaner.ClearManualMoveLifecycleMarkersIfCompleted(ctx, taskID, completedAt) + if err != nil { + s.logger.Warn("failed to clear manual move lifecycle markers", + zap.String("task_id", taskID), zap.Error(err)) + return false, err + } + if cleared { + if task, loadErr := s.repo.GetTask(ctx, taskID); loadErr == nil && task != nil { + s.publishTaskUpdated(ctx, task) + } + } + return cleared, nil } // processManualMoveLifecycleWithFeederBarrier runs the original move lifecycle diff --git a/apps/backend/internal/orchestrator/lifecycle_sweep_fixup_test.go b/apps/backend/internal/orchestrator/lifecycle_sweep_fixup_test.go new file mode 100644 index 0000000000..a615972063 --- /dev/null +++ b/apps/backend/internal/orchestrator/lifecycle_sweep_fixup_test.go @@ -0,0 +1,336 @@ +package orchestrator + +import ( + "context" + "sync" + "testing" + "time" + + eventbus "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" +) + +type manualMoveMarkerClearBarrierRepo struct { + sessionExecutorStore + started chan struct{} + release <-chan struct{} + once sync.Once +} + +func (r *manualMoveMarkerClearBarrierRepo) ClearManualMoveLifecycleMarkersIfCompleted(ctx context.Context, taskID string, completedAt time.Time) (bool, error) { + r.once.Do(func() { close(r.started) }) + <-r.release + return r.sessionExecutorStore.(manualMoveLifecycleMarkerCleaner). + ClearManualMoveLifecycleMarkersIfCompleted(ctx, taskID, completedAt) +} + +func TestRecoveryDoesNotClearPendingMarkerFromNewManualMove(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "manual-marker-race", "manual-marker-race-session", "destination-step") + if err := repo.SetTaskMetadataKey(ctx, "manual-marker-race", models.MetaKeyManualMoveLifecyclePending, + map[string]interface{}{"from_step_id": "old-source"}); err != nil { + t.Fatalf("seed pending marker: %v", err) + } + if err := repo.SetTaskMetadataKey(ctx, "manual-marker-race", models.MetaKeyManualMoveLifecycleCompleted, true); err != nil { + t.Fatalf("seed completed marker: %v", err) + } + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.SetFeederPullReconciler(&countingFeederPullReconciler{}) + release := make(chan struct{}) + guarded := &manualMoveMarkerClearBarrierRepo{ + sessionExecutorStore: repo, + started: make(chan struct{}), + release: release, + } + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + }) + svc.repo = guarded + + recoveryDone := make(chan bool, 1) + go func() { recoveryDone <- svc.recoverTaskLifecycleAttempt(ctx, "manual-marker-race") }() + select { + case <-guarded.started: + case <-time.After(time.Second): + t.Fatal("recovery did not reach the atomic marker clear") + } + + // MoveTaskWithOptions performs this replacement in one task write. Model the + // committed result directly so the test controls the exact interleaving. + task, err := repo.GetTask(ctx, "manual-marker-race") + if err != nil { + t.Fatalf("load task for new move: %v", err) + } + task.WorkflowStepID = "new-destination-step" + task.Metadata[models.MetaKeyManualMoveLifecyclePending] = map[string]interface{}{"from_step_id": "new-source"} + delete(task.Metadata, models.MetaKeyManualMoveLifecycleCompleted) + if err := repo.UpdateTask(ctx, task); err != nil { + t.Fatalf("commit newer manual move: %v", err) + } + close(release) + + if retry := <-recoveryDone; !retry { + t.Fatal("recovery did not report remaining lifecycle work") + } + stored, err := repo.GetTask(ctx, "manual-marker-race") + if err != nil { + t.Fatalf("reload task after recovery: %v", err) + } + if _, pending := stored.Metadata[models.MetaKeyManualMoveLifecyclePending]; !pending { + t.Fatal("new manual move pending marker was cleared by old recovery") + } + if _, completed := stored.Metadata[models.MetaKeyManualMoveLifecycleCompleted]; completed { + t.Fatal("new manual move unexpectedly retained old completion marker") + } +} + +func TestRecoveryRetriesWhenNewManualMoveAlreadyCompleted(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "manual-marker-completed-race", "manual-marker-completed-race-session", "destination-step") + if err := repo.SetTaskMetadataKey(ctx, "manual-marker-completed-race", models.MetaKeyManualMoveLifecyclePending, + map[string]interface{}{"from_step_id": "old-source"}); err != nil { + t.Fatalf("seed pending marker: %v", err) + } + if err := repo.SetTaskMetadataKey(ctx, "manual-marker-completed-race", models.MetaKeyManualMoveLifecycleCompleted, true); err != nil { + t.Fatalf("seed completed marker: %v", err) + } + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.SetFeederPullReconciler(&countingFeederPullReconciler{}) + release := make(chan struct{}) + guarded := &manualMoveMarkerClearBarrierRepo{ + sessionExecutorStore: repo, + started: make(chan struct{}), + release: release, + } + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + }) + svc.repo = guarded + + recoveryDone := make(chan bool, 1) + go func() { recoveryDone <- svc.recoverTaskLifecycleAttempt(ctx, "manual-marker-completed-race") }() + select { + case <-guarded.started: + case <-time.After(time.Second): + t.Fatal("recovery did not reach the atomic marker clear") + } + + // A newer move can finish its lifecycle before the stale recovery clears + // the old completion marker. Its completion marker has the same boolean + // value, but its task update generation is different. + task, err := repo.GetTask(ctx, "manual-marker-completed-race") + if err != nil { + t.Fatalf("load task for newer completion: %v", err) + } + delete(task.Metadata, models.MetaKeyManualMoveLifecyclePending) + task.Metadata[models.MetaKeyManualMoveLifecycleCompleted] = true + if err := repo.UpdateTask(ctx, task); err != nil { + t.Fatalf("commit newer completed move: %v", err) + } + close(release) + + if retry := <-recoveryDone; !retry { + t.Fatal("recovery did not retain a newer completed marker for retry") + } + stored, err := repo.GetTask(ctx, "manual-marker-completed-race") + if err != nil { + t.Fatalf("reload task after recovery: %v", err) + } + if _, completed := stored.Metadata[models.MetaKeyManualMoveLifecycleCompleted]; !completed { + t.Fatal("new manual move completion marker was cleared by old recovery") + } +} + +type failingManualMoveFeederPullReconciler struct{} + +func (failingManualMoveFeederPullReconciler) ReconcileFeederPulls(context.Context, string, string) error { + return errFeederReconcileTest +} + +var errFeederReconcileTest = &feederReconcileTestError{} + +type feederReconcileTestError struct{} + +func (*feederReconcileTestError) Error() string { return "feeder reconciliation failed" } + +func TestFailedManualMoveContinuationRetainsCompletionMarker(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "manual-marker-error", "manual-marker-error-session", "step1") + if err := repo.SetTaskMetadataKey(ctx, "manual-marker-error", models.MetaKeyManualMoveLifecycleCompleted, true); err != nil { + t.Fatalf("seed completed marker: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.SetFeederPullReconciler(failingManualMoveFeederPullReconciler{}) + + if retry := svc.recoverTaskLifecycleAttempt(ctx, "manual-marker-error"); !retry { + t.Fatal("failed continuation did not request a retry") + } + stored, err := repo.GetTask(ctx, "manual-marker-error") + if err != nil { + t.Fatalf("reload task: %v", err) + } + if _, completed := stored.Metadata[models.MetaKeyManualMoveLifecycleCompleted]; !completed { + t.Fatal("completion marker was cleared after feeder reconciliation failed") + } +} + +type watcherStateLifecycleRepo struct { + sessionExecutorStore + watcher *watcher.Watcher + observed chan bool + release <-chan struct{} + once sync.Once +} + +func (r *watcherStateLifecycleRepo) ListTasksWithMetadataKey(ctx context.Context, key string) ([]*models.Task, error) { + r.once.Do(func() { + r.observed <- r.watcher.IsRunning() + }) + <-r.release + return r.sessionExecutorStore.(lifecycleTaskMetadataLister).ListTasksWithMetadataKey(ctx, key) +} + +func TestStartupSweepReadsTokensAfterWatcherStarts(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "watcher-ordering", "watcher-ordering-session", "step1") + if err := repo.SetTaskMetadataKey(ctx, "watcher-ordering", models.MetaKeyQueuePromotionPending, true); err != nil { + t.Fatalf("seed promotion marker: %v", err) + } + log := testLogger() + eventBus := eventbus.NewMemoryEventBus(log) + svc := NewService(DefaultServiceConfig(), eventBus, &mockAgentManager{}, newMockTaskRepo(), repo, nil, nil, nil, log) + svc.SetTurnService(&repoTurnService{repo: repo}) + release := make(chan struct{}) + observed := make(chan bool, 1) + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + if svc.IsRunning() { + if err := svc.Stop(); err != nil { + t.Errorf("Stop: %v", err) + } + } + }) + svc.repo = &watcherStateLifecycleRepo{ + sessionExecutorStore: repo, + watcher: svc.watcher, + observed: observed, + release: release, + } + startDone := make(chan error, 1) + go func() { startDone <- svc.Start(ctx) }() + + select { + case running := <-observed: + if !running { + close(release) + <-startDone + t.Fatal("startup lifecycle sweep read tokens before watcher subscriptions") + } + case <-time.After(2 * time.Second): + close(release) + t.Fatal("Start and lifecycle sweep did not make progress") + } + close(release) + select { + case err := <-startDone: + if err != nil { + t.Fatalf("Start: %v", err) + } + case <-time.After(time.Second): + t.Fatal("Start did not return") + } +} + +type deadlineWorkContextRepo struct { + sessionExecutorStore + release <-chan struct{} + entered chan struct{} + observed chan bool + enteredOnce sync.Once + observedOnce sync.Once +} + +func (r *deadlineWorkContextRepo) GetTask(ctx context.Context, id string) (*models.Task, error) { + r.enteredOnce.Do(func() { close(r.entered) }) + <-r.release + r.observedOnce.Do(func() { r.observed <- ctx.Err() != nil }) + return r.sessionExecutorStore.GetTask(ctx, id) +} + +func (r *deadlineWorkContextRepo) ListTasksWithMetadataKey(ctx context.Context, key string) ([]*models.Task, error) { + return r.sessionExecutorStore.(lifecycleTaskMetadataLister).ListTasksWithMetadataKey(ctx, key) +} + +func TestLifecycleSweepDeadlineDoesNotCancelAcceptedWorkContext(t *testing.T) { + previousDeadline := lifecycleSweepOverallDeadline + lifecycleSweepOverallDeadline = 20 * time.Millisecond + t.Cleanup(func() { lifecycleSweepOverallDeadline = previousDeadline }) + + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "deadline-context", "deadline-context-session", "step1") + if err := repo.SetTaskMetadataKey(ctx, "deadline-context", models.MetaKeyQueuePromotionPending, true); err != nil { + t.Fatalf("seed promotion marker: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + release := make(chan struct{}) + observed := make(chan bool, 1) + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + }) + svc.repo = &deadlineWorkContextRepo{ + sessionExecutorStore: repo, + release: release, + entered: make(chan struct{}), + observed: observed, + } + sweepDone := make(chan struct{}) + go func() { + svc.reconcileTaskLifecycleTokens(ctx) + close(sweepDone) + }() + select { + case <-svc.repo.(*deadlineWorkContextRepo).entered: + case <-time.After(time.Second): + close(release) + t.Fatal("worker was not admitted before the deadline") + } + select { + case <-sweepDone: + case <-time.After(time.Second): + close(release) + t.Fatal("sweep did not return at its deadline") + } + close(release) + select { + case canceled := <-observed: + if canceled { + t.Fatal("deadline canceled a worker that was already admitted") + } + case <-time.After(time.Second): + t.Fatal("admitted worker did not finish after deadline") + } +} diff --git a/apps/backend/internal/orchestrator/lifecycle_sweep_logging_test.go b/apps/backend/internal/orchestrator/lifecycle_sweep_logging_test.go index dc2b15feb6..b00f253449 100644 --- a/apps/backend/internal/orchestrator/lifecycle_sweep_logging_test.go +++ b/apps/backend/internal/orchestrator/lifecycle_sweep_logging_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "testing" "time" @@ -13,6 +14,8 @@ import ( commonlogger "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + v1 "github.com/kandev/kandev/pkg/api/v1" ) // blockingLifecycleRepo blocks GetTask until release is closed, letting a @@ -34,6 +37,83 @@ func (r *blockingLifecycleRepo) ListTasksWithMetadataKey(ctx context.Context, ke }).ListTasksWithMetadataKey(ctx, key) } +// dispatchTrackingLifecycleRepo behaves like blockingLifecycleRepo but also +// records which task IDs GetTask was called for, so a test can assert a job +// was never dispatched to a worker in the first place (as opposed to merely +// still being in flight). +type dispatchTrackingLifecycleRepo struct { + sessionExecutorStore + release <-chan struct{} + mu sync.Mutex + seen map[string]int + done chan struct{} +} + +func (r *dispatchTrackingLifecycleRepo) GetTask(ctx context.Context, id string) (*models.Task, error) { + r.mu.Lock() + if r.seen == nil { + r.seen = make(map[string]int) + } + r.seen[id]++ + r.mu.Unlock() + <-r.release + task, err := r.sessionExecutorStore.GetTask(ctx, id) + if r.done != nil { + select { + case r.done <- struct{}{}: + default: + } + } + return task, err +} + +func (r *dispatchTrackingLifecycleRepo) ListTasksWithMetadataKey(ctx context.Context, key string) ([]*models.Task, error) { + return r.sessionExecutorStore.(interface { + ListTasksWithMetadataKey(context.Context, string) ([]*models.Task, error) + }).ListTasksWithMetadataKey(ctx, key) +} + +func (r *dispatchTrackingLifecycleRepo) callCount(id string) int { + r.mu.Lock() + defer r.mu.Unlock() + return r.seen[id] +} + +// seedTaskAndSessionInWorkspace creates a task and session against an +// already-seeded workspace/workflow (e.g. from seedSession), for tests that +// need several tasks in the same sweep without recreating those rows. +func seedTaskAndSessionInWorkspace(t *testing.T, repo *sqliterepo.Repository, workspaceID, workflowID, taskID, sessionID, workflowStepID string) { + t.Helper() + ctx := context.Background() + now := time.Now().UTC() + + task := &models.Task{ + ID: taskID, + WorkspaceID: workspaceID, + WorkflowID: workflowID, + WorkflowStepID: workflowStepID, + Title: "Test Task", + Description: "Test", + State: v1.TaskStateInProgress, + CreatedAt: now, + UpdatedAt: now, + } + if err := repo.CreateTask(ctx, task); err != nil { + t.Fatalf("failed to create task %s: %v", taskID, err) + } + + session := &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + State: models.TaskSessionStateRunning, + StartedAt: now, + UpdatedAt: now, + } + if err := repo.CreateTaskSession(ctx, session); err != nil { + t.Fatalf("failed to create task session for %s: %v", taskID, err) + } +} + func observedTestLogger(t *testing.T) (*commonlogger.Logger, *observer.ObservedLogs) { t.Helper() core, logs := observer.New(zapcore.DebugLevel) @@ -153,3 +233,278 @@ func TestReconcileTaskLifecycleTokensWarnsWhenStuck(t *testing.T) { t.Fatal("sweep did not finish after release") } } + +// observedTestLoggerCounting is observedTestLoggerWatching plus a threshold: +// the returned channel closes once the message has been logged atLeast +// times, letting a test prove a warning repeats instead of firing once. +func observedTestLoggerCounting(t *testing.T, message string, atLeast int) (*commonlogger.Logger, <-chan struct{}) { + t.Helper() + core, _ := observer.New(zapcore.DebugLevel) + + var count atomic.Int32 + seen := make(chan struct{}) + var once sync.Once + hooked := zapcore.RegisterHooks(core, func(entry zapcore.Entry) error { + if entry.Message == message && int(count.Add(1)) >= atLeast { + once.Do(func() { close(seen) }) + } + return nil + }) + + log, err := commonlogger.NewFromZap(zap.New(hooked)) + if err != nil { + t.Fatalf("create observed logger: %v", err) + } + return log, seen +} + +// TestReconcileTaskLifecycleTokensWarnsRepeatedlyWhenStuck is the regression +// test for the stuck-sweep warning becoming a repeating ticker instead of a +// one-shot timer: a sweep still running after multiple +// lifecycleSweepStuckWarningInterval periods must warn every period, not +// just the first. +// +// Expected pre-fix failure: warnIfLifecycleSweepStuck uses a one-shot +// time.Timer, so only one "startup lifecycle sweep still running" entry is +// ever logged and this test times out waiting for a second. +func TestReconcileTaskLifecycleTokensWarnsRepeatedlyWhenStuck(t *testing.T) { + prevInterval := lifecycleSweepStuckWarningInterval + lifecycleSweepStuckWarningInterval = 20 * time.Millisecond + t.Cleanup(func() { lifecycleSweepStuckWarningInterval = prevInterval }) + + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "stuck-task-repeat", "stuck-session-repeat", "step1") + if err := repo.SetTaskMetadataKey(ctx, "stuck-task-repeat", models.MetaKeyQueuePromotionPending, true); err != nil { + t.Fatalf("seed promotion token: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + log, warnedTwice := observedTestLoggerCounting(t, "startup lifecycle sweep still running", 2) + svc.logger = log + + release := make(chan struct{}) + svc.repo = &blockingLifecycleRepo{sessionExecutorStore: repo, release: release} + + done := make(chan struct{}) + go func() { + defer close(done) + svc.reconcileTaskLifecycleTokens(ctx) + }() + + select { + case <-warnedTwice: + case <-time.After(2 * time.Second): + close(release) + t.Fatal("stuck-sweep warning did not repeat") + } + + close(release) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("sweep did not finish after release") + } +} + +// TestReconcileTaskLifecycleTokensDeadlineIncludesInFlightTaskIDs is the +// regression test for the sweep's overall-deadline exit branch: a worker +// still parked on a per-task resume when lifecycleSweepOverallDeadline fires +// must be named in the log, not just counted, and the function must return +// instead of waiting out the worker's own (much longer) interactive budget. +// +// Expected pre-fix failure: lifecycleSweepOverallDeadline does not exist, so +// this fails to compile. +func TestReconcileTaskLifecycleTokensDeadlineIncludesInFlightTaskIDs(t *testing.T) { + prevDeadline := lifecycleSweepOverallDeadline + lifecycleSweepOverallDeadline = 20 * time.Millisecond + t.Cleanup(func() { lifecycleSweepOverallDeadline = prevDeadline }) + + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "deadline-task", "deadline-session", "step1") + if err := repo.SetTaskMetadataKey(ctx, "deadline-task", models.MetaKeyQueuePromotionPending, true); err != nil { + t.Fatalf("seed promotion token: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + log, logs := observedTestLogger(t) + svc.logger = log + + release := make(chan struct{}) + svc.repo = &blockingLifecycleRepo{sessionExecutorStore: repo, release: release} + + done := make(chan struct{}) + go func() { + defer close(done) + svc.reconcileTaskLifecycleTokens(ctx) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + close(release) + t.Fatal("sweep did not return by its overall deadline while a worker was parked") + } + close(release) + + entries := logs.FilterMessage("startup lifecycle sweep exceeded its deadline; abandoning wait, in-flight tasks continue recovering in the background").All() + if len(entries) != 1 { + t.Fatalf("deadline-exceeded logs = %#v, want exactly one", entries) + } + ids, ok := entries[0].ContextMap()["in_flight_task_ids"].([]interface{}) + if !ok || len(ids) != 1 || ids[0] != "deadline-task" { + t.Fatalf("in_flight_task_ids = %#v, want [\"deadline-task\"]", entries[0].ContextMap()["in_flight_task_ids"]) + } +} + +// TestReconcileTaskLifecycleTokensDeadlineStopsUndispatchedWork is the +// regression test for the sweep's dispatch loop outliving its own overall +// deadline: only ctx.Done() stopped the dispatcher, so once every worker was +// parked on a per-task resume and the deadline fired, the function returned +// but the dispatcher kept the job channel open — any worker that later freed +// up (e.g. once its parked resume unblocked) would immediately pick up the +// next undispatched job, admitting new recovery work well past the advertised +// deadline instead of stopping at it. +// +// Five jobs with a worker pool of four guarantees one job is still sitting +// undispatched in the dispatcher's loop when the (short) deadline fires. +// jobs is a map, so which of the five ends up as that leftover job is +// randomized by Go's map iteration order — the test finds it by call count +// rather than assuming a fixed name. +// +// Expected pre-fix failure: after the parked workers are released, one of +// them picks up the previously-undispatched job and calls GetTask for it, so +// its call count goes from 0 to nonzero. +func TestReconcileTaskLifecycleTokensDeadlineStopsUndispatchedWork(t *testing.T) { + prevDeadline := lifecycleSweepOverallDeadline + lifecycleSweepOverallDeadline = 20 * time.Millisecond + t.Cleanup(func() { lifecycleSweepOverallDeadline = prevDeadline }) + + ctx := context.Background() + repo := setupTestRepo(t) + taskIDs := []string{"deadline-fan-1", "deadline-fan-2", "deadline-fan-3", "deadline-fan-4", "deadline-fan-5"} + seedSession(t, repo, taskIDs[0], taskIDs[0]+"-session", "step1") + for _, id := range taskIDs[1:] { + seedTaskAndSessionInWorkspace(t, repo, "ws1", "wf1", id, id+"-session", "step1") + } + for _, id := range taskIDs { + if err := repo.SetTaskMetadataKey(ctx, id, models.MetaKeyQueuePromotionPending, true); err != nil { + t.Fatalf("seed promotion token for %s: %v", id, err) + } + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + release := make(chan struct{}) + workerDone := make(chan struct{}, 4) + tracking := &dispatchTrackingLifecycleRepo{sessionExecutorStore: repo, release: release, done: workerDone} + svc.repo = tracking + + done := make(chan struct{}) + go func() { + defer close(done) + svc.reconcileTaskLifecycleTokens(ctx) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + close(release) + t.Fatal("sweep did not return by its overall deadline while workers were parked") + } + + // reconcileTaskLifecycleTokens has already returned, so exactly + // workerCount (4) of the 5 jobs should have reached a worker and be + // blocked in GetTask; find the one that was never dispatched before + // releasing the rest. + var undispatched string + var dispatchedBeforeDeadline int + for _, id := range taskIDs { + if tracking.callCount(id) == 0 { + undispatched = id + } else { + dispatchedBeforeDeadline++ + } + } + if undispatched == "" || dispatchedBeforeDeadline != 4 { + t.Fatalf("want exactly 4 of 5 jobs dispatched before the deadline and 1 left over, got %d dispatched, undispatched=%q", dispatchedBeforeDeadline, undispatched) + } + + // Releasing the four parked workers and waiting for each admitted call to + // return proves the leftover job was never handed out, rather than merely + // not yet handed out. There is no production completion signal for this + // detached worker pool to synchronize on instead (see + // stopLifecycleSweepAsync's own comment on this being an intentionally + // un-joined background sweep). + close(release) + for i := 0; i < 4; i++ { + select { + case <-workerDone: + case <-time.After(time.Second): + t.Fatal("parked lifecycle worker did not finish after release") + } + } + + if got := tracking.callCount(undispatched); got != 0 { + t.Fatalf("job %q was dispatched to a worker after the sweep's overall deadline had already fired: called %d times", undispatched, got) + } +} + +// TestReconcileTaskLifecycleTokensCancelIncludesInFlightTaskIDs is the +// regression test for the sweep's ctx-cancellation exit branch: cancelling +// the sweep's context while a worker is parked must return promptly and name +// the in-flight task, mirroring the deadline branch above. +// +// Expected pre-fix failure: the pre-fix "sweep cancelled" log line reported +// only a task count, so in_flight_task_ids is absent from its context. +func TestReconcileTaskLifecycleTokensCancelIncludesInFlightTaskIDs(t *testing.T) { + prevInterval := lifecycleSweepStuckWarningInterval + lifecycleSweepStuckWarningInterval = 20 * time.Millisecond + t.Cleanup(func() { lifecycleSweepStuckWarningInterval = prevInterval }) + + ctx, cancel := context.WithCancel(context.Background()) + repo := setupTestRepo(t) + seedSession(t, repo, "cancel-task", "cancel-session", "step1") + if err := repo.SetTaskMetadataKey(ctx, "cancel-task", models.MetaKeyQueuePromotionPending, true); err != nil { + t.Fatalf("seed promotion token: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + log, logs, stuck := observedTestLoggerWatching(t, "startup lifecycle sweep still running") + svc.logger = log + + release := make(chan struct{}) + svc.repo = &blockingLifecycleRepo{sessionExecutorStore: repo, release: release} + + done := make(chan struct{}) + go func() { + defer close(done) + svc.reconcileTaskLifecycleTokens(ctx) + }() + + // Wait for proof the worker is actually in flight before cancelling, so + // the cancellation log's in_flight_task_ids reflects the parked worker + // instead of racing an empty snapshot. + select { + case <-stuck: + case <-time.After(2 * time.Second): + close(release) + cancel() + t.Fatal("worker never reported in-flight before cancellation") + } + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + close(release) + t.Fatal("sweep did not return after ctx cancellation while a worker was parked") + } + close(release) + + entries := logs.FilterMessage("startup lifecycle sweep cancelled; abandoning wait, in-flight tasks continue recovering in the background").All() + if len(entries) != 1 { + t.Fatalf("cancelled logs = %#v, want exactly one", entries) + } + ids, ok := entries[0].ContextMap()["in_flight_task_ids"].([]interface{}) + if !ok || len(ids) != 1 || ids[0] != "cancel-task" { + t.Fatalf("in_flight_task_ids = %#v, want [\"cancel-task\"]", entries[0].ContextMap()["in_flight_task_ids"]) + } +} diff --git a/apps/backend/internal/orchestrator/lifecycle_sweep_startup_test.go b/apps/backend/internal/orchestrator/lifecycle_sweep_startup_test.go new file mode 100644 index 0000000000..0e9469c25e --- /dev/null +++ b/apps/backend/internal/orchestrator/lifecycle_sweep_startup_test.go @@ -0,0 +1,355 @@ +package orchestrator + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + eventbus "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// TestStartDoesNotBlockOnLifecycleSweep is the regression test for the +// ordering defect described in +// docs/specs/startup-listener-before-recovery/spec.md: reconcileTaskLifecycleTokens +// used to run synchronously inside Service.Start, before the watcher +// subscribed. A pending lifecycle token whose recovery blocks (here, +// deliberately, via blockingLifecycleRepo) therefore held Start — and with +// it every /api/v1/* route behind the bootstrap 503 handler — open +// indefinitely. +// +// Expected pre-fix failure: Start calls reconcileTaskLifecycleTokens +// synchronously before returning, so it blocks on blockingLifecycleRepo's +// release channel and the test times out. +func TestStartDoesNotBlockOnLifecycleSweep(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "sweep-block-task", "sweep-block-session", "step1") + if err := repo.SetTaskMetadataKey(ctx, "sweep-block-task", models.MetaKeyQueuePromotionPending, true); err != nil { + t.Fatalf("seed promotion token: %v", err) + } + + log := testLogger() + eventBus := eventbus.NewMemoryEventBus(log) + cfg := DefaultServiceConfig() + svc := NewService(cfg, eventBus, &mockAgentManager{}, newMockTaskRepo(), repo, nil, nil, nil, log) + svc.SetTurnService(&repoTurnService{repo: repo}) + + // Swap in the blocking wrapper only after construction: NewService's + // internal executor/scheduler wiring needs the real repoStore, but the + // startup sweep itself reads through s.repo (the narrower + // sessionExecutorStore field), so this is the same seam + // lifecycle_sweep_logging_test.go uses. + release := make(chan struct{}) + svc.repo = &blockingLifecycleRepo{sessionExecutorStore: repo, release: release} + + startErr := make(chan error, 1) + go func() { startErr <- svc.Start(ctx) }() + + select { + case err := <-startErr: + if err != nil { + close(release) + t.Fatalf("Start returned error: %v", err) + } + case <-time.After(2 * time.Second): + close(release) + t.Fatal("Start blocked on the startup lifecycle sweep instead of running it in the background") + } + + close(release) + t.Cleanup(func() { + if err := svc.Stop(); err != nil { + t.Errorf("Stop: %v", err) + } + }) +} + +// countingFeederPullReconciler counts ReconcileFeederPulls calls without +// asserting on the arguments, for tests that only care how many times the +// sweep replayed a continuation. +type countingFeederPullReconciler struct { + calls atomic.Int32 + done chan struct{} + once sync.Once +} + +func (r *countingFeederPullReconciler) ReconcileFeederPulls(context.Context, string, string) error { + r.calls.Add(1) + if r.done != nil { + r.once.Do(func() { close(r.done) }) + } + return nil +} + +// TestRecoverTaskLifecycleTokenClearsInertCompletedToken is the regression +// test for the non-convergence defect: a task carrying only +// MetaKeyManualMoveLifecycleCompleted (its pending token already cleared) +// must have its continuation replayed exactly once and the completed marker +// cleared, not be re-listed and re-replayed on every future startup sweep. +// +// Expected pre-fix failure: recoverTaskLifecycleAttempt's exit predicate +// counts manualMoveLifecycleCompleted as "still pending", so +// recoverTaskLifecycleToken retries maxAttempts times (3 continuation +// replays) and the completed token is never cleared. +func TestRecoverTaskLifecycleTokenClearsInertCompletedToken(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "completed-only-task", "completed-only-session", "step1") + if err := repo.SetTaskMetadataKey(ctx, "completed-only-task", models.MetaKeyManualMoveLifecycleCompleted, true); err != nil { + t.Fatalf("seed completed token: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + recorder := &countingFeederPullReconciler{} + svc.SetFeederPullReconciler(recorder) + publisher := &recordingTaskUpdatedPublisher{} + svc.SetTaskEventPublisher(publisher) + + svc.recoverTaskLifecycleToken(ctx, "completed-only-task") + + if got := recorder.calls.Load(); got != 1 { + t.Fatalf("feeder reconcile calls = %d, want exactly 1 (no retries on an inert token)", got) + } + stored, err := repo.GetTask(ctx, "completed-only-task") + if err != nil { + t.Fatalf("reload task: %v", err) + } + if _, completed := stored.Metadata[models.MetaKeyManualMoveLifecycleCompleted]; completed { + t.Fatal("inert manual move lifecycle completion token was not cleared") + } + if len(publisher.updatedTaskIDs) != 1 || publisher.updatedTaskIDs[0] != "completed-only-task" { + t.Fatalf("task.updated publishes = %v, want one update for completed-only-task", publisher.updatedTaskIDs) + } +} + +// TestRecoverTaskLifecycleTokenDoesNotReplayCoexistingPendingAndCompletedTokens +// is the regression test for the crash-window defect: persistManualMoveLifecycleCompletion +// writes MetaKeyManualMoveLifecycleCompleted and clears +// MetaKeyManualMoveLifecyclePending in two separate calls, so a crash (or a +// failed remove) between them can leave both keys set on a move that already +// ran its step exit/enter side effects. manualMoveLifecyclePending is defined +// as pending && !completed, so it is always false while completed is set and +// cannot guard the completed-token clear against resurrecting the leftover +// pending token. +// +// Expected pre-fix failure: the completed marker is cleared on the first +// attempt because the guard checks the always-false manualMoveLifecyclePending +// helper instead of the raw key; the retry loop then treats the leftover raw +// pending token as a live move and a second attempt calls +// recoverManualMoveLifecycle, replaying the step exit/enter side effects for a +// lifecycle that already finished. +// +// It also covers the follow-on defect the raw-key fix introduced by itself: +// once the guard correctly detects the coexisting pending token, it must not +// simply leave both keys in place — that just swaps "replays forever" for +// "never converges," re-listing this task and re-running its continuation on +// every future startup. The stale pending token has to be cleared directly +// (not through recoverManualMoveLifecycle, which would replay the move), and +// the completed token cleared alongside it, so both converge in one pass. +func TestRecoverTaskLifecycleTokenDoesNotReplayCoexistingPendingAndCompletedTokens(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "coexist-task", "coexist-session", "destination-step") + task, err := repo.GetTask(ctx, "coexist-task") + if err != nil { + t.Fatalf("load task: %v", err) + } + task.Metadata = map[string]interface{}{ + models.MetaKeyManualMoveLifecyclePending: map[string]interface{}{"from_step_id": "source-step"}, + models.MetaKeyManualMoveLifecycleCompleted: true, + } + if err := repo.UpdateTask(ctx, task); err != nil { + t.Fatalf("persist coexisting tokens: %v", err) + } + + steps := newMockStepGetter() + steps.steps["source-step"] = &wfmodels.WorkflowStep{ + ID: "source-step", WorkflowID: "wf1", Name: "Source", + Events: wfmodels.StepEvents{OnExit: []wfmodels.OnExitAction{{Type: wfmodels.OnExitDisablePlanMode}}}, + } + steps.steps["destination-step"] = &wfmodels.WorkflowStep{ + ID: "destination-step", WorkflowID: "wf1", Name: "Destination", + Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{{ + Type: wfmodels.OnEnterSetSessionMode, + Config: map[string]interface{}{"mode": "destination"}, + }}}, + } + svc := createTestService(repo, steps, newMockTaskRepo()) + var replays atomic.Int32 + svc.onManualMoveLifecycleStart = func() { replays.Add(1) } + svc.SetFeederPullReconciler(&countingFeederPullReconciler{}) + + svc.recoverTaskLifecycleToken(ctx, "coexist-task") + + if got := replays.Load(); got != 0 { + t.Fatalf("manual move lifecycle side effects replayed %d times, want 0 (move already completed)", got) + } + stored, err := repo.GetTask(ctx, "coexist-task") + if err != nil { + t.Fatalf("reload task: %v", err) + } + if _, pending := stored.Metadata[models.MetaKeyManualMoveLifecyclePending]; pending { + t.Fatal("stale manual move lifecycle pending token was not cleared; sweep will never converge") + } + if _, completed := stored.Metadata[models.MetaKeyManualMoveLifecycleCompleted]; completed { + t.Fatal("manual move lifecycle completion token was not cleared; sweep will never converge") + } +} + +// TestStartLifecycleSweepAsyncSkipsWhenServiceStopped is the regression test +// for the unsynchronized lifecycleSweepCancel access: Start sets s.running +// before it reaches startLifecycleSweepAsync, so a Stop racing in that window +// used to read a still-nil lifecycleSweepCancel, do nothing, and let Start go +// on to launch an uncancellable sweep after shutdown had already begun. +// Mirrors TestLaunchDynamicSuccessorDetached_SkipsWhenServiceStopped. +// +// Expected pre-fix failure: startLifecycleSweepAsync has no stopped check, so +// it always launches the sweep and there is no boolean return value to assert +// against (this test would not compile against the pre-fix signature). +func TestStartLifecycleSweepAsyncSkipsWhenServiceStopped(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + // Simulate a Stop that wins the race before Start ever calls + // startLifecycleSweepAsync. + svc.stopLifecycleSweepAsync() + + if svc.startLifecycleSweepAsync(ctx) { + t.Fatal("a stopped service launched a startup lifecycle sweep it will never cancel") + } + if svc.lifecycleSweepCancel != nil { + t.Fatal("a skipped sweep must not leave a cancel func for a later Stop to find") + } + + svc.resetLifecycleSweepWorkers() + if !svc.startLifecycleSweepAsync(ctx) { + t.Fatal("a restarted service refused to launch the startup lifecycle sweep") + } + svc.stopLifecycleSweepAsync() +} + +// TestLifecycleSweepStartRacingStopDoesNotMisuseWaitGroup is the regression +// test for lifecycleSweepWorkers.Add being published outside lifecycleSweepMu: +// startLifecycleSweepAsync used to unlock the mutex and only then call +// lifecycleSweepWorkers.Add(1), so a concurrent stopLifecycleSweepAsync could +// take the lock, read the (already-published) cancel func, and call +// lifecycleSweepWorkers.Wait() before Add(1) ran — the Go runtime's race +// detector treats an Add not ordered before a concurrent Wait as WaitGroup +// misuse, because nothing in the source establishes a happens-before edge +// between them. The fix moves Add(1) inside the same critical section as the +// cancel-func publish, matching dynamic_launch.go's +// launchDynamicSuccessorDetached. Run with `-race`; iterated because the +// unsynchronized window pre-fix is only a couple of instructions wide, so any +// single iteration has low odds of landing in it — repetition raises the +// odds of catching a reintroduced regression without claiming certainty. +// +// Expected pre-fix failure: `go test -race` reports a data race / WaitGroup +// misuse on lifecycleSweepWorkers (probabilistically, across repeated runs). +func TestLifecycleSweepStartRacingStopDoesNotMisuseWaitGroup(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + const iterations = 200 + for i := 0; i < iterations; i++ { + svc.resetLifecycleSweepWorkers() + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + svc.startLifecycleSweepAsync(ctx) + }() + go func() { + defer wg.Done() + svc.stopLifecycleSweepAsync() + }() + wg.Wait() + } +} + +// TestLifecycleSweepConcurrentWithLiveTaskMovedRunsManualMoveSideEffectsOnce +// is the regression test for the spec's own concurrent-safety claim +// (docs/specs/startup-listener-before-recovery/spec.md, Desired-behaviour): +// running the startup sweep concurrently with live watcher/scheduler traffic +// is safe because of the per-task lock (queuedMoveLifecycleLocks) — "this +// must be confirmed by test, not by assertion." Runs the sweep's recovery +// path (recoverTaskLifecycleToken) concurrently with a live task.moved +// redelivery of the same admitted manual move (handleTaskMoved) and asserts +// the step exit/enter side effects run exactly once regardless of which path +// wins the lock. +func TestLifecycleSweepConcurrentWithLiveTaskMovedRunsManualMoveSideEffectsOnce(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "concurrent-manual-move-task", "concurrent-manual-move-session", "source-step") + task, err := repo.GetTask(ctx, "concurrent-manual-move-task") + if err != nil { + t.Fatalf("load task: %v", err) + } + task.WorkflowStepID = "destination-step" + task.WIPAdmitted = true + task.Metadata = map[string]interface{}{ + models.MetaKeyManualMoveLifecyclePending: map[string]interface{}{"from_step_id": "source-step"}, + } + if err := repo.UpdateTask(ctx, task); err != nil { + t.Fatalf("persist admitted move: %v", err) + } + + steps := newMockStepGetter() + steps.steps["source-step"] = &wfmodels.WorkflowStep{ + ID: "source-step", WorkflowID: "wf1", Name: "Source", + Events: wfmodels.StepEvents{OnExit: []wfmodels.OnExitAction{{Type: wfmodels.OnExitDisablePlanMode}}}, + } + steps.steps["destination-step"] = &wfmodels.WorkflowStep{ + ID: "destination-step", WorkflowID: "wf1", Name: "Destination", + Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{{ + Type: wfmodels.OnEnterSetSessionMode, + Config: map[string]interface{}{"mode": "destination"}, + }}}, + } + svc := createTestService(repo, steps, newMockTaskRepo()) + var starts atomic.Int32 + feederDone := make(chan struct{}) + svc.onManualMoveLifecycleStart = func() { starts.Add(1) } + svc.SetFeederPullReconciler(&countingFeederPullReconciler{done: feederDone}) + + var wg sync.WaitGroup + ready := make(chan struct{}) + wg.Add(2) + go func() { + defer wg.Done() + <-ready + svc.recoverTaskLifecycleToken(ctx, "concurrent-manual-move-task") + }() + go func() { + defer wg.Done() + <-ready + svc.handleTaskMoved(ctx, watcher.TaskMovedEventData{ + TaskID: "concurrent-manual-move-task", SessionID: "concurrent-manual-move-session", + FromStepID: "source-step", ToStepID: "destination-step", WIPAdmitted: true, + }) + }() + close(ready) + wg.Wait() + + select { + case <-feederDone: + case <-time.After(2 * time.Second): + t.Fatal("manual move feeder continuation never completed") + } + stored, err := repo.GetTask(ctx, "concurrent-manual-move-task") + if err != nil { + t.Fatalf("reload task: %v", err) + } + if _, pending := stored.Metadata[models.MetaKeyManualMoveLifecyclePending]; pending { + t.Fatal("manual move lifecycle pending token never cleared") + } + + if got := starts.Load(); got != 1 { + t.Fatalf("manual move step exit/enter side effects ran %d times concurrently, want exactly 1", got) + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 24dd1149a1..474c33b8bc 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -189,7 +189,7 @@ type TaskEventPublisher interface { // lifecycle has completed. The task service remains the owner of candidate // selection and promotion rules. type FeederPullReconciler interface { - ReconcileFeederPulls(ctx context.Context, workflowID, feederStepID string) + ReconcileFeederPulls(ctx context.Context, workflowID, feederStepID string) error } type taskQueuePromotionPublisher interface { @@ -880,6 +880,19 @@ type Service struct { // / stopIdleSessionReaper no-op. See idle_session_reaper.go. idleReaper *idleSessionReaper + // lifecycleSweepCancel / lifecycleSweepWorkers own the one-shot + // background goroutine that runs reconcileTaskLifecycleTokens and + // reconcileDependencyLaunchesOnStartup after the watcher and scheduler + // have started. See startLifecycleSweepAsync / stopLifecycleSweepAsync. + // lifecycleSweepMu guards lifecycleSweepCancel and lifecycleSweepStopped: + // Start sets running=true well before it reaches startLifecycleSweepAsync, + // so a concurrent Stop can otherwise read lifecycleSweepCancel while it is + // still nil and return before Start launches an uncancellable sweep. + lifecycleSweepMu sync.Mutex + lifecycleSweepCancel context.CancelFunc + lifecycleSweepStopped bool + lifecycleSweepWorkers sync.WaitGroup + // unreclaimedWorkspaces holds the automation runs whose workspace removal // was attempted and did not actually free the directory. Retention selects // candidates by "still has a live worktree row", and a failed removal can @@ -2593,6 +2606,12 @@ func (s *Service) Start(ctx context.Context) error { s.mu.Unlock() return ErrServiceAlreadyRunning } + // Publish the next-run state before advertising the service as running. A + // concurrent Stop can then never clear lifecycleSweepStopped and have a + // racing Start reset it after Stop has already returned. Keep this after + // the already-running check so a rejected duplicate Start cannot reopen a + // sweep while the service is active. + s.resetLifecycleSweepWorkers() s.running = true s.startedAt = time.Now() s.mu.Unlock() @@ -2634,12 +2653,6 @@ func (s *Service) Start(ctx context.Context) error { if s.workflowStore != nil { s.workflowStore.ReconcileQueuedTasks(ctx) } - s.reconcileTaskLifecycleTokens(ctx) - // Chains whose predecessor completed while the process was down: the - // dependencies_resolved event is in-memory and is not replayed, so without - // this sweep a chain stalls silently across a restart. Runs after the WIP - // reconciler so an admitted-by-promotion task is already eligible. - s.reconcileDependencyLaunchesOnStartup(ctx) // Start the watcher first to begin receiving events if err := s.watcher.Start(ctx); err != nil { @@ -2660,6 +2673,26 @@ func (s *Service) Start(ctx context.Context) error { return err } + // Run the startup lifecycle sweep (reconcileTaskLifecycleTokens, then + // reconcileDependencyLaunchesOnStartup — sequenced so an + // admitted-by-promotion task is already eligible for the dependency sweep + // when reconcileTaskLifecycleTokens returns normally) in the background, + // now that the watcher is subscribed. That sequencing only holds when + // reconcileTaskLifecycleTokens's own worker pool drains before it + // returns: on its deadline/cancel paths it returns with workers still + // in flight, so reconcileDependencyLaunchesOnStartup can run concurrently + // with lifecycle recovery for the tasks that had not finished yet. + // Recovery it drives (recoverManualMoveLifecycle -> ... -> + // waitForSessionReady) polls the DB for a session state that only + // handleAgentBootReady sets, and that handler is wired by + // watcher.Start. Running the sweep before the watcher meant its own + // completion event was dropped on the floor, so every recovered task + // burned its full interactive timeout regardless of whether the + // underlying agent was actually healthy — see + // docs/specs/startup-listener-before-recovery/spec.md. Service.Stop + // joins this goroutine (bounded) via stopLifecycleSweepAsync. + s.startLifecycleSweepAsync(ctx) + // Subscribe to GitHub integration events s.subscribeGitHubEvents() @@ -2729,6 +2762,7 @@ func (s *Service) Stop() error { // have already stopped, and may launch or recover a session during teardown. s.stopDynamicSuccessorWorkers() s.stopDynamicPolicyRecovery() + s.stopLifecycleSweepAsync() // Stop components in reverse order var errs []error @@ -2758,6 +2792,76 @@ func (s *Service) Stop() error { return nil } +// resetLifecycleSweepWorkers clears a stop recorded by a previous Stop, so a +// restarted service can launch the sweep again. Call at the top of Start, +// before any path that can call Stop concurrently. +func (s *Service) resetLifecycleSweepWorkers() { + s.lifecycleSweepMu.Lock() + defer s.lifecycleSweepMu.Unlock() + s.lifecycleSweepStopped = false +} + +// startLifecycleSweepAsync launches the startup lifecycle sweep +// (reconcileTaskLifecycleTokens, then reconcileDependencyLaunchesOnStartup — +// sequenced so an admitted-by-promotion task is already eligible for the +// dependency sweep when reconcileTaskLifecycleTokens's worker pool drains +// before it returns; on its deadline/cancel paths the two can run +// concurrently instead) in a background goroutine derived from ctx. Returns +// false if the sweep was skipped because the service is stopping. +// stopLifecycleSweepAsync cancels this context so the sweep abandons its own +// bounded wait promptly on shutdown instead of running out its full deadline. +// Start sets s.running before reaching this call, so a Stop racing in +// between must not be answered by a nil lifecycleSweepCancel: check the +// stopped flag under the same lock stopLifecycleSweepAsync uses, mirroring +// stopDynamicSuccessorWorkers / resetDynamicSuccessorWorkers. +func (s *Service) startLifecycleSweepAsync(ctx context.Context) bool { + sweepCtx, cancel := context.WithCancel(ctx) + s.lifecycleSweepMu.Lock() + if s.lifecycleSweepStopped { + s.lifecycleSweepMu.Unlock() + cancel() + s.logger.Warn("startup lifecycle sweep skipped; service is stopping") + return false + } + s.lifecycleSweepCancel = cancel + s.lifecycleSweepWorkers.Add(1) + s.lifecycleSweepMu.Unlock() + go func() { + defer s.lifecycleSweepWorkers.Done() + s.reconcileTaskLifecycleTokens(sweepCtx) + s.reconcileDependencyLaunchesOnStartup(sweepCtx) + }() + return true +} + +// stopLifecycleSweepAsync cancels the sweep's context and joins its +// goroutine, bounded so shutdown cannot hang on a sweep still waiting out a +// per-task resume: the resume path runs under context.WithoutCancel +// (task_operations.go), so cancellation here bounds the sweep's own +// dispatch loop but not an already in-flight resume, which is left to +// finish in the background past Stop's return — the same shutdown contract +// as the send-now workers (queue_send_now.go). +func (s *Service) stopLifecycleSweepAsync() { + s.lifecycleSweepMu.Lock() + s.lifecycleSweepStopped = true + cancel := s.lifecycleSweepCancel + s.lifecycleSweepCancel = nil + s.lifecycleSweepMu.Unlock() + if cancel != nil { + cancel() + } + done := make(chan struct{}) + go func() { + s.lifecycleSweepWorkers.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(sendNowClaimRecoveryTimeout): + s.logger.Warn("timed out waiting for startup lifecycle sweep during shutdown") + } +} + // reconcileSessionsOnStartup adjusts database state for sessions that were active before restart. // It does NOT launch any agent processes — sessions are recovered lazily when the user opens them // (via task.session.status → task.session.resume or by sending a prompt). diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index ac74a22916..0143033255 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -119,8 +119,10 @@ const ( // records the source step so stale deliveries cannot run the wrong exit. MetaKeyManualMoveLifecyclePending = "manual_move_lifecycle_pending" // MetaKeyManualMoveLifecycleCompleted records that the admitted manual move - // lifecycle finished. It remains as an idempotency marker until the next - // step-changing move replaces it. + // lifecycle finished. It is cleared once its continuation has run and no + // MetaKeyManualMoveLifecyclePending token remains, so it does not + // accumulate as permanent startup-recovery work; a fresh manual move + // replaces it with a new pending token before it would be cleared. MetaKeyManualMoveLifecycleCompleted = "manual_move_lifecycle_completed" // MetaKeyAppliedDeferredMoves stores deferred move IDs that have already // been applied, preventing a stale queue rollback from replaying one. diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index a9cbae587e..c634c0dacc 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -946,6 +946,68 @@ func (r *Repository) RemoveTaskMetadataKey(ctx context.Context, taskID, key stri return r.removeTaskMetadataKeyWithExecutor(ctx, r.db, taskID, key) } +// ClearManualMoveLifecycleMarkersIfCompleted atomically removes the pending +// and completed markers for a manual move only while the completed marker and +// task update generation still match the recovery snapshot. A new move clears +// the old completed marker in the same task write that creates its pending +// marker, and a later completion reuses the boolean marker value, so the +// generation predicate prevents either newer move from being erased. +func (r *Repository) ClearManualMoveLifecycleMarkersIfCompleted(ctx context.Context, taskID string, completedAt time.Time) (bool, error) { + var query string + var args []interface{} + if dialect.IsPostgres(r.db.DriverName()) { + query = ` + UPDATE tasks + SET metadata = ( + CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END + #- ARRAY[?]::text[] #- ARRAY[?]::text[] + )::text, updated_at = ? + WHERE id = ? + AND jsonb_extract_path( + CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, + ? + ) IS NOT NULL + AND updated_at = ? + ` + args = []interface{}{ + models.MetaKeyManualMoveLifecyclePending, + models.MetaKeyManualMoveLifecycleCompleted, + r.nowUTC(), + taskID, + models.MetaKeyManualMoveLifecycleCompleted, + completedAt, + } + } else { + query = ` + UPDATE tasks + SET metadata = json_remove( + CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, + ?, ? + ), updated_at = ? + WHERE id = ? + AND json_type( + CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, + ? + ) IS NOT NULL + AND updated_at = ? + ` + args = []interface{}{ + jsonPath(models.MetaKeyManualMoveLifecyclePending), + jsonPath(models.MetaKeyManualMoveLifecycleCompleted), + r.nowUTC(), + taskID, + jsonPath(models.MetaKeyManualMoveLifecycleCompleted), + completedAt, + } + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(query), args...) + if err != nil { + return false, err + } + rows, err := result.RowsAffected() + return rows > 0, err +} + // RemoveTaskMetadataKeyIfStamp removes one metadata object only when its // nested stamp still equals expectedStamp. The comparison and key removal are // one statement so a successful launch cannot erase a newer failure. diff --git a/apps/backend/internal/task/repository/sqlite/task_metadata_cas_test.go b/apps/backend/internal/task/repository/sqlite/task_metadata_cas_test.go index bf4c09f9c0..7790e86e63 100644 --- a/apps/backend/internal/task/repository/sqlite/task_metadata_cas_test.go +++ b/apps/backend/internal/task/repository/sqlite/task_metadata_cas_test.go @@ -4,6 +4,7 @@ import ( "context" "path/filepath" "testing" + "time" "github.com/jmoiron/sqlx" @@ -143,6 +144,47 @@ func runMetadataNoActiveSessionContract(t *testing.T, repo *Repository) { } } +func runManualMoveLifecycleMarkerContract(t *testing.T, repo *Repository) { + t.Helper() + ctx := context.Background() + task, err := repo.GetTask(ctx, casTaskID) + if err != nil { + t.Fatalf("load task generation: %v", err) + } + + cleared, err := repo.ClearManualMoveLifecycleMarkersIfCompleted(ctx, casTaskID, task.UpdatedAt) + if err != nil { + t.Fatalf("clear manual move markers: %v", err) + } + if !cleared { + t.Fatal("completed manual move markers were not cleared") + } + if _, present := metadataValue(t, repo, models.MetaKeyManualMoveLifecyclePending); present { + t.Fatal("manual move pending marker remained after atomic clear") + } + if _, present := metadataValue(t, repo, models.MetaKeyManualMoveLifecycleCompleted); present { + t.Fatal("manual move completed marker remained after atomic clear") + } + if _, preserved := metadataValue(t, repo, "other_key"); !preserved { + t.Fatal("atomic clear removed unrelated task metadata") + } + + if err := repo.SetTaskMetadataKey(ctx, casTaskID, models.MetaKeyManualMoveLifecyclePending, + map[string]interface{}{"from_step_id": "new-source"}); err != nil { + t.Fatalf("seed pending-only marker: %v", err) + } + cleared, err = repo.ClearManualMoveLifecycleMarkersIfCompleted(ctx, casTaskID, time.Now().UTC()) + if err != nil { + t.Fatalf("clear pending-only markers: %v", err) + } + if cleared { + t.Fatal("pending-only marker was cleared without a completed marker") + } + if _, present := metadataValue(t, repo, models.MetaKeyManualMoveLifecyclePending); !present { + t.Fatal("pending-only marker disappeared") + } +} + func newRepoForMetadataCASTests(t *testing.T) *Repository { t.Helper() dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "metadata-cas-test.db")) @@ -161,11 +203,14 @@ func newRepoForMetadataCASTests(t *testing.T) *Repository { func TestSetTaskMetadataKeyIfPresentSQLite(t *testing.T) { repo := newRepoForMetadataCASTests(t) seedMetadataCASTask(t, repo, map[string]interface{}{ - "deferred_launch": map[string]interface{}{"prompt": "stale"}, - "other_key": "keep me", + "deferred_launch": map[string]interface{}{"prompt": "stale"}, + models.MetaKeyManualMoveLifecyclePending: map[string]interface{}{"from_step_id": "source"}, + models.MetaKeyManualMoveLifecycleCompleted: true, + "other_key": "keep me", }) runMetadataCASContract(t, repo) runMetadataNoActiveSessionContract(t, repo) + runManualMoveLifecycleMarkerContract(t, repo) } // The JSON patch and its presence predicate are written per dialect, so SQLite @@ -175,11 +220,14 @@ func TestPostgresSetTaskMetadataKeyIfPresent(t *testing.T) { db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) repo := newPostgresMetadataCASRepo(t, db) seedMetadataCASTask(t, repo, map[string]interface{}{ - "deferred_launch": map[string]interface{}{"prompt": "stale"}, - "other_key": "keep me", + "deferred_launch": map[string]interface{}{"prompt": "stale"}, + models.MetaKeyManualMoveLifecyclePending: map[string]interface{}{"from_step_id": "source"}, + models.MetaKeyManualMoveLifecycleCompleted: true, + "other_key": "keep me", }) runMetadataCASContract(t, repo) runMetadataNoActiveSessionContract(t, repo) + runManualMoveLifecycleMarkerContract(t, repo) } func newPostgresMetadataCASRepo(t *testing.T, db *sqlx.DB) *Repository { diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index c6b39e5d71..06a6385c2c 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -615,29 +615,30 @@ func prepareAutoTitle(req *CreateTaskRequest) error { return nil } -func (s *Service) pullTasksFromNewFeederWork(ctx context.Context, workflowID, feederStepID string) { +func (s *Service) pullTasksFromNewFeederWork(ctx context.Context, workflowID, feederStepID string) error { stepLister, ok := s.workflowStepGetter.(workflowStepLister) if !ok || workflowID == "" || feederStepID == "" { - return + return nil } steps, err := stepLister.ListStepsByWorkflow(ctx, workflowID) if err != nil { s.logger.Warn("failed to list workflow steps after feeder task creation", zap.String("workflow_id", workflowID), zap.Error(err)) - return + return err } for _, step := range steps { if step != nil && step.PullFromStepID == feederStepID { s.pullNextTaskOnVacate(ctx, step.ID, "") } } + return nil } // ReconcileFeederPulls wakes steps that pull from feederStepID. The // orchestrator calls this after an admitted manual move's lifecycle barrier // completes so selection and promotion rules stay owned by this service. -func (s *Service) ReconcileFeederPulls(ctx context.Context, workflowID, feederStepID string) { - s.pullTasksFromNewFeederWork(ctx, workflowID, feederStepID) +func (s *Service) ReconcileFeederPulls(ctx context.Context, workflowID, feederStepID string) error { + return s.pullTasksFromNewFeederWork(ctx, workflowID, feederStepID) } func (s *Service) createTaskWithCapacity(ctx context.Context, task *models.Task) error { diff --git a/apps/web/e2e/tests/kanban/wip-overflow-queue.spec.ts b/apps/web/e2e/tests/kanban/wip-overflow-queue.spec.ts index a609fc5343..79376c8031 100644 --- a/apps/web/e2e/tests/kanban/wip-overflow-queue.spec.ts +++ b/apps/web/e2e/tests/kanban/wip-overflow-queue.spec.ts @@ -42,8 +42,8 @@ test("dragging into a feeder wakes an open pull target without reload", async ({ const sourceY = sourceBox!.y + sourceBox!.height / 2; await testPage.mouse.move(sourceX, sourceY); await testPage.mouse.down(); - await testPage.mouse.move(sourceX + 20, sourceY); - await expect(testPage.getByTestId("desktop-kanban-drag-end-reserve")).toHaveCount(1); + await testPage.mouse.move(sourceX + 20, sourceY, { steps: 4 }); + await expect(sourceCard).toHaveClass(/opacity-50/, { timeout: 5_000 }); // Dragging can temporarily move destinations before the anchored source // column, so scroll them back into the viewport before dropping. diff --git a/docs/specs/startup-listener-before-recovery/spec.md b/docs/specs/startup-listener-before-recovery/spec.md index ec2a639cc0..9d9e9a40ee 100644 --- a/docs/specs/startup-listener-before-recovery/spec.md +++ b/docs/specs/startup-listener-before-recovery/spec.md @@ -1,7 +1,7 @@ --- status: draft created: 2026-08-22 -revised: 2026-08-22 +revised: 2026-09-05 type: repair --- @@ -60,6 +60,39 @@ Three consequences, in severity order: This is startup ordering, not a defect in the recovery logic. The same frames are reached whenever a lifecycle token is pending. +## Correction (2026-09-05) + +This spec's original Desired-behaviour and Out-of-scope sections asserted +that the sweep's position *before* `watcher.Start` was load-bearing and that +moving it would introduce races with live traffic. Runtime evidence gathered +after this spec's socket-bind fix shipped disproves that: with the socket-bind +fix in place, boots with pending lifecycle tokens still never reached `API +configured` — the sweep itself was still holding the readiness path open +indefinitely, just behind the new bootstrap 503 handler instead of behind a +closed socket. + +The reason is a **self-inflicted event blackout**, not merely a slow +individual resume. `Watcher.Start` is what registers +`{events.AgentBootReady, w.handlers.OnAgentBootReady}` (the *only* handler +that flips a launching session to `WAITING_FOR_INPUT` on the resume path). +The event bus used here has no replay, so any `agent.boot_ready` published +while the sweep runs *before* `watcher.Start` is dropped on the floor. +`waitForSessionReady`'s poll can then never observe the state change it is +waiting for, and burns its full `constants.AgentLaunchTimeout` (15 minutes) +on every recovered task, regardless of whether the underlying agent process +is actually healthy. Four workers wide, retried up to three times, this is +indistinguishable from a hang and gates the entire router. + +The fix inverts the ordering this spec forbade: the sweep now runs in a +background goroutine *after* `watcher.Start` and `scheduler.Start` succeed, +so the events its own resume path depends on are deliverable from the +moment recovery starts. Readiness is not gated on the sweep at all — the +socket-bind / bootstrap-503 behavior this spec introduced is unaffected and +still the mechanism that makes an in-progress sweep observable as "starting" +rather than a hang. The Desired-behaviour and Out-of-scope sections below +are amended accordingly; the regression scenarios still hold, since a bound +socket does not depend on this ordering. + ## Broken behavior - No socket is bound, and `/health` cannot be answered, until orchestrator @@ -71,9 +104,19 @@ are reached whenever a lifecycle token is pending. - The backend binds its socket and answers a liveness probe within the release health window, regardless of how much recovery work is pending. -- Startup recovery keeps its current position and ordering: it still runs - before the watcher and scheduler start, so it is not exposed to concurrent - watcher events, scheduler activity, or API mutations. +- Startup recovery runs in the background *after* the watcher and scheduler + start (amended 2026-09-05 — see Correction above): its own resume path + depends on events only the watcher's subscriptions deliver, so running it + earlier does not protect anything and instead guarantees every recovered + task times out. Running concurrently with live watcher/scheduler traffic is + safe because the sweep's operations are already built for concurrent + delivery — per-task locks (`queuedMoveLifecycleLocks`) and one-shot tokens + claimed by an atomic metadata-key removal (`claimTaskEventMetadata`) — + confirmed by + `TestLifecycleSweepConcurrentWithLiveTaskMovedRunsManualMoveSideEffectsOnce`, + which runs the sweep's recovery path concurrently with a live `task.moved` + redelivery of the same admitted manual move and asserts the step exit/enter + side effects run exactly once. - Requests that need a started application are answered with an explicit "starting" response rather than by a hung connection or a refused socket. - Readiness is observable and distinct from liveness: an operator can tell @@ -94,21 +137,29 @@ are reached whenever a lifecycle token is pending. ## Out of scope -- Reordering, parallelising, or moving startup reconciliation relative to the - watcher and scheduler. Its current position is load-bearing and moving it - introduces races with live traffic. - Changing WIP promotion, feeder pull, or manual-move lifecycle semantics. - Changing `AgentLaunchTimeout`, `agentPromptReadyTimeout`, or the cold-resume - retry shape. See the follow-up note below. + retry shape (`context.WithoutCancel`). See the follow-up note below — a + caller-side deadline on the sweep bounds admission and reporting, not an + in-flight resume, precisely because of this contract. - Raising `healthTimeoutReleaseMS`, which hides the coupling instead of fixing it. Empirically a 10-minute window did not help. -## Known follow-up, deliberately separate +## Known follow-up, addressed 2026-09-05 -Fixing the ordering stops the crash loop but does not make a large board start -quickly: a backend with many pending tokens still spends a long time in -"starting" because bulk recovery inherits the interactive 15-minute -`AgentLaunchTimeout` per task. A bulk startup sweep needs its own, much -smaller per-task budget, and a resume path that can actually be abandoned. That -requires changing the `context.WithoutCancel` resume contract and is a separate -piece of work with its own spec. +The note below originally described work deliberately deferred past this +spec's socket-bind fix. It has since been addressed in the same change that +corrected the ordering above (see Correction): the sweep now runs off the +readiness path entirely (background goroutine, not gating `API configured`), +carries its own bounded overall deadline separate from the interactive +`AgentLaunchTimeout`, and no longer accumulates already-completed +`manual_move_lifecycle_completed` tokens as permanent per-boot work. The +`context.WithoutCancel` resume contract itself is unchanged, so an +individual in-flight resume still cannot be aborted early — the sweep's +deadline bounds its own dispatch loop returning, not that resume. + +> Fixing the ordering stops the crash loop but does not make a large board +> start quickly: a backend with many pending tokens still spends a long time +> in "starting" because bulk recovery inherits the interactive 15-minute +> `AgentLaunchTimeout` per task. A bulk startup sweep needs its own, much +> smaller per-task budget, and a resume path that can actually be abandoned.