feat(workspace-cleanup): classify released orchestration worker workspaces for cleanup

A workspace a worker start created gets the 'orchestration-worker' reason once every dispatch on it released its terminal and its work landed in the default branch, a merged review or the coordinator's branch. Commits the coordinator's branch already holds no longer count as unpushed.
This commit is contained in:
Kelvin Amoaba
2026-11-13 02:34:02 +00:00
parent 309975de24
commit 936579113d
14 changed files with 664 additions and 74 deletions
@@ -4,6 +4,11 @@ import type { Worktree } from '../../../shared/worktree/types'
import { resolveGitMetadataPath } from '../../../shared/git/metadata/git-metadata-path'
import { getPersistedWorkspaceCleanupActivityAt } from '../../../shared/workspace-cleanup/workspace-cleanup'
import type { WorkspaceCleanupGitRoute } from './workspace-cleanup-git-route'
import {
WORKSPACE_CLEANUP_GIT_READ_TIMEOUT_MS,
WorkspaceCleanupScanCancelledError,
withWorkspaceCleanupTimeout
} from './workspace-cleanup-scan-primitives'
type StatPath = (targetPath: string) => Promise<{ mtimeMs: number }>
type ReadTextFile = (targetPath: string, options?: { tailBytes?: number }) => Promise<string>
@@ -179,3 +184,34 @@ async function readMtime(targetPath: string, statPath: StatPath): Promise<number
return 0
}
}
export async function resolveWorkspaceCleanupActivityWithTimeout(
route: WorkspaceCleanupGitRoute,
worktree: Worktree,
onActivityStatsUnavailable: () => void,
signal?: AbortSignal,
fsActivityCache?: WorkspaceCleanupFsActivityCache
): Promise<Worktree> {
try {
return await withWorkspaceCleanupTimeout(
() =>
resolveWorkspaceCleanupActivityWorktree(
route,
worktree,
undefined,
undefined,
fsActivityCache
),
WORKSPACE_CLEANUP_GIT_READ_TIMEOUT_MS,
'Timed out reading worktree activity.',
signal
)
} catch (error) {
if (error instanceof WorkspaceCleanupScanCancelledError) {
throw error
}
onActivityStatsUnavailable()
console.warn('Workspace cleanup activity scan failed', error)
return resolvePersistedWorkspaceCleanupActivityWorktree(worktree)
}
}
@@ -9,6 +9,7 @@ import {
getWorkspaceCleanupAutomationRunReasons,
getWorkspaceCleanupInactivityReasons,
getWorkspaceCleanupMergeReasons,
getWorkspaceCleanupOrchestrationWorkerReasons,
isWorkspaceOldForCleanup,
type WorkspaceCleanupAutomationRunLookup,
type WorkspaceCleanupBlocker,
@@ -26,6 +27,7 @@ import {
} from '../../../shared/workspace-cleanup/workspace-cleanup-candidate-fields'
import type { WorkspaceCleanupGitRoute } from './workspace-cleanup-git-route'
import type { RuntimeStore } from '../../runtime/construction/runtime-store-contract'
import type { WorkspaceCleanupReleasedWorker } from './workspace-cleanup-orchestration-workers'
export async function buildWorkspaceCleanupCandidate(args: {
repo: Repo
@@ -36,9 +38,20 @@ export async function buildWorkspaceCleanupCandidate(args: {
skipGit: boolean
forceGitCheck: boolean
findAutomationRun: WorkspaceCleanupAutomationRunLookup
releasedWorker?: WorkspaceCleanupReleasedWorker
signal?: AbortSignal
}): Promise<WorkspaceCleanupCandidate> {
const { repo, worktree, scannedAt, route, prCache, skipGit, forceGitCheck, signal } = args
const {
repo,
worktree,
scannedAt,
route,
prCache,
skipGit,
forceGitCheck,
releasedWorker,
signal
} = args
const blockers: WorkspaceCleanupBlocker[] = []
const reasons = getWorkspaceCleanupWorkspaceReasons(worktree, scannedAt, args.findAutomationRun)
const repoIsFolder = isFolderRepo(repo)
@@ -64,9 +77,20 @@ export async function buildWorkspaceCleanupCandidate(args: {
const gitEvidence = !shouldReadGit
? createEmptyWorkspaceCleanupGitEvidence()
: await readWorkspaceCleanupGitEvidence(worktree, repo, route, prCache, signal)
: await readWorkspaceCleanupGitEvidence(
worktree,
repo,
route,
prCache,
signal,
releasedWorker?.coordinatorBranch
)
appendWorkspaceCleanupItems(blockers, gitEvidence.blockers)
appendWorkspaceCleanupItems(reasons, getWorkspaceCleanupMergeReasons(gitEvidence.merge))
appendWorkspaceCleanupItems(
reasons,
getWorkspaceCleanupOrchestrationWorkerReasons(releasedWorker !== undefined, gitEvidence.merge)
)
const candidateWithoutFingerprint: WorkspaceCleanupCandidate = {
worktreeId: worktree.id,
@@ -54,7 +54,8 @@ export async function readWorkspaceCleanupGitEvidence(
repo: Repo,
repoRoute: WorkspaceCleanupGitRoute,
prCache: PRCacheView<PRInfo>,
signal?: AbortSignal
signal?: AbortSignal,
coordinatorBranch?: string | null
): Promise<WorkspaceCleanupGitEvidence> {
const blockers: WorkspaceCleanupBlocker[] = []
let status: GitStatusResult
@@ -112,8 +113,14 @@ export async function readWorkspaceCleanupGitEvidence(
const upstreamAhead = status.upstreamStatus.hasUpstream ? status.upstreamStatus.ahead : null
const upstreamBehind = status.upstreamStatus.hasUpstream ? status.upstreamStatus.behind : null
const merge = await readWorkspaceCleanupMergeEvidence(worktree, repo, prCache, signal)
// Commits already in the base or a merged review are not at risk, even after a squash rewrote them.
const merge = await readWorkspaceCleanupMergeEvidence(
worktree,
repo,
prCache,
signal,
coordinatorBranch
)
// Commits already in the base, a merged review or the coordinator's branch are not at risk, even after a squash rewrote them.
const headLanded = isWorkspaceCleanupHeadLanded(merge)
if (!headLanded && upstreamAhead !== null && upstreamAhead > 0) {
blockers.push('unpushed-commits')
@@ -24,11 +24,12 @@ export async function readWorkspaceCleanupMergeEvidence(
worktree: Worktree,
repo: Repo,
prCache: PRCacheView<PRInfo>,
signal?: AbortSignal
signal?: AbortSignal,
coordinatorBranch?: string | null
): Promise<WorkspaceCleanupMergeEvidence | undefined> {
try {
return await withWorkspaceCleanupTimeout(
(signal) => readMergeEvidence(worktree, repo, prCache, signal),
(signal) => readMergeEvidence(worktree, repo, prCache, signal, coordinatorBranch),
WORKSPACE_CLEANUP_GIT_READ_TIMEOUT_MS,
'Timed out reading merge evidence.',
signal
@@ -45,7 +46,8 @@ async function readMergeEvidence(
worktree: Worktree,
repo: Repo,
prCache: PRCacheView<PRInfo>,
signal: AbortSignal
signal: AbortSignal,
coordinatorBranch: string | null | undefined
): Promise<WorkspaceCleanupMergeEvidence | undefined> {
const git: CleanupGit = (argv) => gitExecFileAsync(argv, { cwd: worktree.path, signal })
const baseOid = await resolveWorkspaceCleanupBaseOid(worktree, repo, git)
@@ -63,34 +65,48 @@ async function readMergeEvidence(
}
const branchName = worktree.branch.replace(/^refs\/heads\//, '')
// The proof reads the branch ref, so it only speaks for HEAD while HEAD is still that branch.
// Prove against the base alone: the default target list includes the worktree's HEAD, the branch itself.
const headInDefaultBranch =
aheadOfDefault === 0 ||
(branchName.length > 0 &&
(await resolveCommitOid(git, `refs/heads/${branchName}`)) === headOid &&
(await localBranchHasNoUnmergedChanges({
cwd: worktree.path,
branchName,
targetRefs: [baseOid],
refreshRemoteTargets: false,
options: { signal }
})))
// Prove against the target alone: the default target list includes the worktree's HEAD, the branch itself.
let branchAtHead: Promise<boolean> | undefined
const headLandedIn = async (targetOid: string): Promise<boolean> =>
branchName.length > 0 &&
(await (branchAtHead ??= resolveCommitOid(git, `refs/heads/${branchName}`).then(
(oid) => oid === headOid
))) &&
localBranchHasNoUnmergedChanges({
cwd: worktree.path,
branchName,
targetRefs: [targetOid],
refreshRemoteTargets: false,
options: { signal }
})
const headInDefaultBranch = aheadOfDefault === 0 || (await headLandedIn(baseOid))
const review = readWorkspaceReviewPR(worktree, repo, prCache)
const head = headOid.toLowerCase()
const mergedReview =
review?.state === 'merged'
? {
number: review.number,
provider: 'github' as const,
containsHead: [review.headSha, review.confirmedContainedHeadOid].some(
(oid) => oid?.trim().toLowerCase() === head
)
}
: undefined
// Only worth proving when HEAD has not already landed another way.
const coordinatorOid =
coordinatorBranch && !headInDefaultBranch && mergedReview?.containsHead !== true
? await resolveCommitOid(git, `refs/heads/${coordinatorBranch}`)
: null
const headInCoordinatorBranch =
coordinatorOid === null
? undefined
: (await git(['rev-list', '--count', `${coordinatorOid}..${headOid}`])).stdout.trim() ===
'0' || (await headLandedIn(coordinatorOid))
return {
headInDefaultBranch,
aheadOfDefault,
...(review?.state === 'merged'
? {
mergedReview: {
number: review.number,
provider: 'github' as const,
containsHead: [review.headSha, review.confirmedContainedHeadOid].some(
(oid) => oid?.trim().toLowerCase() === head
)
}
}
: {})
...(headInCoordinatorBranch !== undefined ? { headInCoordinatorBranch } : {}),
...(mergedReview ? { mergedReview } : {})
}
}
@@ -0,0 +1,281 @@
// Real Git and a real orchestration database: squash proofs and release rows are what mocks cannot model.
import { execFile } from 'node:child_process'
import { mkdtemp, realpath, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { promisify } from 'node:util'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { installRealWorkerThreadChildProcesses } from '../../worker-thread-child-process/worker-thread-child-process-real.test-fixtures'
import { removeTree } from '../../../shared/windows-transient-lock-removal'
import type { Repo } from '../../../shared/repo-types'
import type { WorktreeMeta } from '../../../shared/worktree/meta-types'
import type { WorktreeLineage } from '../../../shared/worktree/lineage-types'
import type { RuntimeStore } from '../../runtime/construction/runtime-store-contract'
import { OrchestrationDb } from '../../runtime/orchestration/db/orchestration-db'
import '../../runtime/orchestration/db/contract-constants'
import '../../runtime/orchestration/db/messages/mailbox-routing-page'
import '../../runtime/orchestration/db/dispatch-row-writer'
import '../../runtime/orchestration/db/attempt-outcome-projection'
import { scanWorkspaceCleanup } from './workspace-cleanup-scan'
const execFileAsync = promisify(execFile)
// Before any worker the tests start, whose terminals SQLite stamps with the real time.
const WORKSPACE_CREATED_AT = Date.parse('2026-01-01T00:00:00Z')
installRealWorkerThreadChildProcesses()
let scratchDir = ''
let repoPath = ''
let db: OrchestrationDb
const meta: Record<string, WorktreeMeta> = {}
const lineage: Record<string, WorktreeLineage> = {}
async function git(args: string[], cwd: string): Promise<string> {
const { stdout } = await execFileAsync('git', args, { cwd })
return stdout.trim()
}
async function commitFile(cwd: string, name: string): Promise<void> {
await writeFile(join(cwd, name), `${name}\n`)
await git(['add', name], cwd)
await git(['commit', '-qm', name], cwd)
}
async function addWorktree(name: string, from: string): Promise<{ id: string; path: string }> {
const path = join(scratchDir, name)
await git(['worktree', 'add', '-q', '-b', name, path, from], repoPath)
const id = `repo::${path}`
meta[id] = {
instanceId: `${name}-instance`,
displayName: name,
comment: '',
linkedIssue: null,
linkedPR: null,
linkedLinearIssue: null,
isArchived: false,
isUnread: false,
isPinned: false,
sortOrder: 0,
lastActivityAt: WORKSPACE_CREATED_AT,
createdAt: WORKSPACE_CREATED_AT
}
return { id, path }
}
async function addChild(
name: string,
coordinatorId: string
): Promise<{ id: string; path: string }> {
const child = await addWorktree(name, 'coordinator')
lineage[child.id] = {
worktreeId: child.id,
worktreeInstanceId: `${name}-instance`,
parentWorktreeId: coordinatorId,
parentWorktreeInstanceId: 'coordinator-instance',
origin: 'cli',
capture: { source: 'explicit-cli-flag', confidence: 'explicit' },
createdAt: WORKSPACE_CREATED_AT
}
return child
}
/** Starts a worker on the workspace and settles its report; follow-ups reuse the same terminal. */
function startWorker(
worktreeId: string,
handle: string,
options: {
action?: 'created_child' | 'reused'
terminalOwnership?: 'created' | 'external'
hostScope?: string
} = {}
): string {
const task = db.createTask({ runId: 'run_legacy_local', spec: handle })
const dispatchId = db.createStartingWorkerDispatch({
creator: { kind: 'system' },
maxDepth: 9,
taskId: task.id,
startOptions: {}
}).dispatch.id
db.prepareStartingWorkerAuthority({
dispatchId,
handle,
paneKey: `tab_${handle}:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa`,
processIncarnation: `pty:${handle}:1`,
worktreeId,
effects: [{ kind: 'worktree', action: options.action ?? 'created_child', id: worktreeId }],
setupState: 'not_applicable',
terminalOwnership: options.terminalOwnership ?? 'created',
...(options.hostScope ? { hostScope: options.hostScope } : {})
})
db.markWorkerDispatchReady(dispatchId)
db.settleWorkerReport({ taskId: task.id, dispatchId, outcome: 'succeeded', result: 'done' })
return dispatchId
}
function releaseWorker(dispatchId: string): void {
const requested = db.requestWorkerTerminalRelease(dispatchId)
if (requested.disposition !== 'requested') {
throw new Error(`release was ${requested.disposition}`)
}
db.settleWorkerTerminalRelease(requested.resource.id)
}
async function scan(worktreeIds: string[]) {
const repo: Repo = {
id: 'repo',
path: repoPath,
displayName: 'repo',
badgeColor: '#000',
addedAt: 0
}
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the scan reads only these store fields for a local repo.
const store = {
getRepos: () => [repo],
getWorktreeMeta: (id: string) => meta[id],
getAllWorktreeMeta: () => meta,
getWorktreeLineage: (id: string) => lineage[id],
getGitHubCache: () => ({ pr: {}, issue: {} })
} as unknown as RuntimeStore
const result = await scanWorkspaceCleanup(
store,
{ worktreeIds },
{ getOrchestrationDb: () => db }
)
return new Map(
result.candidates.map((candidate) => [
candidate.worktreeId,
{ reasons: candidate.reasons, blockers: candidate.blockers }
])
)
}
beforeEach(async () => {
// Only Date: real timers keep Git subprocess timeouts working. A day after creation keeps rows active.
vi.useFakeTimers({ toFake: ['Date'] })
vi.setSystemTime(WORKSPACE_CREATED_AT + 24 * 60 * 60 * 1000)
scratchDir = await realpath(await mkdtemp(join(tmpdir(), 'orca-cleanup-workers-')))
repoPath = join(scratchDir, 'repo')
await git(['init', '-q', '-b', 'main', repoPath], scratchDir)
await git(['config', 'user.email', 'cleanup@example.invalid'], repoPath)
await git(['config', 'user.name', 'Cleanup'], repoPath)
await git(['config', 'maintenance.auto', 'false'], repoPath)
await git(['config', 'gc.auto', '0'], repoPath)
await commitFile(repoPath, 'seed.txt')
db = new OrchestrationDb(':memory:')
})
afterEach(async () => {
vi.useRealTimers()
db.close()
for (const key of Object.keys(meta)) {
delete meta[key]
}
for (const key of Object.keys(lineage)) {
delete lineage[key]
}
await removeTree(scratchDir)
})
describe('workspace cleanup for released orchestration workers', () => {
it('suggests a released worker whose work its coordinator squash-merged, and keeps the rest', async () => {
const coordinator = await addWorktree('coordinator', 'main')
const landed = await addChild('landed', coordinator.id)
const unreleased = await addChild('unreleased', coordinator.id)
const unlanded = await addChild('unlanded', coordinator.id)
await commitFile(landed.path, 'landed.txt')
await commitFile(unreleased.path, 'unreleased.txt')
await commitFile(unlanded.path, 'unlanded.txt')
// The coordinator squashes landed's and unreleased's work, never unlanded's.
await git(['merge', '-q', '--squash', 'landed'], coordinator.path)
await git(['commit', '-qm', 'squash landed'], coordinator.path)
await git(['merge', '-q', '--squash', 'unreleased'], coordinator.path)
await git(['commit', '-qm', 'squash unreleased'], coordinator.path)
releaseWorker(startWorker(landed.id, 'term_landed'))
startWorker(unreleased.id, 'term_unreleased')
releaseWorker(startWorker(unlanded.id, 'term_unlanded'))
const candidates = await scan([landed.id, unreleased.id, unlanded.id])
expect(candidates.get(landed.id)).toEqual({ reasons: ['orchestration-worker'], blockers: [] })
expect(candidates.get(unreleased.id)).toEqual({ reasons: [], blockers: ['unpushed-commits'] })
expect(candidates.get(unlanded.id)).toEqual({ reasons: [], blockers: ['unpushed-commits'] })
})
it('suggests a released worker with no commits of its own but never a workspace a worker only reused', async () => {
const coordinator = await addWorktree('coordinator', 'main')
const created = await addChild('created', coordinator.id)
const reused = await addWorktree('reused', 'main')
releaseWorker(startWorker(created.id, 'term_created'))
releaseWorker(startWorker(reused.id, 'term_reused', { action: 'reused' }))
const candidates = await scan([created.id, reused.id])
expect(candidates.get(created.id)?.reasons).toEqual(['unchanged', 'orchestration-worker'])
expect(candidates.get(reused.id)?.reasons).toEqual(['unchanged'])
})
it('suggests a worker workspace once the follow-up that took over its terminal is released', async () => {
const coordinator = await addWorktree('coordinator', 'main')
const child = await addChild('child', coordinator.id)
startWorker(child.id, 'term_child')
const followUp = startWorker(child.id, 'term_child', {
action: 'reused',
terminalOwnership: 'external'
})
expect((await scan([child.id])).get(child.id)?.reasons).toEqual(['unchanged'])
releaseWorker(followUp)
expect((await scan([child.id])).get(child.id)?.reasons).toEqual([
'unchanged',
'orchestration-worker'
])
})
it('never suggests a workspace recreated at the old path of a worker that a later worker only reused', async () => {
const coordinator = await addWorktree('coordinator', 'main')
const child = await addChild('child', coordinator.id)
const first = startWorker(child.id, 'term_child')
releaseWorker(first)
// The first worker ran before the path was reused.
db.db
.prepare(
"UPDATE worker_terminal_resources SET created_at = '2025-01-01 00:00:00' WHERE origin_dispatch_id = ?"
)
.run(first)
// The user removes the worker's workspace and makes a new one at the same path.
await git(['worktree', 'remove', child.path], repoPath)
await git(['branch', '-D', 'child'], repoPath)
await addChild('child', coordinator.id)
releaseWorker(startWorker(child.id, 'term_later', { action: 'reused' }))
expect((await scan([child.id])).get(child.id)?.reasons).toEqual(['unchanged'])
})
it('keeps a released worker workspace while another worker is still starting in it', async () => {
const coordinator = await addWorktree('coordinator', 'main')
const child = await addChild('child', coordinator.id)
releaseWorker(startWorker(child.id, 'term_child'))
const retry = db.createTask({ runId: 'run_legacy_local', spec: 'retry' })
db.createStartingWorkerDispatch({
creator: { kind: 'system' },
maxDepth: 9,
taskId: retry.id,
startOptions: { worktree: `id:${child.id}`, resolvedWorktreeId: child.id }
})
expect((await scan([child.id])).get(child.id)?.reasons).toEqual(['unchanged'])
})
it('never suggests a workspace on this machine for an SSH worker that used the same id', async () => {
const coordinator = await addWorktree('coordinator', 'main')
const child = await addChild('child', coordinator.id)
releaseWorker(
startWorker(child.id, 'term_ssh', {
hostScope: JSON.stringify({ kind: 'ssh', targetId: 'ssh-target' })
})
)
expect((await scan([child.id])).get(child.id)?.reasons).toEqual(['unchanged'])
})
})
@@ -0,0 +1,74 @@
import type { Worktree } from '../../../shared/worktree/types'
import { isValidResolvedWorktreeLineageEdge } from '../../../shared/resolved-worktree-lineage'
import { normalizeLocalBranchName } from '../../runtime/workspaces/catalog/runtime-worktree-selection'
import type { RuntimeStore } from '../../runtime/construction/runtime-store-contract'
import type { OrchestrationDb } from '../../runtime/orchestration/db/orchestration-db'
import {
listWorkerCreatedWorktrees,
type WorkerCreatedWorktree
} from '../../runtime/orchestration/db/worker-dispatch/worker-created-worktrees'
/** A workspace a worker start created whose worker has been released. */
export type WorkspaceCleanupReleasedWorker = {
/** The coordinator's branch, when the coordinator's workspace is in the same repo; work landed there is not lost. */
coordinatorBranch: string | null
}
export type WorkspaceCleanupReleasedWorkerLookup = (
worktree: Worktree,
repoWorktrees: readonly Worktree[]
) => WorkspaceCleanupReleasedWorker | undefined
/** Reads the host's workers at most once per scan; a host that never ran orchestration has none. */
export function createWorkspaceCleanupReleasedWorkerLookup(
store: RuntimeStore,
getOrchestrationDb: (() => OrchestrationDb | null) | undefined
): WorkspaceCleanupReleasedWorkerLookup {
let workerCreatedWorktrees: ReadonlyMap<string, WorkerCreatedWorktree> | undefined
return (worktree, repoWorktrees) => {
workerCreatedWorktrees ??= readWorkerCreatedWorktrees(getOrchestrationDb)
return resolveReleasedWorker(worktree, repoWorktrees, workerCreatedWorktrees, store)
}
}
function readWorkerCreatedWorktrees(
getOrchestrationDb: (() => OrchestrationDb | null) | undefined
): ReadonlyMap<string, WorkerCreatedWorktree> {
try {
const db = getOrchestrationDb?.()
return db ? listWorkerCreatedWorktrees(db) : new Map()
} catch (error) {
console.warn('Workspace cleanup could not read orchestration workers', error)
return new Map()
}
}
function resolveReleasedWorker(
worktree: Worktree,
repoWorktrees: readonly Worktree[],
workerCreatedWorktrees: ReadonlyMap<string, WorkerCreatedWorktree>,
store: RuntimeStore
): WorkspaceCleanupReleasedWorker | undefined {
const worker = workerCreatedWorktrees.get(worktree.id)
// SQLite stamps whole seconds. A workspace made after the worker started reuses the path; it is not the worker's.
if (
!worker?.released ||
worktree.createdAt === undefined ||
worktree.createdAt >= worker.workerStartedAt + 1000
) {
return undefined
}
const lineage = store.getWorktreeLineage?.(worktree.id)
const coordinator = lineage
? repoWorktrees.find((candidate) =>
isValidResolvedWorktreeLineageEdge(worktree, candidate, lineage)
)
: undefined
const coordinatorBranch = normalizeLocalBranchName(coordinator?.branch)
return {
coordinatorBranch:
coordinatorBranch && coordinatorBranch !== normalizeLocalBranchName(worktree.branch)
? coordinatorBranch
: null
}
}
@@ -4,12 +4,15 @@ import type {
WorkspaceCleanupScanProgress
} from '../../../shared/workspace-cleanup/workspace-cleanup'
import { appendWorkspaceCleanupItems } from './workspace-cleanup-scan-primitives'
import type { OrchestrationDb } from '../../runtime/orchestration/db/orchestration-db'
const WORKSPACE_CLEANUP_PROGRESS_EMIT_INTERVAL_MS = 100
export type WorkspaceCleanupScanOptions = {
onProgress?: (progress: WorkspaceCleanupScanProgress) => void
signal?: AbortSignal
/** The scanning host's orchestration database, so released workers' workspaces can be classified. */
getOrchestrationDb?: () => OrchestrationDb | null
}
export type WorkspaceCleanupProgressEmitter = {
@@ -4,7 +4,7 @@ import type { PRInfo } from '../../../shared/github/pull-requests/pull-request-t
import type { PRCacheView } from '../../../shared/sidebar-list/workspace-review-facts'
import { readWorktreeMetaForHost } from '../../persistence/host-qualified-worktree-meta'
import type { Repo } from '../../../shared/repo-types'
import type { GitWorktreeInfo, Worktree } from '../../../shared/worktree/types'
import type { GitWorktreeInfo } from '../../../shared/worktree/types'
import { mergeWorktree } from '../worktree-metadata-merge'
import type {
WorkspaceCleanupAutomationRunLookup,
@@ -21,7 +21,7 @@ import type { WorkspaceCleanupGitRoute } from './workspace-cleanup-git-route'
import { shouldScanBroadWorkspaceCleanupWorktree } from './workspace-cleanup-scan-eligibility'
import {
resolvePersistedWorkspaceCleanupActivityWorktree,
resolveWorkspaceCleanupActivityWorktree,
resolveWorkspaceCleanupActivityWithTimeout,
type WorkspaceCleanupFsActivityCache
} from './workspace-cleanup-activity'
import {
@@ -32,12 +32,10 @@ import {
} from './workspace-cleanup-candidate'
import { synthesizeDisconnectedSshCleanupCandidates } from './workspace-cleanup-disconnected-ssh'
import {
WORKSPACE_CLEANUP_GIT_READ_TIMEOUT_MS,
WorkspaceCleanupScanCancelledError,
appendWorkspaceCleanupItems,
mapWorkspaceCleanupWithConcurrency,
throwIfWorkspaceCleanupScanAborted,
withWorkspaceCleanupTimeout
throwIfWorkspaceCleanupScanAborted
} from './workspace-cleanup-scan-primitives'
import { listWorkspaceCleanupFolderWorkspaces } from './workspace-cleanup-folder-workspaces'
import {
@@ -50,6 +48,10 @@ import {
} from './workspace-cleanup-scan-targets'
import { isWorktreeMetaOwnedByRepo } from '../../workspaces/catalog/worktree-metadata-ownership'
import { runtimeVisibleRepos } from '../../runtime/workspaces/catalog/runtime-visible-catalog'
import {
createWorkspaceCleanupReleasedWorkerLookup,
type WorkspaceCleanupReleasedWorkerLookup
} from './workspace-cleanup-orchestration-workers'
const WORKTREE_SCAN_CONCURRENCY = 3
// Overlap repo scans to avoid summing SSH listing round-trip latency.
@@ -86,6 +88,10 @@ export async function scanWorkspaceCleanup(
// One snapshot per scan: merged-review evidence reads it without the network.
const prCache = store.getGitHubCache().pr
const findAutomationRun = createWorkspaceCleanupAutomationRunLookup(store)
const findReleasedWorker = createWorkspaceCleanupReleasedWorkerLookup(
store,
options.getOrchestrationDb
)
const progress = createWorkspaceCleanupProgressEmitter(args.scanId, scannedAt, options)
const errors: WorkspaceCleanupScanResult['errors'] = []
const candidates: WorkspaceCleanupCandidate[] = []
@@ -103,6 +109,7 @@ export async function scanWorkspaceCleanup(
scannedAt,
prCache,
findAutomationRun,
findReleasedWorker,
targetWorktreeIds: targetWorktreeIdsByRepo.get(repo.id),
refreshTargetActivity: args.worktreeId !== undefined || args.refreshActivity === true,
includeAllWorkspaces: args.includeAllWorkspaces === true,
@@ -132,6 +139,7 @@ async function scanRepoWorkspaces(
scannedAt: number
prCache: PRCacheView<PRInfo>
findAutomationRun: WorkspaceCleanupAutomationRunLookup
findReleasedWorker: WorkspaceCleanupReleasedWorkerLookup
targetWorktreeIds?: ReadonlySet<string>
refreshTargetActivity: boolean
includeAllWorkspaces: boolean
@@ -146,6 +154,7 @@ async function scanRepoWorkspaces(
scannedAt,
prCache,
findAutomationRun,
findReleasedWorker,
targetWorktreeIds,
refreshTargetActivity,
includeAllWorkspaces,
@@ -243,7 +252,7 @@ async function scanRepoWorkspaces(
activityStatsUnavailable ||
(targetWorktreeIds ? !refreshTargetActivity : persistedActivityIsRecent)
? persistedActivityWorktree
: await resolveCleanupActivityWithTimeout(
: await resolveWorkspaceCleanupActivityWithTimeout(
route,
worktree,
() => {
@@ -269,6 +278,7 @@ async function scanRepoWorkspaces(
skipGit: skipGitWorktreeIds.has(worktreeWithActivity.id) || !isInactive,
forceGitCheck: Boolean(targetWorktreeIds),
findAutomationRun,
releasedWorker: findReleasedWorker(worktreeWithActivity, mergedWorktrees),
signal
}).catch((error) => {
if (error instanceof WorkspaceCleanupScanCancelledError) {
@@ -287,34 +297,3 @@ async function scanRepoWorkspaces(
return { scannedAt, candidates, errors }
}
async function resolveCleanupActivityWithTimeout(
route: WorkspaceCleanupGitRoute,
worktree: Worktree,
onActivityStatsUnavailable: () => void,
signal?: AbortSignal,
fsActivityCache?: WorkspaceCleanupFsActivityCache
): Promise<Worktree> {
try {
return await withWorkspaceCleanupTimeout(
() =>
resolveWorkspaceCleanupActivityWorktree(
route,
worktree,
undefined,
undefined,
fsActivityCache
),
WORKSPACE_CLEANUP_GIT_READ_TIMEOUT_MS,
'Timed out reading worktree activity.',
signal
)
} catch (error) {
if (error instanceof WorkspaceCleanupScanCancelledError) {
throw error
}
onActivityStatsUnavailable()
console.warn('Workspace cleanup activity scan failed', error)
return resolvePersistedWorkspaceCleanupActivityWorktree(worktree)
}
}
@@ -1,5 +1,6 @@
import { ipcMain } from 'electron'
import type { Store } from '../../persistence/loading-store/store'
import type { OrchestrationDb } from '../../runtime/orchestration/db/orchestration-db'
import {
WORKSPACE_CLEANUP_CLASSIFIER_VERSION,
type WorkspaceCleanupDismissArgs,
@@ -32,7 +33,10 @@ function getBroadScanModeKey(senderId: number, args: WorkspaceCleanupScanArgs):
return `${senderId}\0${args.includeAllWorkspaces === true}`
}
export function registerWorkspaceCleanupHandlers(store: Store): void {
export function registerWorkspaceCleanupHandlers(
store: Store,
getOrchestrationDb?: () => OrchestrationDb | null
): void {
const snapshotDirectory = store.getProfileStorageDirectory()
ipcMain.removeHandler('workspaceCleanup:scan')
ipcMain.removeHandler('workspaceCleanup:cancelScan')
@@ -66,6 +70,7 @@ export function registerWorkspaceCleanupHandlers(store: Store): void {
try {
const result = await scanWorkspaceCleanup(store, scanArgs, {
signal: controller.signal,
getOrchestrationDb,
onProgress: scanArgs.scanId
? (progress) => {
if (!controller.signal.aborted && !sender.isDestroyed()) {
@@ -0,0 +1,144 @@
import {
WORKER_SETTLED_STATES,
type WorkerTerminalReleaseState
} from '../../terminal-authority/release/worker-terminal-ownership'
import type { WorkerDispatchState } from '../orchestration-records'
import { dispatchCompletionEpoch } from '../dispatch-context/dispatch-completion-epoch'
import type { OrchestrationDb } from '../orchestration-db'
import { parseWorkerTerminalPriorOwnerIds } from '../pane-key-match'
import { readWorkerTerminalHostScope } from '../../../../../shared/orchestration/workers/worker-terminal-host-scope'
export type WorkerCreatedWorktree = {
/** When the newest creating worker's terminal started; a workspace made later at that path is not the worker's. */
workerStartedAt: number
/** Every dispatch that used the workspace has released its terminal or settled without one. */
released: boolean
}
type WorkerWorktreeUseRow = {
dispatch_id: string
worktree_id: string
state: WorkerDispatchState
effects: string
terminal_handle: string | null
}
type WorkerTerminalRow = {
origin_dispatch_id: string
owner_dispatch_id: string
prior_owner_dispatch_ids: string
created_at: string
host_scope: string | null
worktree_id: string | null
release_state: WorkerTerminalReleaseState
}
/** Workspaces a worker start created, keyed by worktree id. Reused workspaces are never listed. */
export function listWorkerCreatedWorktrees(
db: OrchestrationDb
): Map<string, WorkerCreatedWorktree> {
// Local workers live in worker_dispatches; on a federated worker's host they live in remote_dispatch_attachments.
// A worker still starting in an existing workspace has only its resolved target, not worktree_id yet.
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the query selects exactly these columns.
const rows = db.db
.prepare(
`SELECT * FROM (
SELECT dispatch_id, state, effects, agent_terminal_handle AS terminal_handle,
COALESCE(worktree_id, CASE WHEN json_valid(start_options)
THEN json_extract(start_options, '$.resolvedWorktreeId') END) AS worktree_id
FROM worker_dispatches
) WHERE worktree_id IS NOT NULL
UNION ALL
SELECT dispatch_id, state, effects, terminal_handle, worktree_id
FROM remote_dispatch_attachments WHERE worktree_id IS NOT NULL`
)
.all() as WorkerWorktreeUseRow[]
const terminals = readWorkerTerminals(db)
const uses = new Map<string, { startedAt: number | null; held: boolean }>()
for (const row of rows) {
const use = uses.get(row.worktree_id) ?? { startedAt: null, held: false }
const startedAt = terminals.startedAtByOrigin.get(row.dispatch_id)
if (startedAt !== undefined && createdWorktree(row.effects, row.worktree_id)) {
use.startedAt = Math.max(use.startedAt ?? 0, startedAt)
}
// A terminal row speaks for its owner; a dispatch whose terminal moved on, or never existed, holds nothing once settled.
use.held ||=
!terminals.owners.has(row.dispatch_id) &&
((row.terminal_handle !== null && !terminals.handedOff.has(row.dispatch_id)) ||
!WORKER_SETTLED_STATES.includes(row.state))
uses.set(row.worktree_id, use)
}
const created = new Map<string, WorkerCreatedWorktree>()
for (const [worktreeId, use] of uses) {
if (use.startedAt !== null) {
created.set(worktreeId, {
workerStartedAt: use.startedAt,
released:
!use.held &&
terminals.releasedWorktreeIds.has(worktreeId) &&
!terminals.heldWorktreeIds.has(worktreeId)
})
}
}
return created
}
function readWorkerTerminals(db: OrchestrationDb): {
startedAtByOrigin: Map<string, number>
owners: Set<string>
handedOff: Set<string>
releasedWorktreeIds: Set<string>
heldWorktreeIds: Set<string>
} {
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the query selects exactly these columns.
const rows = db.db
.prepare(
`SELECT origin_dispatch_id, owner_dispatch_id, prior_owner_dispatch_ids, created_at,
host_scope, worktree_id, release_state
FROM worker_terminal_resources`
)
.all() as WorkerTerminalRow[]
const terminals = {
startedAtByOrigin: new Map<string, number>(),
owners: new Set<string>(),
handedOff: new Set<string>(),
releasedWorktreeIds: new Set<string>(),
heldWorktreeIds: new Set<string>()
}
for (const row of rows) {
const startedAt = dispatchCompletionEpoch(row.created_at)
// An SSH worker's workspace can share its id with one on this machine; only this machine's workers count.
const hostScope = readWorkerTerminalHostScope(row.host_scope).kind
if (startedAt !== null && (hostScope === 'absent' || hostScope === 'local')) {
terminals.startedAtByOrigin.set(row.origin_dispatch_id, startedAt)
}
terminals.owners.add(row.owner_dispatch_id)
for (const dispatchId of parseWorkerTerminalPriorOwnerIds(row.prior_owner_dispatch_ids) ?? []) {
terminals.handedOff.add(dispatchId)
}
// Includes a terminal placed in the workspace before its dispatch recorded the workspace.
if (row.worktree_id !== null && row.release_state === 'released') {
terminals.releasedWorktreeIds.add(row.worktree_id)
} else if (row.worktree_id !== null) {
terminals.heldWorktreeIds.add(row.worktree_id)
}
}
return terminals
}
function createdWorktree(effectsJson: string, worktreeId: string): boolean {
try {
const effects: unknown = JSON.parse(effectsJson)
return (
Array.isArray(effects) &&
effects.some(
(effect) =>
effect?.kind === 'worktree' &&
effect.id === worktreeId &&
(effect.action === 'created_child' || effect.action === 'created_top_level')
)
)
} catch {
return false
}
}
@@ -8,5 +8,8 @@ export const WORKSPACE_CLEANUP_SCAN_METHOD = defineMethod({
permission: 'workspace',
params: WorkspaceCleanupScanParams,
handler: (params, { runtime, signal }) =>
scanWorkspaceCleanup(runtime.requireStore(), params, { signal })
scanWorkspaceCleanup(runtime.requireStore(), params, {
signal,
getOrchestrationDb: () => runtime.getExistingOrchestrationDb()
})
})
@@ -92,7 +92,7 @@ export function attachMainWindowServices(
scheduleWorktreeBaseDirectoryWatcherSync(store, mainWindow)
// Folder projects have no watcher target; poll for external git init so they upgrade without restart (#11477).
startFolderRepoGitUpgradeWatch(store, mainWindow)
registerWorkspaceCleanupHandlers(store)
registerWorkspaceCleanupHandlers(store, () => runtime.getExistingOrchestrationDb())
registerPtyHandlers(
mainWindow,
runtime,
@@ -26,7 +26,11 @@ const GIT_DERIVED_BLOCKERS: ReadonlySet<WorkspaceCleanupBlocker> = new Set([
'unknown-base'
])
const GIT_DERIVED_REASONS: ReadonlySet<WorkspaceCleanupReason> = new Set(['merged', 'unchanged'])
const GIT_DERIVED_REASONS: ReadonlySet<WorkspaceCleanupReason> = new Set([
'merged',
'unchanged',
'orchestration-worker'
])
export function needsWorkspaceCleanupGitEvidence(
filters: WorkspaceCleanupFilterState,
@@ -20,6 +20,7 @@ export type WorkspaceCleanupReason =
| 'merged'
| 'unchanged'
| 'automation-run'
| 'orchestration-worker'
/** What HEAD holds beyond the workspace's base ref (worktree base, else project base, else repo default). */
export type WorkspaceCleanupMergeEvidence = {
@@ -33,6 +34,8 @@ export type WorkspaceCleanupMergeEvidence = {
headInDefaultBranch: boolean
/** Commits on HEAD the base ref cannot reach. */
aheadOfDefault: number
/** Set only for a released orchestration worker's workspace: every change on HEAD is in its coordinator's branch. */
headInCoordinatorBranch?: boolean
}
export type WorkspaceCleanupInactivityInput = {
@@ -240,15 +243,26 @@ export function getWorkspaceCleanupInactivityReasons(
return reasons
}
/** HEAD's commits already live in the base or in a merged review, so removing it loses no commits. */
/** HEAD's commits already live in the base, a merged review or the coordinator's branch, so removing it loses no commits. */
export function isWorkspaceCleanupHeadLanded(
merge: WorkspaceCleanupMergeEvidence | undefined
): boolean {
return (
merge !== undefined && (merge.headInDefaultBranch || merge.mergedReview?.containsHead === true)
merge !== undefined &&
(merge.headInDefaultBranch ||
merge.mergedReview?.containsHead === true ||
merge.headInCoordinatorBranch === true)
)
}
/** A workspace a worker start created, whose worker was released, and whose work already landed. */
export function getWorkspaceCleanupOrchestrationWorkerReasons(
releasedWorker: boolean,
merge: WorkspaceCleanupMergeEvidence | undefined
): WorkspaceCleanupReason[] {
return releasedWorker && isWorkspaceCleanupHeadLanded(merge) ? ['orchestration-worker'] : []
}
export function getWorkspaceCleanupMergeReasons(
merge: WorkspaceCleanupMergeEvidence | undefined
): WorkspaceCleanupReason[] {