fix(tasks): reap orphaned processes left running after workspace cleanup ↗

Durable snapshot and job integration

The job gathers reap roots before removal, confirms them after removal, and runs the reap phase last when stops are clean.

type taskResourceCleanupSnapshot struct {
	Sessions               []*models.TaskSession     `json:"sessions,omitempty"`
	Worktrees              []*worktree.Worktree      `json:"worktrees,omitempty"`
	SSHTaskDirs []sshReclaimTarget `json:"ssh_task_dirs,omitempty"`
	OrphanReapRoots []string `json:"orphan_reap_roots,omitempty"`
	OrphanReapRecords []orphanReapCandidateRecord `json:"orphan_reap_records,omitempty"`
	OrphanReapSkips []orphanReapSkipRecord `json:"orphan_reap_skips,omitempty"`
}

reapRootCandidates := s.gatherOrphanReapRootCandidates(
	snapshot, cleanupSessionIDs(snapshot.Sessions, targets),
)
errs := s.performTaskCleanup(ctx, job.TaskID, snapshot.Sessions, snapshot.Worktrees, targets,
	taskEnvironmentCleanup{
		env: snapshot.TaskEnvironment, deleteRow: snapshot.DeleteEnvironmentRow,
		preserveBranches:       job.IsArchive(),
		discardWorktreeChanges: snapshot.DiscardWorktreeChanges,
	},
	taskCleanupPreserveRows(stopOutcome))
snapshot.OrphanReapRoots = mergeOrphanReapRoots(
	snapshot.OrphanReapRoots, confirmOrphanReapRootsRemoved(reapRootCandidates),
)
if len(failedStops) == 0 {
	errs = append(errs, s.reclaimSSHTaskDirs(ctx, job, snapshot)...)
}
if len(failedStops) == 0 {
	errs = append(errs, s.runOrphanReapPhase(ctx, job, snapshot)...)
}

Root candidates and phase orchestration

The phase resolves roots, skips roots that exist again, takes one snapshot, attributes candidates, and applies ownership and signalling.

func (s *Service) gatherOrphanReapRootCandidates(
	snapshot *taskResourceCleanupSnapshot,
	sessionIDs []string,
) []string {
	seen := make(map[string]struct{})
	var candidates []string
	add := func(path string) {
		path = resolveOrphanReapPathBestEffort(path)
		if path == "" {
			return
		}
		if _, err := os.Lstat(path); err != nil {
			return
		}
		if _, ok := seen[path]; ok {
			return
		}
		seen[path] = struct{}{}
		candidates = append(candidates, path)
	}
	for _, wt := range snapshot.Worktrees {
		if wt == nil || wt.Path == "" {
			continue
		}
		add(wt.Path)
		add(filepath.Dir(wt.Path))
	}
	if s.quickChatDir != "" {
		for _, sessionID := range sessionIDs {
			add(filepath.Join(s.quickChatDir, sessionID))
		}
	}
	return candidates
}

