feat(orchestrator): periodic idle-session reaper on Service (#2836)

Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/agent/runtime/lifecycle/manager_launch.go

@@ -1133,7 +1133,7 @@ func (m *Manager) markAgentStartPending(execution *AgentExecution) {
 // dedicated singleflight key so they don't race on the shared AgentExecution
 // pointer.
 func (m *Manager) promoteWorkspaceExecution(ctx context.Context, execution *AgentExecution, req *LaunchRequest) error {
-	_, err := m.doCoalescedExecution(ctx, "promote:"+req.SessionID, func(sharedCtx context.Context) (interface{}, error) {
+	_, err := m.doCoalescedExecution(ctx, req.SessionID, func(sharedCtx context.Context) (interface{}, error) {
 		activityLease, acquireErr := m.acquireActivity(sharedCtx, activity.KindExecutionPreparing)
 		if acquireErr != nil {
 			return nil, acquireErr

apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go

@@ -270,6 +270,44 @@ func (m *Manager) CleanupStaleExecutionBySessionID(ctx context.Context, sessionI
 	if !exists {
 		return nil // No execution to clean up
 	}
+	return m.cleanupStaleExecution(ctx, execution)
+}
+
+// CleanupStaleExecutionBySessionIDIfCurrent cleans up only the execution that
+// the caller observed. The operation shares the session-keyed singleflight
+// used by launch paths, so a successor launch either wins before cleanup or
+// runs after the stale execution has been removed. The persisted timestamp
+// check closes the remaining same-ID workspace-promotion window.
+func (m *Manager) CleanupStaleExecutionBySessionIDIfCurrent(
+	ctx context.Context,
+	sessionID, expectedExecutionID string,
+	expectedUpdatedAt time.Time,
+) error {
+	if sessionID == "" || expectedExecutionID == "" {
+		return nil
+	}
+	_, err := m.doCoalescedExecution(ctx, sessionID, func(sharedCtx context.Context) (interface{}, error) {
+		execution, exists := m.executionStore.GetBySessionID(sessionID)
+		if !exists || execution == nil || execution.ID != expectedExecutionID {
+			return nil, nil
+		}
+		if reader, ok := m.runningWriter.(executorRunningReader); ok && !expectedUpdatedAt.IsZero() {
+			current, readErr := reader.GetExecutorRunningBySessionID(sharedCtx, sessionID)
+			if readErr != nil || current == nil || current.AgentExecutionID != expectedExecutionID ||
+				!current.UpdatedAt.Equal(expectedUpdatedAt) {
+				return nil, nil
+			}
+		}
+		return nil, m.cleanupStaleExecution(sharedCtx, execution)
+	})
+	return err
+}
+
+func (m *Manager) cleanupStaleExecution(ctx context.Context, execution *AgentExecution) error {
+	if execution == nil {
+		return nil
+	}
+	sessionID := execution.SessionID
 
 	m.logger.Info("cleaning up stale agent execution",
 		zap.String("session_id", sessionID),
@@ -298,7 +336,7 @@ func (m *Manager) CleanupStaleExecutionBySessionID(ctx context.Context, sessionI
 	// Delete the persistence row in lockstep with store removal so we never
 	// leave a phantom executors_running row pointing at a non-existent
 	// execution. Best-effort; "not found" is expected and silently swallowed.
-	m.deleteExecutorRunning(ctx, sessionID)
+	m.deleteExecutorRunning(ctx, sessionID, execution.ID)
 
 	return nil
 }

apps/backend/internal/agent/runtime/lifecycle/manager_startup.go

@@ -84,7 +84,21 @@ func (m *Manager) routePassthrough(ctx context.Context, execution *AgentExecutio
 // StartAgentProcess configures and starts the agent subprocess for an execution.
 // This must be called after Launch() to actually start the agent (e.g., auggie, codex).
 // The command is built internally based on the execution's agent profile.
-func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) (retErr error) {
+func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) error {
+	execution, exists := m.executionStore.Get(executionID)
+	if !exists {
+		return fmt.Errorf("execution %q not found", executionID)
+	}
+	if execution.SessionID == "" {
+		return m.startAgentProcess(ctx, executionID)
+	}
+	_, err := m.doCoalescedExecution(ctx, execution.SessionID, func(sharedCtx context.Context) (interface{}, error) {
+		return nil, m.startAgentProcess(sharedCtx, executionID)
+	})
+	return err
+}
+
+func (m *Manager) startAgentProcess(ctx context.Context, executionID string) (retErr error) {
 	execution, exists := m.executionStore.Get(executionID)
 	if !exists {
 		return fmt.Errorf("execution %q not found", executionID)

apps/backend/internal/agent/runtime/lifecycle/persistence.go

@@ -330,57 +330,114 @@ func (m *Manager) persistExecutorRunningResult(ctx context.Context, execution *A
 // orchestrator applies the full task-state invariant during reconciliation.
 //
 // Best-effort: a failure here is logged but doesn't propagate.
-func (m *Manager) deleteExecutorRunning(ctx context.Context, sessionID string) {
+func (m *Manager) deleteExecutorRunning(ctx context.Context, sessionID string, expectedExecutionIDs ...string) {
 	if m.runningWriter == nil {
 		return
 	}
-
-	// Inspect the row first so we never delete one we couldn't read (fail-safe:
-	// an unreadable row might hold a resume_token).
-	if reader, ok := m.runningWriter.(executorRunningReader); ok {
-		existing, err := reader.GetExecutorRunningBySessionID(ctx, sessionID)
-		switch {
-		case err == nil && existing != nil &&
-			(existing.ResumeToken != "" || existing.Status == models.ExecutorRunningStatusRunning):
-			if repairErr := m.runningWriter.RepairExecutorRunningDead(ctx, sessionID); repairErr != nil &&
-				!errors.Is(repairErr, models.ErrExecutorRunningNotFound) {
-				m.logger.Warn("failed to repair resumable executors_running row on cleanup; leaving row intact",
-					zap.String("session_id", sessionID),
-					zap.Error(repairErr))
-			} else {
-				m.logger.Info("repaired resumable executors_running row instead of deleting (resume-safety invariant)",
-					zap.String("session_id", sessionID))
-			}
-			return
-		case errors.Is(err, models.ErrExecutorRunningNotFound):
-			m.logger.Debug("delete executors_running on cleanup: row not found",
-				zap.String("session_id", sessionID))
-			return
-		case err != nil:
-			m.logger.Warn("skipping executors_running delete: prior-row read failed, refusing to risk deleting a resumable row",
-				zap.String("session_id", sessionID),
-				zap.Error(err))
-			return
-		}
-	} else {
-		m.logger.Warn("delete executors_running on cleanup: writer does not support reading; resume-safety check skipped",
-			zap.String("session_id", sessionID))
+	expectedExecutionID := ""
+	if len(expectedExecutionIDs) > 0 {
+		expectedExecutionID = expectedExecutionIDs[0]
 	}
 
-	if err := m.runningWriter.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil {
+	if m.skipExecutorRunningCleanup(ctx, sessionID, expectedExecutionID) {
+		return
+	}
+
+	deleteErr := m.deleteExecutorRunningRow(ctx, sessionID, expectedExecutionID)
+	if deleteErr != nil {
 		// "not found" is expected for sessions that were never launched; everything
 		// else is a real I/O failure (write timeout, locked DB) and should surface.
-		if errors.Is(err, models.ErrExecutorRunningNotFound) {
+		if errors.Is(deleteErr, models.ErrExecutorRunningNotFound) {
 			m.logger.Debug("delete executors_running on cleanup: row not found",
 				zap.String("session_id", sessionID))
 		} else {
 			m.logger.Warn("delete executors_running on cleanup",
 				zap.String("session_id", sessionID),
-				zap.Error(err))
+				zap.Error(deleteErr))
 		}
 	}
 }
 
+// skipExecutorRunningCleanup inspects the row before deletion. It returns true
+// when cleanup must stop because the row is rotated, resumable, missing, or
+// unreadable. A writer without a reader keeps the historical best-effort path.
+func (m *Manager) skipExecutorRunningCleanup(ctx context.Context, sessionID, expectedExecutionID string) bool {
+	reader, ok := m.runningWriter.(executorRunningReader)
+	if !ok {
+		m.logger.Warn("delete executors_running on cleanup: writer does not support reading; resume-safety check skipped",
+			zap.String("session_id", sessionID))
+		return false
+	}
+	existing, err := reader.GetExecutorRunningBySessionID(ctx, sessionID)
+	if expectedExecutionID != "" && existing != nil && existing.AgentExecutionID != expectedExecutionID {
+		m.logger.Info("skipping executors_running cleanup for rotated execution",
+			zap.String("session_id", sessionID),
+			zap.String("expected_execution_id", expectedExecutionID),
+			zap.String("current_execution_id", existing.AgentExecutionID))
+		return true
+	}
+	if err == nil && existing != nil &&
+		(existing.ResumeToken != "" || existing.Status == models.ExecutorRunningStatusRunning) {
+		return m.repairExecutorRunningAfterCleanup(ctx, sessionID, expectedExecutionID, existing.UpdatedAt)
+	}
+	if errors.Is(err, models.ErrExecutorRunningNotFound) {
+		m.logger.Debug("delete executors_running on cleanup: row not found",
+			zap.String("session_id", sessionID))
+		return true
+	}
+	if err != nil {
+		m.logger.Warn("skipping executors_running delete: prior-row read failed, refusing to risk deleting a resumable row",
+			zap.String("session_id", sessionID),
+			zap.Error(err))
+		return true
+	}
+	return false
+}
+
+func (m *Manager) repairExecutorRunningAfterCleanup(
+	ctx context.Context,
+	sessionID, expectedExecutionID string,
+	expectedUpdatedAt time.Time,
+) bool {
+	var repairErr error
+	if casWriter, ok := m.runningWriter.(executorRunningCASWriter); ok && expectedExecutionID != "" {
+		repairErr = casWriter.RepairExecutorRunningDeadIfCurrent(
+			ctx, sessionID, expectedExecutionID, expectedUpdatedAt,
+		)
+	} else {
+		repairErr = m.runningWriter.RepairExecutorRunningDead(ctx, sessionID)
+	}
+	if repairErr == nil || errors.Is(repairErr, models.ErrExecutorRunningNotFound) {
+		m.logger.Info("repaired resumable executors_running row instead of deleting (resume-safety invariant)",
+			zap.String("session_id", sessionID))
+		return true
+	}
+	if errors.Is(repairErr, models.ErrExecutionRotated) {
+		m.logger.Info("skipping executors_running repair for rotated execution",
+			zap.String("session_id", sessionID),
+			zap.String("expected_execution_id", expectedExecutionID))
+		return true
+	}
+	m.logger.Warn("failed to repair resumable executors_running row on cleanup; leaving row intact",
+		zap.String("session_id", sessionID),
+		zap.Error(repairErr))
+	return true
+}
+
+func (m *Manager) deleteExecutorRunningRow(ctx context.Context, sessionID, expectedExecutionID string) error {
+	casWriter, hasCAS := m.runningWriter.(executorRunningCASWriter)
+	if !hasCAS || expectedExecutionID == "" {
+		return m.runningWriter.DeleteExecutorRunningBySessionID(ctx, sessionID)
+	}
+	var expectedUpdatedAt time.Time
+	if reader, ok := m.runningWriter.(executorRunningReader); ok {
+		if current, readErr := reader.GetExecutorRunningBySessionID(ctx, sessionID); readErr == nil && current != nil {
+			expectedUpdatedAt = current.UpdatedAt
+		}
+	}
+	return casWriter.DeleteExecutorRunningIfCurrent(ctx, sessionID, expectedExecutionID, expectedUpdatedAt)
+}
+
 // executorRunningReader is the optional read-side of the writer used to fetch
 // a prior row before re-upserting. Implementing it lets the writer carry forward
 // orchestrator-owned columns; not implementing it just means we always write
@@ -388,3 +445,16 @@ func (m *Manager) deleteExecutorRunning(ctx context.Context, sessionID string) {
 type executorRunningReader interface {
 	GetExecutorRunningBySessionID(ctx context.Context, sessionID string) (*models.ExecutorRunning, error)
 }
+
+type executorRunningCASWriter interface {
+	RepairExecutorRunningDeadIfCurrent(
+		ctx context.Context,
+		sessionID, expectedExecutionID string,
+		expectedUpdatedAt time.Time,
+	) error
+	DeleteExecutorRunningIfCurrent(
+		ctx context.Context,
+		sessionID, expectedExecutionID string,
+		expectedUpdatedAt time.Time,
+	) error
+}

apps/backend/internal/backendapp/adapters.go

@@ -614,6 +614,16 @@ func (a *lifecycleAdapter) CleanupStaleExecutionBySessionID(ctx context.Context,
 	return a.mgr.CleanupStaleExecutionBySessionID(ctx, sessionID)
 }
 
+func (a *lifecycleAdapter) CleanupStaleExecutionBySessionIDIfCurrent(
+	ctx context.Context,
+	sessionID, expectedExecutionID string,
+	expectedUpdatedAt time.Time,
+) error {
+	return a.mgr.CleanupStaleExecutionBySessionIDIfCurrent(
+		ctx, sessionID, expectedExecutionID, expectedUpdatedAt,
+	)
+}
+
 func (a *lifecycleAdapter) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID, sessionID string) error {
 	_, err := a.mgr.EnsureWorkspaceExecutionForSession(ctx, taskID, sessionID)
 	return err

apps/backend/internal/orchestrator/idle_session_reaper.go

@@ -0,0 +1,250 @@
+package orchestrator
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+// Idle-session reaper.
+//
+// The reaper is a single owner of one background goroutine on Service. It
+// periodically scans executors_running rows for sessions that have been
+// idle (no live agent process, no active turn, session state in
+// WaitingForInput/Idle/Completed) past a minimum-idle threshold, then
+// hands each off to the same fail-closed reclaimIdleSession primitive
+// that the synchronous settle points (subtask terminal collapse,
+// agent.completed post-settle) use. Rows whose owner process is
+// still alive are never touched: IsAgentRunningForSession returns
+// true and reclaimIdleSession aborts.
+//
+// Two constants:
+//   - idleReaperInterval: how often the reaper scans. Short enough to
+//     reclaim idle rows within a minute, long enough to avoid per-tick
+//     amplification when no rows are candidates.
+//   - idleReaperMinIdle: the minimum row age (UpdatedAt → now) before a
+//     row is eligible. New rows are never reaped even if their state
+//     already looks idle — this gives the system a settling window
+//     after a normal turn ends so a row that is "WaitingForInput
+//     since 1ms" does not get reclaimed on the very next tick.
+//
+// Both constants are intentional configuration knobs in code source —
+// not env vars — because the orchestrator has no other runtime config
+// of this shape, and the values are bounded by fail-closed invariants
+// rather than deployment shape. Operators who need to tune them can
+// edit and re-deploy; consumers do not.
+//
+// Hard invariants:
+//   - Never call StopAgent / StopAgentWithReason / Kill. reclaimIdleSession
+//     routes through repairDeadRowLiveness (status=stopped, LocalPID=0,
+//     resume_token preserved); a live executor is short-circuited inside
+//     reclaimIdleSession before any side effect.
+//   - reclaimIdleSession is fail-closed: an uncertain guard is a skip,
+//     never a force. The reaper just calls the primitive in a loop,
+//     so the same invariant carries.
+//   - Rows whose UpdatedAt is unset / zero / in the future are treated
+//     as ineligible; they would otherwise be reaped on the first tick.
+//   - The reaper scans regardless of session lifecycle: it does NOT
+//     depend on session.ensure / session.launch / agent.ready. The
+//     startup reconciliation covers fresh restarts; the reaper covers
+//     ongoing drift.
+const (
+	// idleReaperInterval is the cadence at which the reaper scans
+	// executors_running for reclaim candidates. Short enough to keep
+	// idle windows tight, long enough that no tick overlaps with the
+	// previous tick on a busy system.
+	idleReaperInterval = 30 * time.Second
+
+	// idleReaperMinIdle is the minimum age (now - row.UpdatedAt)
+	// before a row is eligible. Below this, the system is still
+	// settling — on_enter / on_turn_complete / agent.ready may have
+	// just transitioned the row and the next event has not fired yet.
+	idleReaperMinIdle = 60 * time.Second
+)
+
+// idleSessionReaper owns the lifecycle of one background goroutine.
+// All methods are nil-safe: a Service with reaper == nil treats the
+// reaper as off (useful for tests that don't want a loop).
+type idleSessionReaper struct {
+	mu       sync.Mutex
+	cancel   context.CancelFunc
+	workers  sync.WaitGroup
+	started  bool
+	interval time.Duration
+	minIdle  time.Duration
+}
+
+func newIdleSessionReaper() *idleSessionReaper {
+	return &idleSessionReaper{
+		interval: idleReaperInterval,
+		minIdle:  idleReaperMinIdle,
+	}
+}
+
+// start launches the reaper loop with the given tick callback.
+// Idempotent: a second call on a running reaper is a no-op. Tests
+// inject a custom tick to observe invocation count; production
+// wires Service.reclaimIdleSessionsOnce as the tick.
+func (r *idleSessionReaper) start(parent context.Context, tick func(ctx context.Context)) bool {
+	if r == nil {
+		return false
+	}
+	if tick == nil {
+		return false
+	}
+	if parent == nil {
+		parent = context.Background()
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.started {
+		return false
+	}
+	loopCtx, cancel := context.WithCancel(parent)
+	r.cancel = cancel
+	r.started = true
+	r.workers.Add(1)
+	go func() {
+		defer r.workers.Done()
+		r.runLoop(loopCtx, tick)
+	}()
+	return true
+}
+
+// stop signals the reaper to exit and waits for it. Safe to call on
+// a never-started / already-stopped reaper. Blocks until the loop
+// observes cancellation; tests and Service.Stop rely on this for
+// clean teardown.
+func (r *idleSessionReaper) stop() {
+	if r == nil {
+		return
+	}
+	r.mu.Lock()
+	if r.cancel != nil {
+		r.cancel()
+	}
+	r.mu.Unlock()
+	r.workers.Wait()
+	r.mu.Lock()
+	r.cancel = nil
+	r.started = false
+	r.mu.Unlock()
+}
+
+// runLoop selects on the ticker + the loop context. Tick callback runs
+// under the loop context, so a slow tick is cancelled on shutdown.
+func (r *idleSessionReaper) runLoop(ctx context.Context, tick func(ctx context.Context)) {
+	ticker := time.NewTicker(r.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			tick(ctx)
+		}
+	}
+}
+
+// Service-level hooks. The reaper is owned by Service so the lifecycle
+// matches the rest of the orchestration: Service.Start launches it,
+// Service.Stop joins it.
+
+// startIdleSessionReaper wires the reaper into Service.Start. The
+// reaper captures the canonical fail-closed reclaim primitive so
+// the per-tick work is a thin scan + loop, not a new policy.
+func (s *Service) startIdleSessionReaper(ctx context.Context) {
+	if s.idleReaper == nil {
+		return
+	}
+	if !s.idleReaper.start(ctx, func(tickCtx context.Context) {
+		s.reclaimIdleSessionsOnce(tickCtx)
+	}) {
+		return
+	}
+	s.logger.Info("idle-session reaper started",
+		zap.Duration("interval", s.idleReaper.interval),
+		zap.Duration("min_idle", s.idleReaper.minIdle))
+}
+
+// stopIdleSessionReaper joins the reaper. Must be called before
+// Service.Stop returns to honour the goroutine-ownership invariant.
+func (s *Service) stopIdleSessionReaper() {
+	if s.idleReaper == nil {
+		return
+	}
+	s.idleReaper.stop()
+	s.logger.Info("idle-session reaper stopped")
+}
+
+// reclaimIdleSessionsOnce is the per-tick scan. It reads every
+// executors_running row, applies the minimum-idle filter (UpdatedAt
+// age), and delegates each candidate to reclaimIdleSession. The
+// primitive enforces the fail-closed guard (state, live agent,
+// active turn); this scan only adds the age filter that the
+// settle-point callers do not need.
+//
+// Each row's reclaim is best-effort: a failure is logged at warn and
+// skipped. The next tick will retry. The scan never aborts the loop
+// on a single-row error.
+func (s *Service) reclaimIdleSessionsOnce(ctx context.Context) {
+	now := time.Now().UTC()
+	cutoff := now.Add(-s.idleReaper.minIdle)
+	var (
+		rows []*models.ExecutorRunning
+		err  error
+	)
+	if candidates, ok := s.repo.(idleExecutorCandidateLister); ok {
+		rows, err = candidates.ListExecutorsRunningIdle(ctx, cutoff)
+	} else {
+		rows, err = s.repo.ListExecutorsRunning(ctx)
+	}
+	if err != nil {
+		s.logger.Warn("idle reaper: list executors_running failed; tick skipped",
+			zap.Error(err))
+		return
+	}
+	if len(rows) == 0 {
+		return
+	}
+	for _, row := range rows {
+		if row == nil {
+			continue
+		}
+		if row.Status == models.ExecutorRunningStatusStopped {
+			continue
+		}
+		sessionID := row.SessionID
+		if sessionID == "" {
+			continue
+		}
+		// Reject rows with zero / unset / future UpdatedAt — they would
+		// otherwise look older than the threshold on the first tick.
+		if row.UpdatedAt.IsZero() {
+			continue
+		}
+		if row.UpdatedAt.After(now) || now.Sub(row.UpdatedAt) < s.idleReaper.minIdle {
+			continue
+		}
+		// Delegate to the fail-closed primitive. It reads the session,
+		// checks the triple guard (state × live runtime × active turn),
+		// and either reclaims the row or skips it. Errors here are
+		// recoverable on the next tick — log and move on.
+		if err := s.reclaimIdleSession(ctx, sessionID); err != nil {
+			s.logger.Warn("idle reaper: reclaim failed; row preserved for next tick",
+				zap.String("session_id", sessionID),
+				zap.Error(err))
+		}
+	}
+}
+
+// idleExecutorCandidateLister is implemented by the durable repository. The
+// fallback to ListExecutorsRunning keeps lightweight test doubles compatible,
+// while production can filter stopped and young rows in SQL.
+type idleExecutorCandidateLister interface {
+	ListExecutorsRunningIdle(ctx context.Context, cutoff time.Time) ([]*models.ExecutorRunning, error)
+}

apps/backend/internal/orchestrator/idle_session_reaper_test.go

@@ -0,0 +1,322 @@
+package orchestrator
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/agentruntime"
+	"github.com/kandev/kandev/internal/orchestrator/executor"
+	"github.com/kandev/kandev/internal/task/models"
+	taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+func seedIdleReaperRow(
+	t *testing.T,
+	repo interface {
+		UpsertExecutorRunning(context.Context, *models.ExecutorRunning) error
+	},
+	taskID, sessionID string,
+	state models.TaskSessionState,
+	status string,
+) {
+	t.Helper()
+	if sqliteRepo, ok := repo.(interface {
+		CreateTask(context.Context, *models.Task) error
+		CreateTaskSession(context.Context, *models.TaskSession) error
+	}); ok {
+		now := time.Now().UTC()
+		if err := sqliteRepo.CreateTask(context.Background(), &models.Task{
+			ID: taskID, Title: taskID, State: v1.TaskStateInProgress, CreatedAt: now, UpdatedAt: now,
+		}); err != nil {
+			t.Fatalf("create task: %v", err)
+		}
+		if err := sqliteRepo.CreateTaskSession(context.Background(), &models.TaskSession{
+			ID: sessionID, TaskID: taskID, State: state, StartedAt: now, UpdatedAt: now,
+		}); err != nil {
+			t.Fatalf("create session: %v", err)
+		}
+	}
+	if err := repo.UpsertExecutorRunning(context.Background(), &models.ExecutorRunning{
+		ID:               sessionID,
+		SessionID:        sessionID,
+		TaskID:           taskID,
+		AgentExecutionID: "exec-" + sessionID,
+		Runtime:          agentruntime.RuntimeStandalone,
+		Status:           status,
+	}); err != nil {
+		t.Fatalf("upsert executor row: %v", err)
+	}
+}
+
+func TestIdleReaper_TickFiltersRowsAndStopsPreservedRows(t *testing.T) {
+	repo := setupTestRepo(t)
+	ctx := context.Background()
+	seedIdleReaperRow(t, repo, "task-recent", "s-recent", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusRunning)
+	seedIdleReaperRow(t, repo, "task-zero", "s-zero", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusRunning)
+	seedIdleReaperRow(t, repo, "task-future", "s-future", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusRunning)
+	seedIdleReaperRow(t, repo, "task-stopped", "s-stopped", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusStopped)
+	seedIdleReaperRow(t, repo, "task-old", "s-old", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusRunning)
+
+	now := time.Now().UTC()
+	for sessionID, updatedAt := range map[string]time.Time{
+		"s-recent":  now,
+		"s-zero":    time.Time{},
+		"s-future":  now.Add(time.Hour),
+		"s-stopped": now.Add(-time.Hour),
+		"s-old":     now.Add(-time.Hour),
+	} {
+		if _, err := repo.DB().ExecContext(ctx, `UPDATE executors_running SET updated_at = ? WHERE session_id = ?`, updatedAt, sessionID); err != nil {
+			t.Fatalf("set updated_at for %s: %v", sessionID, err)
+		}
+	}
+
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{isAgentRunning: false})
+	svc.turnService = &inactiveTurnService{}
+	svc.idleReaper = newIdleSessionReaper()
+	svc.idleReaper.minIdle = 100 * time.Millisecond
+
+	svc.reclaimIdleSessionsOnce(ctx)
+	for sessionID, wantStatus := range map[string]string{
+		"s-recent":  models.ExecutorRunningStatusRunning,
+		"s-zero":    models.ExecutorRunningStatusRunning,
+		"s-future":  models.ExecutorRunningStatusRunning,
+		"s-stopped": models.ExecutorRunningStatusStopped,
+		"s-old":     models.ExecutorRunningStatusStopped,
+	} {
+		row, err := repo.GetExecutorRunningBySessionID(ctx, sessionID)
+		if err != nil {
+			t.Fatalf("load %s: %v", sessionID, err)
+		}
+		if row.Status != wantStatus {
+			t.Fatalf("%s status = %q, want %q", sessionID, row.Status, wantStatus)
+		}
+	}
+	// A preserved stopped row must not be reconsidered on the next tick.
+	svc.reclaimIdleSessionsOnce(ctx)
+	row, err := repo.GetExecutorRunningBySessionID(ctx, "s-old")
+	if err != nil || row.Status != models.ExecutorRunningStatusStopped {
+		t.Fatalf("old row changed after second tick: row=%+v err=%v", row, err)
+	}
+}
+
+func TestIdleReaper_TickSkipsLiveRuntime(t *testing.T) {
+	repo := setupTestRepo(t)
+	ctx := context.Background()
+	seedIdleReaperRow(t, repo, "task-live", "s-live", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusRunning)
+	if _, err := repo.DB().ExecContext(ctx, `UPDATE executors_running SET updated_at = ?, local_pid = ? WHERE session_id = ?`, time.Now().UTC().Add(-time.Hour), 9999, "s-live"); err != nil {
+		t.Fatal(err)
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{isAgentRunning: true})
+	svc.turnService = &inactiveTurnService{}
+	svc.idleReaper = newIdleSessionReaper()
+	svc.idleReaper.minIdle = 0
+
+	svc.reclaimIdleSessionsOnce(ctx)
+	row, err := repo.GetExecutorRunningBySessionID(ctx, "s-live")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if row.Status != models.ExecutorRunningStatusRunning || row.LocalPID != 9999 {
+		t.Fatalf("live row changed: status=%q local_pid=%d", row.Status, row.LocalPID)
+	}
+}
+
+func TestIdleReaper_TickSkipsActiveTurn(t *testing.T) {
+	repo := setupTestRepo(t)
+	ctx := context.Background()
+	seedIdleReaperRow(t, repo, "task-turn", "s-turn", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusRunning)
+	if _, err := repo.DB().ExecContext(ctx, `UPDATE executors_running SET updated_at = ? WHERE session_id = ?`, time.Now().UTC().Add(-time.Hour), "s-turn"); err != nil {
+		t.Fatal(err)
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{isAgentRunning: false})
+	svc.turnService = &alwaysActiveTurnService{}
+	svc.idleReaper = newIdleSessionReaper()
+	svc.idleReaper.minIdle = 0
+
+	svc.reclaimIdleSessionsOnce(ctx)
+	row, err := repo.GetExecutorRunningBySessionID(ctx, "s-turn")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if row.Status != models.ExecutorRunningStatusRunning {
+		t.Fatalf("active-turn row status = %q, want running", row.Status)
+	}
+}
+
+func TestIdleReaper_TickSkipsWrongState(t *testing.T) {
+	repo := setupTestRepo(t)
+	ctx := context.Background()
+	seedIdleReaperRow(t, repo, "task-running", "s-running", models.TaskSessionStateRunning, models.ExecutorRunningStatusRunning)
+	if _, err := repo.DB().ExecContext(ctx, `UPDATE executors_running SET updated_at = ? WHERE session_id = ?`, time.Now().UTC().Add(-time.Hour), "s-running"); err != nil {
+		t.Fatal(err)
+	}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{isAgentRunning: false})
+	svc.turnService = &inactiveTurnService{}
+	svc.idleReaper = newIdleSessionReaper()
+	svc.idleReaper.minIdle = 0
+
+	svc.reclaimIdleSessionsOnce(ctx)
+	row, err := repo.GetExecutorRunningBySessionID(ctx, "s-running")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if row.Status != models.ExecutorRunningStatusRunning {
+		t.Fatalf("running session status = %q, want running", row.Status)
+	}
+}
+
+func TestIdleReaper_DoesNotRepairAReplacedExecution(t *testing.T) {
+	repo := setupTestRepo(t)
+	ctx := context.Background()
+	seedIdleReaperRow(t, repo, "task-rotate", "s-rotate", models.TaskSessionStateWaitingForInput, models.ExecutorRunningStatusRunning)
+	if _, err := repo.DB().ExecContext(ctx, `UPDATE executors_running SET updated_at = ? WHERE session_id = ?`, time.Now().UTC().Add(-time.Hour), "s-rotate"); err != nil {
+		t.Fatal(err)
+	}
+
+	base := &mockAgentManager{isAgentRunning: false}
+	agent := &rotatingReaperAgentManager{AgentManagerClient: base, repo: repo}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agent)
+	svc.turnService = &inactiveTurnService{}
+	svc.idleReaper = newIdleSessionReaper()
+	svc.idleReaper.minIdle = 0
+
+	svc.reclaimIdleSessionsOnce(ctx)
+	row, err := repo.GetExecutorRunningBySessionID(ctx, "s-rotate")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if row.AgentExecutionID != "execution-new" || row.Status != models.ExecutorRunningStatusRunning {
+		t.Fatalf("successor row was repaired: execution=%q status=%q", row.AgentExecutionID, row.Status)
+	}
+}
+
+type rotatingReaperAgentManager struct {
+	executor.AgentManagerClient
+	repo *taskrepo.Repository
+}
+
+func (m *rotatingReaperAgentManager) IsAgentRunningForSession(context.Context, string) bool {
+	return false
+}
+
+func (m *rotatingReaperAgentManager) CleanupStaleExecutionBySessionIDIfCurrent(
+	ctx context.Context,
+	sessionID, _ string,
+	_ time.Time,
+) error {
+	_, err := m.repo.DB().ExecContext(ctx, `
+		UPDATE executors_running
+		SET agent_execution_id = ?, status = ?, updated_at = ?
+		WHERE session_id = ?
+	`, "execution-new", models.ExecutorRunningStatusRunning, time.Now().UTC(), sessionID)
+	return err
+}
+
+func TestIdleReaper_LoopInheritsParentCancellation(t *testing.T) {
+	r := newIdleSessionReaper()
+	r.interval = 5 * time.Millisecond
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var ticks atomic.Int32
+	if !r.start(ctx, func(context.Context) { ticks.Add(1) }) {
+		t.Fatal("start returned false")
+	}
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for ticks.Load() == 0 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if ticks.Load() == 0 {
+		t.Fatal("reaper did not tick")
+	}
+	cancel()
+	count := ticks.Load()
+	time.Sleep(30 * time.Millisecond)
+	if ticks.Load() != count {
+		t.Fatalf("ticks after parent cancellation = %d, want %d", ticks.Load(), count)
+	}
+	r.stop()
+}
+
+func TestIdleReaper_StopStartRestartsLoop(t *testing.T) {
+	r := newIdleSessionReaper()
+	r.interval = 5 * time.Millisecond
+	ctx := context.Background()
+	var ticks atomic.Int32
+	if !r.start(ctx, func(context.Context) { ticks.Add(1) }) {
+		t.Fatal("first start returned false")
+	}
+	r.stop()
+	if r.started {
+		t.Fatal("reaper remains started after stop")
+	}
+	firstCount := ticks.Load()
+	if !r.start(ctx, func(context.Context) { ticks.Add(1) }) {
+		t.Fatal("second start returned false")
+	}
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for ticks.Load() == firstCount && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if ticks.Load() == firstCount {
+		t.Fatal("reaper did not tick after restart")
+	}
+	r.stop()
+}
+
+func TestIdleReaper_StopPreventsFurtherTicks(t *testing.T) {
+	r := newIdleSessionReaper()
+	r.interval = 5 * time.Millisecond
+	var ticks atomic.Int32
+	if !r.start(context.Background(), func(context.Context) { ticks.Add(1) }) {
+		t.Fatal("start returned false")
+	}
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for ticks.Load() < 2 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if ticks.Load() < 2 {
+		t.Fatalf("ticks before stop = %d, want at least 2", ticks.Load())
+	}
+	r.stop()
+	afterStop := ticks.Load()
+	time.Sleep(30 * time.Millisecond)
+	if ticks.Load() != afterStop {
+		t.Fatalf("ticks after stop = %d, want %d", ticks.Load(), afterStop)
+	}
+}
+
+type alwaysActiveTurnService struct{}
+
+func (*alwaysActiveTurnService) GetActiveTurn(context.Context, string) (*models.Turn, error) {
+	return &models.Turn{ID: "turn-always"}, nil
+}
+func (*alwaysActiveTurnService) StartTurn(context.Context, string) (*models.Turn, error) {
+	panic("alwaysActiveTurnService: StartTurn should not be called")
+}
+func (*alwaysActiveTurnService) ReserveTurn(context.Context, string, *models.PromptDispatchRecovery) (*models.Turn, error) {
+	panic("alwaysActiveTurnService: ReserveTurn should not be called")
+}
+func (*alwaysActiveTurnService) MarkReservedTurnDispatchAttempted(context.Context, *models.Turn) error {
+	panic("alwaysActiveTurnService: MarkReservedTurnDispatchAttempted should not be called")
+}
+func (*alwaysActiveTurnService) PublishReservedTurn(context.Context, *models.Turn) error {
+	panic("alwaysActiveTurnService: PublishReservedTurn should not be called")
+}
+func (*alwaysActiveTurnService) RollbackReservedTurn(context.Context, string, string) (bool, error) {
+	panic("alwaysActiveTurnService: RollbackReservedTurn should not be called")
+}
+func (*alwaysActiveTurnService) ReconcileUnpublishedPromptTurns(context.Context) (int, error) {
+	return 0, nil
+}
+func (*alwaysActiveTurnService) CompleteTurn(context.Context, string) error { return nil }
+func (*alwaysActiveTurnService) GetTurn(context.Context, string) (*models.Turn, error) {
+	return nil, nil
+}
+func (*alwaysActiveTurnService) UpdateTurn(context.Context, *models.Turn) error { return nil }
+func (*alwaysActiveTurnService) PatchTurnMetadata(context.Context, string, string, map[string]interface{}) error {
+	return nil
+}
+func (*alwaysActiveTurnService) AbandonOpenTurns(context.Context, string) error { return nil }

apps/backend/internal/orchestrator/reconcile_liveness.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"go.uber.org/zap"
 
@@ -174,18 +175,44 @@ func (s *Service) pruneOrRepairExecutorRow(ctx context.Context, running *models.
 // satisfying #1597's "never leave a row claiming a dead process" expected
 // behavior. Best-effort; a missing row is not an error.
 func (s *Service) repairDeadRowLiveness(ctx context.Context, running *models.ExecutorRunning) error {
-	err := s.repo.RepairExecutorRunningDead(ctx, running.SessionID)
+	var err error
+	if repairer, ok := s.repo.(executionIdentityRepairer); ok {
+		err = repairer.RepairExecutorRunningDeadIfCurrent(
+			ctx,
+			running.SessionID,
+			running.AgentExecutionID,
+			running.UpdatedAt,
+		)
+	} else {
+		err = s.repo.RepairExecutorRunningDead(ctx, running.SessionID)
+	}
 	if err != nil && !errors.Is(err, models.ErrExecutorRunningNotFound) {
 		s.logger.Warn("failed to repair executor row liveness during reconciliation",
 			zap.String("session_id", running.SessionID),
 			zap.Error(err))
 	}
-	if errors.Is(err, models.ErrExecutorRunningNotFound) {
+	if errors.Is(err, models.ErrExecutorRunningNotFound) || errors.Is(err, models.ErrExecutionRotated) {
 		return nil
 	}
 	return err
 }
 
+type executionIdentityRepairer interface {
+	RepairExecutorRunningDeadIfCurrent(
+		ctx context.Context,
+		sessionID, expectedExecutionID string,
+		expectedUpdatedAt time.Time,
+	) error
+}
+
+type executionIdentityCleaner interface {
+	CleanupStaleExecutionBySessionIDIfCurrent(
+		ctx context.Context,
+		sessionID, expectedExecutionID string,
+		expectedUpdatedAt time.Time,
+	) error
+}
+
 func (s *Service) probeAgentRunning(ctx context.Context, sessionID string) (bool, error) {
 	if prober, ok := s.agentManager.(agentRunningProber); ok && prober != nil {
 		return prober.ProbeAgentRunningForSession(ctx, sessionID)
@@ -240,9 +267,8 @@ func classifyIdleReclaim(sessionState models.TaskSessionState, agentRunning bool
 // Completed}) when no live agent process and no active turn can be
 // observed. Resume token and worktree path are preserved so a later resume
 // can re-attach the session without losing context. Never touches a
-// running executor; never called from a periodic reaper (a runtime
-// auto-convergence tick is a pre-existing design gap handled by an
-// independent task, see plan docs).
+// running executor. Synchronous settle points and the periodic reaper use
+// the same fail-closed primitive.
 //
 // Synchronous call sites so far:
 //   - setSessionWaitingForInputIfRequested: subtask terminal collapse to
@@ -293,10 +319,21 @@ func (s *Service) reclaimIdleSession(ctx context.Context, sessionID string) erro
 			zap.Bool("has_active_turn", hasActiveTurn))
 		return nil
 	}
-	if err := s.agentManager.CleanupStaleExecutionBySessionID(ctx, sessionID); err != nil {
+	var cleanupErr error
+	if cleaner, ok := s.agentManager.(executionIdentityCleaner); ok {
+		cleanupErr = cleaner.CleanupStaleExecutionBySessionIDIfCurrent(
+			ctx,
+			sessionID,
+			running.AgentExecutionID,
+			running.UpdatedAt,
+		)
+	} else {
+		cleanupErr = s.agentManager.CleanupStaleExecutionBySessionID(ctx, sessionID)
+	}
+	if cleanupErr != nil {
 		s.logger.Warn("idle reclaim: cleanup failed; row preserved",
 			zap.String("session_id", sessionID),
-			zap.Error(err))
+			zap.Error(cleanupErr))
 		return nil
 	}
 	if err := s.repairDeadRowLiveness(ctx, running); err != nil {

apps/backend/internal/orchestrator/service.go

@@ -698,6 +698,13 @@ type Service struct {
 	// manager simply keeps every run's checkout. Set via SetWorktreeManager.
 	worktreeReaper automationWorktreeReaper
 
+	// idleReaper is the periodic background loop that calls
+	// reclaimIdleSession on rows older than idleReaperMinIdle. Nil-safe:
+	// callers that don't need the reaper (most tests, ephemeral
+	// orchestrator instances) leave it nil and startIdleSessionReaper
+	// / stopIdleSessionReaper no-op. See idle_session_reaper.go.
+	idleReaper *idleSessionReaper
+
 	// 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
@@ -1013,6 +1020,7 @@ func NewService(
 		reservedPromptCallbacks:      newReservedPromptCallbackOwner(),
 		sendNowCtx:                   sendNowCtx,
 		sendNowCancel:                sendNowCancel,
+		idleReaper:                   newIdleSessionReaper(),
 	}
 	// Always publish queue-status after a task-scoped queue purge so the
 	// status-summary projector zeros queued_prompt_count. Unlike the
@@ -2155,6 +2163,14 @@ func (s *Service) Start(ctx context.Context) error {
 	// Reconcile queued tasks when WIP limits or feeder settings change.
 	s.subscribeWorkflowQueueEvents()
 
+	// Start the idle-session reaper last. It depends on s.repo
+	// (already wired), s.agentManager (already wired), and the
+	// turnService-cleared reconciler so a turnService == nil error
+	// would already have aborted Start above. After this returns the
+	// background goroutine owns the reclaim tick; Service.Stop joins it
+	// before tearing down repo / agentManager.
+	s.startIdleSessionReaper(ctx)
+
 	s.logger.Info("orchestrator service started successfully")
 	return nil
 }
@@ -2173,6 +2189,7 @@ func (s *Service) Stop() error {
 
 	// Stop components in reverse order
 	var errs []error
+	s.stopIdleSessionReaper()
 	s.stopReservedPromptCallbacks()
 
 	if err := s.scheduler.Stop(); err != nil {

apps/backend/internal/orchestrator/task_operations.go

@@ -365,6 +365,9 @@ func (s *Service) StartCreatedSession(
 	attachments []v1.MessageAttachment,
 	references []v1.EntityReference,
 ) (*executor.TaskExecution, error) {
+	releaseLifecycleLock := s.acquireSessionLifecycleLock(sessionID)
+	defer releaseLifecycleLock()
+
 	// One GetWorkflowMeta read shared by profile resolution and prompt build.
 	ctx = withWorkflowMetaCache(ctx)
 

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

@@ -237,6 +237,27 @@ func (r *Repository) ListExecutorsRunning(ctx context.Context) ([]*models.Execut
 	return scanExecutorRunningRows(rows)
 }
 
+// ListExecutorsRunningIdle returns rows that can be candidates for periodic
+// stale-runtime recovery. Filtering in SQL avoids rescanning preserved stopped
+// rows and recent executions on every reaper tick.
+func (r *Repository) ListExecutorsRunningIdle(ctx context.Context, cutoff time.Time) ([]*models.ExecutorRunning, error) {
+	rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
+		SELECT id, session_id, task_id, execution_profile_id, executor_id, runtime, status, resumable, resume_token,
+			last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, local_pid,
+			worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, metadata,
+			created_at, updated_at
+		FROM executors_running
+		WHERE status <> ? AND updated_at <= ?
+		ORDER BY updated_at DESC
+	`), models.ExecutorRunningStatusStopped, cutoff)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+
+	return scanExecutorRunningRows(rows)
+}
+
 func (r *Repository) ListExecutorsRunningByTaskID(ctx context.Context, taskID string) ([]*models.ExecutorRunning, error) {
 	if taskID == "" {
 		return nil, fmt.Errorf("task_id is required")
@@ -486,6 +507,83 @@ func (r *Repository) RepairExecutorRunningDead(ctx context.Context, sessionID st
 	return nil
 }
 
+// RepairExecutorRunningDeadIfCurrent is the compare-and-set variant used by
+// concurrent cleanup. The row must still belong to expectedExecID and, when
+// supplied, must have the same UpdatedAt value observed by the caller.
+func (r *Repository) RepairExecutorRunningDeadIfCurrent(
+	ctx context.Context,
+	sessionID, expectedExecID string,
+	expectedUpdatedAt time.Time,
+) error {
+	if sessionID == "" {
+		return fmt.Errorf("session_id is required")
+	}
+	now := time.Now().UTC()
+	query := `
+		UPDATE executors_running
+		   SET status = ?, local_pid = 0, last_seen_at = ?, updated_at = ?
+		 WHERE session_id = ? AND agent_execution_id = ?
+	`
+	args := []interface{}{
+		models.ExecutorRunningStatusStopped, now, now, sessionID, expectedExecID,
+	}
+	if !expectedUpdatedAt.IsZero() {
+		query += " AND updated_at = ?\n"
+		args = append(args, expectedUpdatedAt)
+	}
+	result, err := r.db.ExecContext(ctx, r.db.Rebind(query), args...)
+	if err != nil {
+		return err
+	}
+	rows, _ := result.RowsAffected()
+	if rows > 0 {
+		return nil
+	}
+	exists, err := r.HasExecutorRunningRow(ctx, sessionID)
+	if err != nil {
+		return err
+	}
+	if !exists {
+		return fmt.Errorf("%w for session: %s", models.ErrExecutorRunningNotFound, sessionID)
+	}
+	return models.ErrExecutionRotated
+}
+
+// DeleteExecutorRunningIfCurrent deletes a row only when its execution
+// identity and observed timestamp still match. It prevents cleanup of a row
+// that a successor launch has already replaced.
+func (r *Repository) DeleteExecutorRunningIfCurrent(
+	ctx context.Context,
+	sessionID, expectedExecID string,
+	expectedUpdatedAt time.Time,
+) error {
+	if sessionID == "" {
+		return fmt.Errorf("session_id is required")
+	}
+	query := `DELETE FROM executors_running WHERE session_id = ? AND agent_execution_id = ?`
+	args := []interface{}{sessionID, expectedExecID}
+	if !expectedUpdatedAt.IsZero() {
+		query += " AND updated_at = ?\n"
+		args = append(args, expectedUpdatedAt)
+	}
+	result, err := r.db.ExecContext(ctx, r.db.Rebind(query), args...)
+	if err != nil {
+		return err
+	}
+	rows, _ := result.RowsAffected()
+	if rows > 0 {
+		return nil
+	}
+	exists, err := r.HasExecutorRunningRow(ctx, sessionID)
+	if err != nil {
+		return err
+	}
+	if !exists {
+		return fmt.Errorf("%w for session: %s", models.ErrExecutorRunningNotFound, sessionID)
+	}
+	return models.ErrExecutionRotated
+}
+
 // UpdateExecutorRunningStatus narrowly updates the status column.
 // Returns ErrExecutorRunningNotFound when no row exists for the session.
 func (r *Repository) UpdateExecutorRunningStatus(ctx context.Context, sessionID, status string) error {

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

@@ -201,6 +201,54 @@ func TestRepairExecutorRunningDead(t *testing.T) {
 	}
 }
 
+func TestRepairExecutorRunningDeadIfCurrentRejectsRotatedExecution(t *testing.T) {
+	repo := newRepoForSessionTests(t)
+	ctx := context.Background()
+	seedExecutorRunningCleanupTask(t, repo, "task-cas")
+	if err := repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID: "session-cas", TaskID: "task-cas", State: models.TaskSessionStateWaitingForInput,
+	}); err != nil {
+		t.Fatalf("CreateTaskSession: %v", err)
+	}
+	if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{
+		ID: "session-cas", SessionID: "session-cas", TaskID: "task-cas",
+		AgentExecutionID: "execution-old", Status: models.ExecutorRunningStatusRunning,
+	}); err != nil {
+		t.Fatalf("UpsertExecutorRunning: %v", err)
+	}
+	observed, err := repo.GetExecutorRunningBySessionID(ctx, "session-cas")
+	if err != nil {
+		t.Fatalf("GetExecutorRunningBySessionID: %v", err)
+	}
+
+	if _, err := repo.DB().ExecContext(ctx, `
+		UPDATE executors_running
+		SET agent_execution_id = ?, status = ?, updated_at = ?
+		WHERE session_id = ?
+	`, "execution-new", models.ExecutorRunningStatusRunning, time.Now().UTC(), "session-cas"); err != nil {
+		t.Fatalf("rotate execution row: %v", err)
+	}
+	if err := repo.RepairExecutorRunningDeadIfCurrent(ctx, "session-cas", observed.AgentExecutionID, observed.UpdatedAt); !errors.Is(err, models.ErrExecutionRotated) {
+		t.Fatalf("CAS repair error = %v, want ErrExecutionRotated", err)
+	}
+	current, err := repo.GetExecutorRunningBySessionID(ctx, "session-cas")
+	if err != nil {
+		t.Fatalf("GetExecutorRunningBySessionID after rejected repair: %v", err)
+	}
+	if current.AgentExecutionID != "execution-new" || current.Status != models.ExecutorRunningStatusRunning {
+		t.Fatalf("successor row changed after rejected repair: execution=%q status=%q", current.AgentExecutionID, current.Status)
+	}
+	if err := repo.DeleteExecutorRunningIfCurrent(ctx, "session-cas", observed.AgentExecutionID, observed.UpdatedAt); !errors.Is(err, models.ErrExecutionRotated) {
+		t.Fatalf("CAS delete error = %v, want ErrExecutionRotated", err)
+	}
+	if err := repo.DeleteExecutorRunningIfCurrent(ctx, "session-cas", current.AgentExecutionID, current.UpdatedAt); err != nil {
+		t.Fatalf("delete current execution: %v", err)
+	}
+	if _, err := repo.GetExecutorRunningBySessionID(ctx, "session-cas"); !errors.Is(err, models.ErrExecutorRunningNotFound) {
+		t.Fatalf("row after current delete error = %v, want ErrExecutorRunningNotFound", err)
+	}
+}
+
 // TestExecutorRunningLocalPIDMigrationOnLegacyDB is the same-database replay test
 // mandated by ADR 0027 for a schema change: it proves the `local_pid` ADD COLUMN
 // migration upgrades a DB that PREDATES the column, adding it with the default

docs/decisions/0025-runtime-cleanup-uses-executors-running.md

@@ -1,25 +1,30 @@
 # 0025: Runtime Cleanup Uses `executors_running`
 
-**Status:** accepted (amended 2026-07-06 — see "Update")
+**Status:** accepted (amended 2026-08-20 — see "Update")
 **Date:** 2026-06-22
 **Area:** backend
 
+## Update (2026-08-20, #2836 idle-session-reaper)
+
+The event path remains primary. Startup reconciliation and a bounded periodic
+reaper now provide recovery when an event is lost while Kandev is running. The
+reaper selects old, non-stopped rows and delegates to the same fail-closed
+reclaim primitive. Execution identity and row timestamps are checked before
+cleanup and repair, so a successor launch cannot be stopped by stale work.
+
 ## Update (2026-07-06, #1597 executor-row-desync)
 
 This decision **stays**: `executors_running` remains the authoritative durable
 runtime inventory. It was made *trustworthy* rather than reverted. Three
 clarifications now hold:
 
-- **Events are the primary producer; startup reconciliation heals what events
-  cannot.** Every lifecycle transition writes the row (launch, boot-ready,
-  turn-complete, cancel, process-exit/crash, stop), populating a host-local
-  liveness handle (`executors_running.local_pid`) for local/standalone rows. A
-  startup pass repairs rows whose process is confirmed dead and prunes only
-  terminal ones — a backend restart is exactly the moment events could not
-  have fired. A periodic in-run polling pass was prototyped and deliberately
-  not merged: it defended against failure modes (OOM kill, dropped event)
-  that have not been observed, and polling never updates a row the moment an
-  event could.
+- **Events are the primary producer; reconciliation heals what events cannot.**
+  Every lifecycle transition writes the row (launch, boot-ready, turn-complete,
+  cancel, process-exit/crash, stop), populating a host-local liveness handle
+  (`executors_running.local_pid`) for local/standalone rows. Startup
+  reconciliation repairs rows after a backend restart. The periodic reaper
+  covers lost events during normal uptime and uses the same guarded reclaim
+  policy; neither recovery path replaces event-driven updates.
 - **`local_pid` is a NEW column, deliberately not the existing `pid`.**
   `executors_running.pid` holds the agentctl PID *on the remote host* for SSH
   rows; overloading it with a host-local pid would silently change that