diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go b/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go index a35859bd2b..f0b991133c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go +++ b/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 diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go b/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go index 76161b15ac..2f08d1f1d5 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go +++ b/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 } diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_startup.go b/apps/backend/internal/agent/runtime/lifecycle/manager_startup.go index 6b3d0d2cc4..b62c6fa13e 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_startup.go +++ b/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) diff --git a/apps/backend/internal/agent/runtime/lifecycle/persistence.go b/apps/backend/internal/agent/runtime/lifecycle/persistence.go index d43d7088e5..89a8e4fe2c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/persistence.go +++ b/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 +} diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index d9d0487ce0..0715e85767 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/idle_session_reaper.go b/apps/backend/internal/orchestrator/idle_session_reaper.go new file mode 100644 index 0000000000..104fa514c8 --- /dev/null +++ b/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) +} diff --git a/apps/backend/internal/orchestrator/idle_session_reaper_test.go b/apps/backend/internal/orchestrator/idle_session_reaper_test.go new file mode 100644 index 0000000000..74d3b45e90 --- /dev/null +++ b/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 } diff --git a/apps/backend/internal/orchestrator/reconcile_liveness.go b/apps/backend/internal/orchestrator/reconcile_liveness.go index 149bad43b1..3952991f0f 100644 --- a/apps/backend/internal/orchestrator/reconcile_liveness.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 0788f1070d..564b25084c 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 2cf4670002..dba2df2e99 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/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) diff --git a/apps/backend/internal/task/repository/sqlite/executor.go b/apps/backend/internal/task/repository/sqlite/executor.go index dc792c8a90..1ed2f0c537 100644 --- a/apps/backend/internal/task/repository/sqlite/executor.go +++ b/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 { diff --git a/apps/backend/internal/task/repository/sqlite/executor_localpid_test.go b/apps/backend/internal/task/repository/sqlite/executor_localpid_test.go index b3a512c186..b709d5f871 100644 --- a/apps/backend/internal/task/repository/sqlite/executor_localpid_test.go +++ b/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 diff --git a/docs/decisions/0025-runtime-cleanup-uses-executors-running.md b/docs/decisions/0025-runtime-cleanup-uses-executors-running.md index 1c3fd4084d..e98ed1ec91 100644 --- a/docs/decisions/0025-runtime-cleanup-uses-executors-running.md +++ b/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