func (s *Service) runOrphanReapPhase(ctx context.Context, job *models.TaskResourceCleanupJob, snapshot *taskResourceCleanupSnapshot) []error {
	if len(snapshot.OrphanReapRoots) == 0 {
		return nil
	}
	resolvedRoots := resolveOrphanReapRoots(snapshot.OrphanReapRoots)
	activeRoots := make([]string, 0, len(resolvedRoots))
	for _, root := range resolvedRoots {
		_, err := os.Lstat(root)
		switch {
		case err == nil:
			s.recordOrphanReapRootSkip(snapshot, root, "reap root exists again at reap time")
		case errors.Is(err, os.ErrNotExist):
			activeRoots = append(activeRoots, root)
		default:
			s.recordOrphanReapRootSkipDetectionFailure(snapshot, root, "cannot confirm reap root state: "+err.Error())
		}
	}
	if len(activeRoots) == 0 {
		return nil
	}
	snap, err := s.takeOrphanReapHostSnapshot(ctx)
	if err != nil {
		if errors.Is(err, errOrphanReapUnsupportedPlatform) {
			s.recordOrphanReapPhaseSkip(snapshot, "unsupported platform "+runtime.GOOS)
			return nil
		}
		s.recordOrphanReapPhaseSkipDetectionFailure(snapshot, "host process snapshot unavailable: "+err.Error())
		return nil
	}
	byRoot := attributeOrphanReapCandidates(snap, activeRoots)
	toSignal := s.applyOrphanReapOwnership(ctx, job.TaskID, snap, byRoot, snapshot)
	sort.Slice(toSignal, func(i, j int) bool { return toSignal[i].PID < toSignal[j].PID })
	phaseResult := s.signalOrphanReapCandidatesWithLimit(ctx, job.TaskID, toSignal, snapshot, orphanReapMaxCandidates)
	return phaseResult.errs
}

Fail-closed ownership filter

The filter blocks a root or candidate when another task owns the workspace, the PID, or an ancestor, and fails closed on any inconclusive check.

func (s *Service) applyOrphanReapOwnership(ctx context.Context, taskID string, snap []hostProcess, byRoot map[string][]orphanReapCandidate, snapshot *taskResourceCleanupSnapshot) []orphanReapCandidate {
	ppidByPID := make(map[int]int, len(snap))
	for _, p := range snap {
		ppidByPID[p.PID] = p.PPID
	}
	protected, protectedInconclusive := orphanReapProtectedPIDs(ppidByPID)
	if protectedInconclusive {
		s.skipEveryOrphanReapRoot(snapshot, byRoot, "ownership check inconclusive: could not resolve the backend's own ancestry")
		return nil
	}
	otherExecutors, execErr := s.executors.ListExecutorsRunning(ctx)
	if execErr != nil {
		s.skipEveryOrphanReapRoot(snapshot, byRoot, "ownership check failed: "+execErr.Error())
		return nil
	}
	otherSessions, sessErr := s.sessions.ListLiveWorkspaceSessions(ctx)
	if sessErr != nil {
		s.skipEveryOrphanReapRoot(snapshot, byRoot, "ownership check failed: "+sessErr.Error())
		return nil
	}
	for root, candidates := range byRoot {
		if owner, blocked := orphanReapFindOverlap(otherSessionPaths, root); blocked {
			s.recordOrphanReapRootSkip(snapshot, root, "workspace not exclusively owned: session of task "+owner)
			continue
		}
		if owner, blocked := orphanReapFindContainment(liveWorktreeRoots, root); blocked {
			s.recordOrphanReapRootSkip(snapshot, root, "another task's live recorded execution occupies this workspace: task "+owner)
			continue
		}
		for _, cand := range candidates {
			toSignal = s.applyOrphanReapPerCandidateOwnership(snapshot, taskID, cand, protected, ppidByPID, localPIDOwner, toSignal)
		}
	}
	return toSignal
}

SIGTERM to SIGKILL escalation

The phase sends SIGTERM to all candidates, waits 2 seconds, sends SIGKILL to survivors, and re-verifies cwd before each signal.

