fix(backend): stop restarts from leaving the API stuck in "starting" (#3445)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

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
 	}
 

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)

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 {

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

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")
+	}
+}

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"])
+	}
+}

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)
+	}
+}

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).

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.

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.

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 {

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 {

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.

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.