func (s *Service) signalOrphanReapCandidatesWithLimit(ctx context.Context, taskID string, candidates []orphanReapCandidate, snapshot *taskResourceCleanupSnapshot, maxCandidates int) orphanReapSignalPhaseResult {
	verifier := s.orphanReapVerifier
	if verifier == nil {
		verifier = defaultOrphanReapVerifier()
	}
	signaler := s.orphanReapSignaler
	if signaler == nil {
		signaler = realOrphanReapSignaler{}
	}
	if ctx.Err() != nil {
		return orphanReapSignalPhaseResult{errs: []error{errOrphanReapCancelledMidPhase}}
	}
	termResult := s.sendOrphanReapSigterms(ctx, taskID, candidates, snapshot, verifier, signaler, maxCandidates)
	pending := termResult.pending
	if len(pending) == 0 {
		return result
	}
	graceTimer := time.NewTimer(orphanReapGraceDelay)
	select {
		case <-ctx.Done():
			result.errs = s.recordOrphanReapCancelledMidPhase(snapshot, taskID, pending)
			return result
		case <-graceTimer.C:
		}
	killPending := s.sendOrphanReapSigkills(ctx, taskID, pending, snapshot, verifier, signaler)
	settleTimer := time.NewTimer(orphanReapSettleDelay)
	select {
		case <-ctx.Done():
			result.errs = s.recordOrphanReapCancelledMidPhase(snapshot, taskID, killPending)
			return result
		case <-settleTimer.C:
		}
	result.errs = s.resolveOrphanReapSurvivors(ctx, taskID, killPending, snapshot, verifier, signaler)
	return result
}

func orphanReapReverifyInsideRoot(ctx context.Context, verifier orphanReapVerifier, pid int, root string) bool {
	if !orphanReapRootIsAbsent(root) {
		return false
	}
	verifyCtx, cancel := context.WithTimeout(ctx, orphanReapVerifyTimeout)
	defer cancel()
	cwd, err := verifier.VerifyCwd(verifyCtx, pid)
	if err != nil || cwd == "" {
		return false
	}
	if !orphanReapRootIsAbsent(root) {
		return false
	}
	return orphanReapPathWithinRoot(root, cwd)
}

Effective workspace path for ownership

The ownership check needs the effective workspace path that prefers the linked environment path over the stale session column.

func (r *Repository) ListLiveWorkspaceSessions(ctx context.Context) ([]*models.TaskSession, error) {
	ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListLiveWorkspaceSessions")
	defer span.End()
	rows, err := r.ro.QueryContext(ctx,
		`SELECT `+taskSessionSelectCols+` `+taskSessionFromClause+
			` WHERE ts.state IN ('CREATED', 'STARTING', 'RUNNING', 'IDLE', 'WAITING_FOR_INPUT')`,
	)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	return r.scanTaskSessions(ctx, rows)
}

type SessionRepository interface {
	ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error)
	ListActiveTaskSessionsByTaskID(ctx context.Context, taskID string) ([]*models.TaskSession, error)
	ListLiveWorkspaceSessions(ctx context.Context) ([]*models.TaskSession, error)
}

Host snapshot and path matching

The phase matches processes by working directory on a component boundary and attributes a process to the deepest root.

func attributeOrphanReapCandidates(snapshot []hostProcess, roots []string) map[string][]orphanReapCandidate {
	ordered := append([]string(nil), roots...)
	sortByDescendingLength(ordered)
	byRoot := make(map[string][]orphanReapCandidate)
	for _, proc := range snapshot {
		if proc.Cwd == "" {
			continue
		}
		for _, root := range ordered {
			if orphanReapPathWithinRoot(root, proc.Cwd) {
				byRoot[root] = append(byRoot[root], orphanReapCandidate{hostProcess: proc, Root: root})
				break
			}
		}
	}
	return byRoot
}
func orphanReapPathWithinRoot(root, candidate string) bool {
	if root == "" || candidate == "" {
		return false
	}
	if candidate == root {
		return true
	}
	prefix := root
	if !strings.HasSuffix(prefix, string(filepath.Separator)) {
		prefix += string(filepath.Separator)
	}
	return strings.HasPrefix(candidate, prefix)
}

func readProcCwd(pid int) (string, error) {
	target, err := os.Readlink("/proc/" + strconv.Itoa(pid) + "/cwd")
	if err != nil {
		return "", err
	}
	return trimProcCwdDeletedSuffix(target), nil
}
func trimProcCwdDeletedSuffix(target string) string {
	return strings.TrimSuffix(target, procDeletedSuffix)
}