diff --git a/src/main/ipc/worktrees-create-registry.test.ts b/src/main/ipc/worktrees-create-registry.test.ts index 2f16fe69825..bb54fc84cff 100644 --- a/src/main/ipc/worktrees-create-registry.test.ts +++ b/src/main/ipc/worktrees-create-registry.test.ts @@ -87,10 +87,10 @@ vi.mock('./pty/provider/state-cleanup', async () => ) function creationsOf(stub: WorktreeRuntimeStub): Record { - if (!stub.workspaceCreations) { + if (!stub.workspaceLifecycle) { throw new Error('the switch is mocked on') } - return stub.workspaceCreations.snapshot().creations + return stub.workspaceLifecycle.creationsSnapshot().creations } describe('worktree create registry', () => { diff --git a/src/main/ipc/worktrees-test-runtime-stub.ts b/src/main/ipc/worktrees-test-runtime-stub.ts index fe751b0e422..b186c96054f 100644 --- a/src/main/ipc/worktrees-test-runtime-stub.ts +++ b/src/main/ipc/worktrees-test-runtime-stub.ts @@ -1,6 +1,6 @@ import { _resetPendingWorktreeRemovalsForTests } from '../worktree-background-removal' import { vi } from 'vitest' -import { WorkspaceCreationRegistry } from '../runtime/sidebar-inputs/workspace-creation-registry' +import { WorkspaceLifecycleQueue } from '../runtime/workspaces/workspace-lifecycle-queue' import { ORCA_SIDEBAR_INPUTS_IS_MAIN } from '../../shared/sidebar-inputs-switch' export type WorktreeRuntimeStub = { @@ -23,7 +23,7 @@ export type WorktreeRuntimeStub = { acquireFileWatcherRemoval: ReturnType hydrateInferredWorktreeLineage: ReturnType publishWorktreeRemovalChange: ReturnType - workspaceCreations: WorkspaceCreationRegistry | null + workspaceLifecycle: WorkspaceLifecycleQueue | null } /** Why: create-flow tests need a minimal runtime; full fetchRemoteWithCache behavior lives in fetch-remote-cache.test.ts. */ @@ -34,7 +34,7 @@ export function createWorktreeRuntimeStub(mainWindow?: { _resetPendingWorktreeRemovalsForTests() const runtimeStub: WorktreeRuntimeStub = { // Mirrors the runtime field: listed only when the sidebar-inputs switch is on. - workspaceCreations: ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceCreationRegistry() : null, + workspaceLifecycle: ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceLifecycleQueue() : null, resolveRemoteTrackingBase: vi.fn().mockResolvedValue(null), hasRemoteTrackingRef: vi.fn().mockResolvedValue(false), getOrStartRemoteTrackingBaseRefresh: vi.fn().mockResolvedValue({ ok: true }), diff --git a/src/main/ipc/worktrees/create/register-worktree-create-handlers.ts b/src/main/ipc/worktrees/create/register-worktree-create-handlers.ts index 737ab10d229..9a6d9629c0f 100644 --- a/src/main/ipc/worktrees/create/register-worktree-create-handlers.ts +++ b/src/main/ipc/worktrees/create/register-worktree-create-handlers.ts @@ -28,6 +28,8 @@ import type { CreateWorktreeArgsWithSystemProvenance } from '../ipc-context-sche import { createFolderWorkspace } from './folder-workspace-creation' import { findExactRepoOwner, isCapturedRepoCurrent } from '../listing/worktree-host-ownership' import { requireWorktreeCreateRoute } from '../../../worktree-create-execution-host-route' +import { getRepoExecutionHostId } from '../../../../shared/execution-host' +import { workspaceLifecycleKey } from '../../../runtime/workspaces/workspace-lifecycle-commands' import type { WorktreeIpcContext } from '../worktree-ipc-context' import { beginWorkspaceCreateTelemetry } from '../../../workspace-create-telemetry' import { assertWslMoveRepoWritable } from '../../../persistence/moving-wsl-projects/wsl-move-scope-guard' @@ -67,9 +69,10 @@ export function registerWorktreeCreateHandlers(context: WorktreeIpcContext): voi fromExistingBranch: typeof args.baseBranch === 'string' && args.baseBranch.length > 0, isFolder: isFolderRepo(repo) }) - const creation = runtime.workspaceCreations?.open({ + const creation = runtime.workspaceLifecycle?.openCreate({ creationId: args.creationId, - repoId: repo.id + repoId: repo.id, + keyOf: (worktreeId) => workspaceLifecycleKey(getRepoExecutionHostId(repo), worktreeId) }) try { // Why: wrap only the helpers; the pre-validation throws above are IPC-shape bugs, not the git/filesystem failures the funnel tracks. diff --git a/src/main/ipc/worktrees/removal/register-worktree-removal-handlers.ts b/src/main/ipc/worktrees/removal/register-worktree-removal-handlers.ts index 9a2323d6277..dee7de87ce5 100644 --- a/src/main/ipc/worktrees/removal/register-worktree-removal-handlers.ts +++ b/src/main/ipc/worktrees/removal/register-worktree-removal-handlers.ts @@ -23,6 +23,12 @@ import { } from '../../../worktree-background-removal' import { runSerializedWorktreeRemovalAcceptance } from '../../../worktree-removal-acceptance-queue' import { assertWslMoveRepoWritable } from '../../../persistence/moving-wsl-projects/wsl-move-scope-guard' +import { + deleteLostContact, + observeTerminals, + runWorkspaceLifecycleCommand, + workspaceLifecycleKey +} from '../../../runtime/workspaces/workspace-lifecycle-commands' export function registerWorktreeRemovalHandlers(context: WorktreeIpcContext): void { const { store, options, worktreeRemovalsInFlight } = context @@ -129,5 +135,29 @@ export function registerWorktreeRemovalHandlers(context: WorktreeIpcContext): vo } } } - ipcMain.handle('worktrees:remove', (_event, args: RemoveWorktreeArgs) => remove(args)) + ipcMain.handle('worktrees:remove', (_event, args: RemoveWorktreeArgs) => { + const repo = resolveRepoForExecutionHost( + store, + parseWorktreeId(args.worktreeId).repoId, + args.hostId + ) + // Without its repo the removal refuses as before; there is no workspace to queue under. + return repo + ? runWorkspaceLifecycleCommand( + context.runtime.workspaceLifecycle, + workspaceLifecycleKey(getRepoExecutionHostId(repo), args.worktreeId), + { + kind: 'delete', + run: () => remove(args), + lostContact: deleteLostContact, + observe: observeTerminals(context.runtime, `id:${args.worktreeId}`), + delete: { + key: getWorktreeRemovalOptionsKey(args), + force: args.allowUnverifiedPtyStop === true, + label: args.worktreeId + } + } + ) + : remove(args) + }) } diff --git a/src/main/runtime/managed-worktree-create-registry.test.ts b/src/main/runtime/managed-worktree-create-registry.test.ts index 2bcc8bdc5f3..cce1a24e6d2 100644 --- a/src/main/runtime/managed-worktree-create-registry.test.ts +++ b/src/main/runtime/managed-worktree-create-registry.test.ts @@ -95,10 +95,10 @@ function makeRuntime(repo: Record = gitRepo): OrcaRuntimeServic type LocalCreateArgs = { onWorktreeMetadataPersisted: (created: { id: string }) => unknown } function creationsOf(runtime: OrcaRuntimeService): Record { - if (!runtime.workspaceCreations) { + if (!runtime.workspaceLifecycle) { throw new Error('the switch is mocked on') } - return runtime.workspaceCreations.snapshot().creations + return runtime.workspaceLifecycle.creationsSnapshot().creations } describe('runtime creation registry', () => { diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index d54943103d9..3a6be414d02 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -99,6 +99,7 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase { scan: () => this.scanResolvedWorktreeRepos(), join: (scans) => this.joinResolvedWorktreeScans(scans), visible: (snapshot) => this.visibleResolvedWorktreeSnapshot(snapshot), + lifecycle: this.workspaceLifecycle, onClientEvent: (listener, options) => this.onClientEvent(listener, options) }) : null @@ -109,7 +110,7 @@ class OrcaRuntimeService extends OrcaRuntimeSidebarBase { sidebarInputsFeedDomains({ facts: requireSwitchedOn(this.sidebarFacts), pairedServerRows: requireSwitchedOn(this.pairedServerRows), - workspaceCreations: requireSwitchedOn(this.workspaceCreations), + workspaceLifecycle: requireSwitchedOn(this.workspaceLifecycle), workspaceRows: requireSwitchedOn(this.workspaceRows), repoIcons: requireSwitchedOn(this.repoIconDataUrls), store: () => this.store, diff --git a/src/main/runtime/rpc/methods/sidebar-inputs.ts b/src/main/runtime/rpc/methods/sidebar-inputs.ts index 089db300a71..258da319750 100644 --- a/src/main/runtime/rpc/methods/sidebar-inputs.ts +++ b/src/main/runtime/rpc/methods/sidebar-inputs.ts @@ -109,7 +109,7 @@ export const SIDEBAR_INPUT_METHODS = [ permission: 'workspace', params: null, handler: async (_params, { runtime }) => - requireSwitchedOn(runtime.workspaceCreations).snapshot() + requireSwitchedOn(runtime.workspaceLifecycle).creationsSnapshot() }), defineMethod({ name: 'sidebarInputs.reportView', diff --git a/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts b/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts index a9d15b8a565..00cc719d9a0 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-lifecycle-methods.ts @@ -1,4 +1,5 @@ import { terminalImageViewerSupported } from '../../../terminal-image-viewer-env' +import { TERMINAL_WORKSPACE_STOP_METHODS } from './terminal-workspace-stop-methods' import { ORCA_SIDEBAR_INPUTS_IS_MAIN } from '../../../../../shared/sidebar-inputs-switch' import { defineMethod } from '../../core' import { @@ -9,15 +10,11 @@ import { withTerminalCloseAttribution } from '../../terminal-close-attribution' import { AgentTeamsPrepareLaunch, AgentTeamsTmuxCompat, - TerminalCloseAll, TerminalCreateParams, TerminalFocus, TerminalHandle, TerminalReveal, - TerminalSleep, TerminalSplit, - TerminalStop, - TerminalStopExact, TerminalWait } from '../../../../../shared/rpc-contract/terminal-unary-params' import { TerminalResizeForClient } from '../../../../../shared/rpc-contract/terminal-stream-params' @@ -161,34 +158,7 @@ export const TERMINAL_LIFECYCLE_METHODS = [ }) }) }), - defineMethod({ - name: 'terminal.stop', - permission: 'workspace', - params: TerminalStop, - handler: async (params, { runtime }) => runtime.stopTerminalsForWorktree(params.worktree) - }), - defineMethod({ - name: 'terminal.closeAll', - permission: 'workspace', - params: TerminalCloseAll, - handler: async (params, { runtime }) => runtime.closeTerminalsForWorktree(params.worktree) - }), - defineMethod({ - name: 'terminal.sleep', - permission: 'workspace', - params: TerminalSleep, - handler: async (params, { runtime }) => runtime.sleepTerminalsForWorktree(params.worktree) - }), - defineMethod({ - name: 'terminal.stopExact', - permission: 'workspace', - params: TerminalStopExact, - handler: async (params, { runtime }) => - runtime.stopExactTerminalsForWorktree(params.worktree, params.expectedPtyIds, { - keepHistory: params.keepHistory, - targetOnly: params.targetOnly - }) - }), + ...TERMINAL_WORKSPACE_STOP_METHODS, defineMethod({ name: 'terminal.resizeForClient', permission: 'workspace', diff --git a/src/main/runtime/rpc/methods/terminal/terminal-workspace-stop-methods.ts b/src/main/runtime/rpc/methods/terminal/terminal-workspace-stop-methods.ts new file mode 100644 index 00000000000..a12e55ffa31 --- /dev/null +++ b/src/main/runtime/rpc/methods/terminal/terminal-workspace-stop-methods.ts @@ -0,0 +1,60 @@ +import { + closeOutcome, + observeTerminals, + queueTerminalWorkspaceCommand, + sleepOutcome, + terminalLostContact +} from '../../../workspaces/workspace-lifecycle-commands' +import { defineMethod } from '../../core' +import { + TerminalCloseAll, + TerminalSleep, + TerminalStop, + TerminalStopExact +} from '../../../../../shared/rpc-contract/terminal-unary-params' + +/** Stopping every terminal of a workspace; closing and sleeping them wait in its lifecycle queue. */ +export const TERMINAL_WORKSPACE_STOP_METHODS = [ + defineMethod({ + name: 'terminal.stop', + permission: 'workspace', + params: TerminalStop, + handler: async (params, { runtime }) => runtime.stopTerminalsForWorktree(params.worktree) + }), + defineMethod({ + name: 'terminal.closeAll', + permission: 'workspace', + params: TerminalCloseAll, + handler: async (params, { runtime }) => + queueTerminalWorkspaceCommand(runtime, params.worktree, { + kind: 'close', + run: () => runtime.closeTerminalsForWorktree(params.worktree), + outcome: closeOutcome, + lostContact: terminalLostContact, + observe: observeTerminals(runtime, params.worktree) + }) + }), + defineMethod({ + name: 'terminal.sleep', + permission: 'workspace', + params: TerminalSleep, + handler: async (params, { runtime }) => + queueTerminalWorkspaceCommand(runtime, params.worktree, { + kind: 'sleep', + run: () => runtime.sleepTerminalsForWorktree(params.worktree), + outcome: sleepOutcome, + lostContact: terminalLostContact, + observe: observeTerminals(runtime, params.worktree) + }) + }), + defineMethod({ + name: 'terminal.stopExact', + permission: 'workspace', + params: TerminalStopExact, + handler: async (params, { runtime }) => + runtime.stopExactTerminalsForWorktree(params.worktree, params.expectedPtyIds, { + keepHistory: params.keepHistory, + targetOnly: params.targetOnly + }) + }) +] as const diff --git a/src/main/runtime/rpc/methods/worktree-sleep-method.ts b/src/main/runtime/rpc/methods/worktree-sleep-method.ts new file mode 100644 index 00000000000..7373e4a5ba7 --- /dev/null +++ b/src/main/runtime/rpc/methods/worktree-sleep-method.ts @@ -0,0 +1,10 @@ +import { WorktreeSelector } from '../../../../shared/rpc-contract/worktree-params' +import { queueWorktreeSleep } from '../../workspaces/workspace-lifecycle-commands' +import { defineMethod } from '../core' + +export const WORKTREE_SLEEP_METHOD = defineMethod({ + name: 'worktree.sleep', + permission: 'workspace', + params: WorktreeSelector, + handler: async (params, { runtime }) => queueWorktreeSleep(runtime, params.worktree) +}) diff --git a/src/main/runtime/rpc/methods/worktree.ts b/src/main/runtime/rpc/methods/worktree.ts index 7f48b08048b..8a21f6a15cb 100644 --- a/src/main/runtime/rpc/methods/worktree.ts +++ b/src/main/runtime/rpc/methods/worktree.ts @@ -33,6 +33,8 @@ import { import { WORKTREE_CATALOG_METHODS } from './worktree-catalog-methods' import { WORKTREE_PROVISIONED_ROOT_METHOD } from './worktree-provisioned-root-method' import { readsWorktreeRemovalMarker } from '../worktree-removal-marker-projection' +import { queueWorkspaceDelete } from '../../workspaces/workspace-lifecycle-commands' +import { WORKTREE_SLEEP_METHOD } from './worktree-sleep-method' export const WORKTREE_METHODS = [ ...WORKTREE_CATALOG_METHODS, @@ -64,12 +66,7 @@ export const WORKTREE_METHODS = [ worktree: await runtime.showManagedWorktree(params.worktree) }) }), - defineMethod({ - name: 'worktree.sleep', - permission: 'workspace', - params: WorktreeSelector, - handler: async (params, { runtime }) => runtime.sleepManagedWorktree(params.worktree) - }), + WORKTREE_SLEEP_METHOD, defineMethod({ name: 'worktree.activate', permission: 'workspace', @@ -280,16 +277,19 @@ export const WORKTREE_METHODS = [ // resolution costs a scan and throws for an id two hosts share. Other selectors stay unstamped. const explicitWorktreeId = getExplicitWorktreeIdSelector(params.worktree) const repoId = explicitWorktreeId ? splitWorktreeId(explicitWorktreeId)?.repoId : undefined - const result = await runtime.removeManagedWorktree(params.worktree, { - force: params.force === true, - runHooks: params.runHooks === true, - allowUnverifiedPtyStop: params.allowUnverifiedPtyStop === true, - allowFailedArchiveHook: params.allowFailedArchiveHook === true, - ...(resolvedHostId ? { hostId: resolvedHostId } : {}), - // Why: only a client that shows the `removing` marker can wait out Git's delete; older - // clients get the acceptance and never see the row again. - ...(readsWorktreeRemovalMarker(context) ? { waitForBackgroundRemoval: true } : {}) - }) + const result = await queueWorkspaceDelete( + runtime, + params.worktree, + resolvedHostId, + params, + (removal) => + runtime.removeManagedWorktree(params.worktree, { + ...removal, + // Why: only a client that shows the `removing` marker can wait out Git's delete; older + // clients get the acceptance and never see the row again. + ...(readsWorktreeRemovalMarker(context) ? { waitForBackgroundRemoval: true } : {}) + }) + ) return { removed: true, ...result, diff --git a/src/main/runtime/sidebar-inputs/cached-scan.test.ts b/src/main/runtime/sidebar-inputs/cached-scan.test.ts index 4a3bd9dd3d9..6f2164f914f 100644 --- a/src/main/runtime/sidebar-inputs/cached-scan.test.ts +++ b/src/main/runtime/sidebar-inputs/cached-scan.test.ts @@ -85,4 +85,36 @@ describe('CachedScan', () => { expect(cache.current).toBe(1) vi.restoreAllMocks() }) + + it('calls a fresh waiter right after a scan started after it applies, not an older one', async () => { + const { cache, pending, applied, finish } = scanner() + cache.request() + const seen: number[][] = [] + // A scan already in flight started before the wait, so it cannot satisfy it. + cache.afterFresh(() => seen.push([...applied])) + await finish() + expect(applied).toEqual([1]) + expect(seen).toEqual([]) + expect(pending).toHaveLength(1) + await finish() + // Called synchronously after the apply, so its change publishes in the same revision. + expect(seen).toEqual([[1, 2]]) + }) + + it('calls a fresh waiter at once with no demand, and after a failed scan', async () => { + const idle = scanner(false) + const calls: string[] = [] + idle.cache.afterFresh(() => calls.push('idle')) + expect(calls).toEqual(['idle']) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + const failing = new CachedScan({ + label: 'test', + demand: () => true, + applied: () => {}, + scan: () => Promise.reject(new Error('git failed')) + }) + failing.afterFresh(() => calls.push('failed')) + await vi.waitFor(() => expect(calls).toEqual(['idle', 'failed'])) + vi.restoreAllMocks() + }) }) diff --git a/src/main/runtime/sidebar-inputs/cached-scan.ts b/src/main/runtime/sidebar-inputs/cached-scan.ts index ceb5fdfeb8b..80c6ffed4d8 100644 --- a/src/main/runtime/sidebar-inputs/cached-scan.ts +++ b/src/main/runtime/sidebar-inputs/cached-scan.ts @@ -11,6 +11,7 @@ export class CachedScan { private scanning: Promise | null = null private rescanQueued = false private stale = false + private readonly freshWaiters: { after: number; done: () => void }[] = [] constructor( private readonly deps: { @@ -41,8 +42,17 @@ export class CachedScan { this.stale = false this.scanning = this.deps .scan() - .then((value) => this.apply(version, value)) - .catch((error: unknown) => console.warn(`[${this.deps.label}] scan failed`, error)) + .then((value) => { + try { + this.apply(version, value) + } finally { + this.settleFresh(version) + } + }) + .catch((error: unknown) => { + console.warn(`[${this.deps.label}] scan failed`, error) + this.settleFresh(version) + }) .finally(() => { this.scanning = null if (this.rescanQueued) { @@ -62,6 +72,29 @@ export class CachedScan { } } + /** Calls `done` once a scan started after this call has finished; at once with no demand. */ + afterFresh(done: () => void): void { + if (!this.deps.demand()) { + done() + return + } + this.freshWaiters.push({ after: this.version, done }) + this.request(true) + } + + // Why synchronous after apply: a waiter's change lands in the same revision as the result. + private settleFresh(version: number): void { + const due = this.freshWaiters.filter((waiter) => waiter.after < version) + if (due.length === 0) { + return + } + const waiting = this.freshWaiters.filter((waiter) => waiter.after >= version) + this.freshWaiters.splice(0, this.freshWaiters.length, ...waiting) + for (const waiter of due) { + waiter.done() + } + } + /** Settles once every scan requested so far has applied; for one-shot readers. */ async settled(): Promise { await this.ready() diff --git a/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-inputs-feed.test.ts b/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-inputs-feed.test.ts index c9ff32e2855..72934993a12 100644 --- a/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-inputs-feed.test.ts +++ b/src/main/runtime/sidebar-inputs/orca-runtime-sidebar-inputs-feed.test.ts @@ -28,7 +28,7 @@ describe('the runtime sidebar inputs feed', () => { await phone.settle() expect(phone.view().temporal).toEqual({ sessionId: null, retention: null }) - const creation = requireSwitchedOn(runtime.workspaceCreations).open({ + const creation = requireSwitchedOn(runtime.workspaceLifecycle).openCreate({ creationId: 'c1', repoId: 'repo-1' }) diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-announcements.test.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-announcements.test.ts index 31f1728f606..0d086fb453c 100644 --- a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-announcements.test.ts +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-announcements.test.ts @@ -85,10 +85,11 @@ async function host() { catalogs: () => [] } as unknown as SidebarInputsFeedSources['pairedServerRows'], // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: as above. - workspaceCreations: { + workspaceLifecycle: { revision: revisions.creations, - snapshot: () => ({ creations: {} }) - } as unknown as SidebarInputsFeedSources['workspaceCreations'], + creationsSnapshot: () => ({ creations: {} }), + hints: () => ({}) + } as unknown as SidebarInputsFeedSources['workspaceLifecycle'], // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: as above; one cached worktree, already scanned. workspaceRows: { revision: revisions.rows, @@ -118,7 +119,8 @@ async function host() { store.patchWorkspaceSession({ activeTabId: 'tab-1' }, toRuntimeExecutionHostId('env-1')) }, temporal: { 'retention owner': () => temporal.forEach((listener) => listener()) }, - creations: { 'creation registry': () => revisions.creations.bump() }, + creations: { 'lifecycle queue': () => revisions.creations.bump() }, + inProgress: { 'lifecycle queue': () => revisions.creations.bump() }, workspaces: { 'workspace rows producer': () => revisions.rows.bump() }, detected: { 'detected checkouts producer': () => revisions.detected.bump() }, liveContent: { 'live content producer': () => revisions.liveContent.bump() }, @@ -215,7 +217,8 @@ describe('sidebar inputs feed announcements', () => { detach?.() return heard.some((change) => change?.urgent === true) } - expect(urgency('creations', 'creation registry')).toBe(true) + expect(urgency('creations', 'lifecycle queue')).toBe(true) + expect(urgency('inProgress', 'lifecycle queue')).toBe(true) expect(urgency('desktop', 'pin write')).toBe(true) expect(urgency('view', 'ui write')).toBe(true) }) diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-domains.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-domains.ts index 92edad60af4..dd243ec3edc 100644 --- a/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-domains.ts +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-feed-domains.ts @@ -10,7 +10,7 @@ import { WORKTREE_METADATA_DOMAINS } from '../../persistence/loading-store/workt import type { PersistedUIState } from '../../../shared/persisted-ui-state-types' import type { OrcaRuntimeWithSidebarTemporal } from './orca-runtime-sidebar-temporal' import type { PairedServerRowsProducer } from './paired-server-rows-producer' -import type { WorkspaceCreationRegistry } from './workspace-creation-registry' +import type { WorkspaceLifecycleQueue } from '../workspaces/workspace-lifecycle-queue' import type { WorkspaceRowsProducer } from './workspace-rows-producer' import type { RepoIconDataUrls } from './repo-icon-data-urls' import type { SidebarInputsFeedDomain } from './sidebar-inputs-feed' @@ -68,7 +68,7 @@ const DESKTOP_WRITE_DOMAINS = [...WORKTREE_METADATA_DOMAINS, 'settings', 'ui'] export type SidebarInputsFeedSources = { facts: OrcaRuntimeWithSidebarTemporal pairedServerRows: PairedServerRowsProducer - workspaceCreations: WorkspaceCreationRegistry + workspaceLifecycle: WorkspaceLifecycleQueue workspaceRows: WorkspaceRowsProducer repoIcons: Pick store: () => RuntimeStore | null @@ -82,7 +82,7 @@ export type SidebarInputsFeedSources = { export function sidebarInputsFeedDomains( sources: SidebarInputsFeedSources ): SidebarInputsFeedDomain[] { - const { facts, pairedServerRows, workspaceCreations, workspaceRows } = sources + const { facts, pairedServerRows, workspaceLifecycle, workspaceRows } = sources const urgent = { urgent: true } const watchCatalog = (changed: () => void): (() => void) => sources.onClientEvent( @@ -142,10 +142,17 @@ export function sidebarInputsFeedDomains( watch: (changed) => facts.onSidebarTemporalChange(changed) }, { + // The queue's creates, as the creations readers read them. name: 'creations', - read: () => workspaceCreations.snapshot().creations, + read: () => workspaceLifecycle.creationsSnapshot().creations, // An accepted command shows at once. - watch: (changed) => workspaceCreations.revision.onChange(() => changed(urgent)) + watch: (changed) => workspaceLifecycle.revision.onChange(() => changed(urgent)) + }, + { + // Each workspace's running, waiting or paused lifecycle command, keyed host-qualified. + name: 'inProgress', + read: () => workspaceLifecycle.hints(), + watch: (changed) => workspaceLifecycle.revision.onChange(() => changed(urgent)) }, { name: 'workspaces', diff --git a/src/main/runtime/sidebar-inputs/sidebar-inputs-switch-off.test.ts b/src/main/runtime/sidebar-inputs/sidebar-inputs-switch-off.test.ts index c0945585853..216fb12274d 100644 --- a/src/main/runtime/sidebar-inputs/sidebar-inputs-switch-off.test.ts +++ b/src/main/runtime/sidebar-inputs/sidebar-inputs-switch-off.test.ts @@ -10,7 +10,7 @@ describe('sidebar inputs with the switch off', () => { expect(runtime.sidebarFacts).toBeNull() expect(runtime.pairedServerRows).toBeNull() expect(runtime.repoIconDataUrls).toBeNull() - expect(runtime.workspaceCreations).toBeNull() + expect(runtime.workspaceLifecycle).toBeNull() expect(runtime.collapseWriteOrder).toBeNull() const gated = new Set(SIDEBAR_INPUT_METHODS.map((method) => method.name)) expect(ALL_RPC_METHODS.filter((method) => gated.has(method.name))).toEqual([]) diff --git a/src/main/runtime/sidebar-inputs/workspace-creation-registry.test.ts b/src/main/runtime/sidebar-inputs/workspace-creation-registry.test.ts deleted file mode 100644 index bcf23aa3e51..00000000000 --- a/src/main/runtime/sidebar-inputs/workspace-creation-registry.test.ts +++ /dev/null @@ -1,52 +0,0 @@ -import { describe, expect, it } from 'vitest' -import { WorkspaceCreationRegistry } from './workspace-creation-registry' - -describe('WorkspaceCreationRegistry', () => { - it('lists a creation once opened, names its workspace, and drops it on close', () => { - const registry = new WorkspaceCreationRegistry() - const creation = registry.open({ creationId: 'c1', repoId: 'repo-1' }) - expect(registry.snapshot().creations).toEqual({ c1: { repoId: 'repo-1' } }) - const opened = registry.snapshot().revision - - creation.named('repo-1::/wt') - expect(registry.snapshot().creations).toEqual({ - c1: { repoId: 'repo-1', worktreeId: 'repo-1::/wt' } - }) - expect(registry.snapshot().revision).toBeGreaterThan(opened) - - creation.close() - expect(registry.snapshot().creations).toEqual({}) - }) - - it('mints an id for a caller that sends none', () => { - const registry = new WorkspaceCreationRegistry() - const creation = registry.open({ repoId: 'repo-1' }) - const ids = Object.keys(registry.snapshot().creations) - expect(ids).toHaveLength(1) - expect(ids[0]).not.toBe('') - creation.close() - expect(registry.snapshot().creations).toEqual({}) - }) - - it('leaves a concurrent retry under the same id listed when the first create closes', () => { - const registry = new WorkspaceCreationRegistry() - const first = registry.open({ creationId: 'c1', repoId: 'repo-1' }) - const second = registry.open({ creationId: 'c1', repoId: 'repo-1' }) - - first.named('repo-1::/stale') - first.close() - expect(registry.snapshot().creations).toEqual({ c1: { repoId: 'repo-1' } }) - - second.close() - expect(registry.snapshot().creations).toEqual({}) - }) - - it('closes once', () => { - const registry = new WorkspaceCreationRegistry() - const creation = registry.open({ creationId: 'c1', repoId: 'repo-1' }) - creation.close() - const closed = registry.snapshot().revision - creation.close() - expect(registry.snapshot().revision).toBe(closed) - }) -}) diff --git a/src/main/runtime/sidebar-inputs/workspace-creation-registry.ts b/src/main/runtime/sidebar-inputs/workspace-creation-registry.ts deleted file mode 100644 index 2240c62ccac..00000000000 --- a/src/main/runtime/sidebar-inputs/workspace-creation-registry.ts +++ /dev/null @@ -1,51 +0,0 @@ -import { ProducerRevision } from './producer-revision' -import { randomUUID } from 'node:crypto' -import type { - WorkspaceCreation, - WorkspaceCreationsSnapshot -} from '../../../shared/sidebar-list/workspace-creations' - -/** One listed creation, owned by the create that opened it. */ -export type WorkspaceCreationHandle = { - /** The workspace is persisted; called before the host announces it. */ - named: (worktreeId: string) => void - close: () => void -} - -/** Every creation this host is running, whoever asked: its own window, a phone, the CLI. */ -export class WorkspaceCreationRegistry { - private readonly creations = new Map() - readonly revision = new ProducerRevision() - - /** Opened once the project is known. A caller without an id gets one, so a CLI create is listed. */ - open(start: { creationId?: string; repoId: string }): WorkspaceCreationHandle { - const creationId = start.creationId ?? randomUUID() - const entry: WorkspaceCreation = { repoId: start.repoId } - this.creations.set(creationId, entry) - this.revision.bump() - // Identity, not id: a concurrent retry under the same id owns the entry from then on. - const owns = (): boolean => this.creations.get(creationId) === entry - return { - named: (worktreeId) => { - if (owns() && entry.worktreeId !== worktreeId) { - entry.worktreeId = worktreeId - this.revision.bump() - } - }, - close: () => { - if (owns()) { - this.creations.delete(creationId) - this.revision.bump() - } - } - } - } - - snapshot(): WorkspaceCreationsSnapshot { - const creations: WorkspaceCreationsSnapshot['creations'] = {} - for (const [creationId, entry] of this.creations) { - creations[creationId] = { ...entry } - } - return { revision: this.revision.value, creations } - } -} diff --git a/src/main/runtime/sidebar-inputs/workspace-rows-producer.ts b/src/main/runtime/sidebar-inputs/workspace-rows-producer.ts index b26496a5713..b80274d0dcf 100644 --- a/src/main/runtime/sidebar-inputs/workspace-rows-producer.ts +++ b/src/main/runtime/sidebar-inputs/workspace-rows-producer.ts @@ -17,6 +17,7 @@ import { WORKTREE_METADATA_DOMAINS } from '../../persistence/loading-store/workt import { ProducerRevision } from './producer-revision' import { WorkspaceGitIdentity } from './workspace-git-identity' import { CachedScan } from './cached-scan' +import type { WorkspaceLifecycleQueue } from '../workspaces/workspace-lifecycle-queue' // The settings the visibility filter reads: the defaults, and the layouts that classify a checkout. const VISIBILITY_SETTINGS = [ @@ -51,6 +52,8 @@ export type WorkspaceRowsSources = { join: (scans: ResolvedWorktreeRepoScans) => ResolvedWorktreeScanJoin /** worktree.ps's visibility filters over a listing. */ visible: (snapshot: ResolvedWorktreeSnapshot) => ResolvedWorktreeSnapshot + /** Its finished creates and deletes keep their hints until a fresh scan lists them. */ + lifecycle?: Pick | null onClientEvent: ( listener: (event: RuntimeClientEvent) => void, options: { consumesTerminalSideEffects: false } @@ -89,6 +92,7 @@ export class WorkspaceRowsProducer { demand: () => this.revision.hasListeners(), applied: () => this.applied() }) + sources.lifecycle?.setMembership((done) => this.scans.afterFresh(done)) sources.onClientEvent( (event) => { if (event.type === 'worktreesChanged' || event.type === 'reposChanged') { diff --git a/src/main/runtime/workspaces/workspace-creation.ts b/src/main/runtime/workspaces/workspace-creation.ts index c918123a376..eb53a9947e5 100644 --- a/src/main/runtime/workspaces/workspace-creation.ts +++ b/src/main/runtime/workspaces/workspace-creation.ts @@ -23,10 +23,10 @@ import type { RuntimeWorkspaceCreateEvents } from '../../workspace-create-teleme import { createdWorkspaceActivationCallbacks } from './workspace-create-activation' import { assertWslMoveRepoWritable } from '../../persistence/moving-wsl-projects/wsl-move-scope-guard' import { resolveManagedWorktreeCreateStartup } from './managed-worktree-create-startup' -import { - WorkspaceCreationRegistry, - type WorkspaceCreationHandle -} from '../sidebar-inputs/workspace-creation-registry' +import { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue' +import type { WorkspaceCreationHandle } from './workspace-lifecycle-creations' +import { workspaceLifecycleKey } from './workspace-lifecycle-commands' +import { getRepoExecutionHostId } from '../../../shared/execution-host' import { ORCA_SIDEBAR_INPUTS_IS_MAIN } from '../../../shared/sidebar-inputs-switch' export abstract class OrcaRuntimeWithCreateManagedWorktree extends OrcaRuntimeWithGetWorktreeTerminalProvisioningHost { @@ -35,8 +35,8 @@ export abstract class OrcaRuntimeWithCreateManagedWorktree extends OrcaRuntimeWi childRepo?: Repo ): Promise - // Switch point: only a build where main produces the sidebar's inputs lists creations. - readonly workspaceCreations = ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceCreationRegistry() : null + // Switch point: only a build where main produces the sidebar's inputs queues lifecycle commands. + readonly workspaceLifecycle = ORCA_SIDEBAR_INPUTS_IS_MAIN ? new WorkspaceLifecycleQueue() : null async createManagedWorktree( args: RuntimeManagedWorktreeCreateArgs @@ -68,7 +68,11 @@ export abstract class OrcaRuntimeWithCreateManagedWorktree extends OrcaRuntimeWi const repo = await this.resolveRepoSelector(args.repoSelector) // Listed from here: a create refused before its project resolved has nothing to show. - const creation = this.workspaceCreations?.open({ creationId: args.creationId, repoId: repo.id }) + const creation = this.workspaceLifecycle?.openCreate({ + creationId: args.creationId, + repoId: repo.id, + keyOf: (worktreeId) => workspaceLifecycleKey(getRepoExecutionHostId(repo), worktreeId) + }) try { return await this.createInResolvedProject(args, repo, creation, rearm, events) } finally { diff --git a/src/main/runtime/workspaces/workspace-lifecycle-command.ts b/src/main/runtime/workspaces/workspace-lifecycle-command.ts new file mode 100644 index 00000000000..54663da7c7a --- /dev/null +++ b/src/main/runtime/workspaces/workspace-lifecycle-command.ts @@ -0,0 +1,52 @@ +import type { WorkspaceLifecycleKind } from '../../../shared/sidebar-list/workspace-lifecycle-hints' + +export const WORKSPACE_DELETED_ERROR = 'workspace_deleted' + +/** `unverifiable`: the execution host never confirmed the outcome, so the queue pauses. */ +export type LifecycleOutcome = 'settled' | 'unverifiable' + +/** Delete options: an equal delete joins the pending one, a different one is refused. */ +export type LifecycleDeleteOptions = { + key: string + /** The explicit waiver (Force Delete, `--force`): runs past a pause, as it always has. */ + force: boolean + /** Names the workspace in the "already in progress" refusal. */ + label: string +} + +export type LifecycleCommand = { + kind: Exclude + run: () => Promise + /** Reads a reply; defaults to settled. */ + outcome?: (result: T) => LifecycleOutcome + /** Whether a thrown error means contact was lost mid-run rather than a definite failure. */ + lostContact?: (error: unknown) => boolean + /** After an unconfirmed outcome: re-reads the owner's state, throwing while it cannot answer. */ + observe?: () => Promise + delete?: LifecycleDeleteOptions +} + +/** Calls `done` once a listing that started after this call has applied. */ +export type LifecycleMembership = (done: () => void) => void + +export type LifecycleAttempt = { + outcome: LifecycleOutcome + value?: unknown + error?: unknown + threw: boolean +} + +export async function attemptLifecycleCommand( + command: LifecycleCommand +): Promise { + try { + const value = await command.run() + return { outcome: command.outcome?.(value) ?? 'settled', value, threw: false } + } catch (error) { + return { + outcome: command.lostContact?.(error) ? 'unverifiable' : 'settled', + error, + threw: true + } + } +} diff --git a/src/main/runtime/workspaces/workspace-lifecycle-commands.test.ts b/src/main/runtime/workspaces/workspace-lifecycle-commands.test.ts new file mode 100644 index 00000000000..73d40c30f78 --- /dev/null +++ b/src/main/runtime/workspaces/workspace-lifecycle-commands.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it } from 'vitest' +import { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue' +import { + closeOutcome, + deleteLostContact, + observeTerminals, + queueTerminalWorkspaceCommand, + queueWorkspaceDelete, + sleepOutcome, + terminalLostContact, + workspaceLifecycleKey +} from './workspace-lifecycle-commands' + +const flush = () => new Promise((resolve) => setTimeout(resolve, 0)) + +const repos = [ + { id: 'repo-local', connectionId: null, executionHostId: null }, + { id: 'repo-ssh', connectionId: 'box', executionHostId: null } +] + +const BY_NAME: Record = { feature: 'repo-local::/a' } + +function runtime(queue: WorkspaceLifecycleQueue | null) { + return { + workspaceLifecycle: queue, + listRepos: () => repos, + showManagedWorktree: async (selector: string) => { + const id = BY_NAME[selector] ?? selector.replace(/^id:/, '') + if (!id.includes('::')) { + throw new Error('selector_not_found') + } + return { id, repoId: id.split('::')[0] } + }, + listTerminals: async () => ({ totalCount: 0 }) + } +} + +const removal = { + force: true, + runHooks: false, + allowUnverifiedPtyStop: false, + allowFailedArchiveHook: false +} + +describe('workspace lifecycle commands', () => { + it('runs directly in a build with no queue', async () => { + let ran = false + const reply = await queueTerminalWorkspaceCommand(runtime(null), 'id:repo-local::/a', { + kind: 'sleep', + run: async () => { + ran = true + return 'slept' + } + }) + expect([ran, reply]).toEqual([true, 'slept']) + }) + + it('keys a command by the host its repo routes to, the same key a delete uses', async () => { + const queue = new WorkspaceLifecycleQueue() + let release!: () => void + const gate = new Promise((resolve) => (release = resolve)) + const sleeping = queueTerminalWorkspaceCommand(runtime(queue), 'id:repo-ssh::/a', { + kind: 'sleep', + run: () => gate + }) + await flush() + expect(queue.hints()).toEqual({ + [workspaceLifecycleKey('ssh:box', 'repo-ssh::/a')]: { kind: 'sleep', queuedBehind: 0 } + }) + release() + await sleeping + }) + + it('makes a delete by name wait for the running sleep of the workspace it names', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + let release!: () => void + const gate = new Promise((resolve) => (release = resolve)) + const sleeping = queueTerminalWorkspaceCommand(runtime(queue), 'id:repo-local::/a', { + kind: 'sleep', + run: async () => { + log.push('sleep') + await gate + log.push('slept') + } + }) + const deleting = queueWorkspaceDelete( + runtime(queue), + 'feature', + undefined, + removal, + async () => { + log.push('delete') + } + ) + await flush() + expect(log).toEqual(['sleep']) + expect(queue.hints()).toEqual({ + [workspaceLifecycleKey('local', 'repo-local::/a')]: { kind: 'delete', queuedBehind: 0 } + }) + release() + await Promise.all([sleeping, deleting]) + expect(log).toEqual(['sleep', 'slept', 'delete']) + }) + + it('runs a delete of a workspace no lookup finds without queueing it', async () => { + const queue = new WorkspaceLifecycleQueue() + let ran = false + await queueWorkspaceDelete(runtime(queue), 'gone', undefined, removal, async () => { + ran = true + }) + expect(ran).toBe(true) + }) + + it('reads lost contact the same way for sleep, close and delete', async () => { + expect(terminalLostContact(new Error('terminal_liveness_unavailable'))).toBe(true) + expect(terminalLostContact(new Error('terminal_worktree_sleep_failed'))).toBe(false) + expect( + deleteLostContact(new Error('Timed out waiting for physical PTY teardown: repo::/a.')) + ).toBe(true) + expect(deleteLostContact(new Error('Repo not found: repo'))).toBe(false) + const silent = { + listTerminals: async () => { + throw new Error('terminal_liveness_unavailable') + } + } + await expect(observeTerminals(silent, 'id:a')()).rejects.toThrow() + const gone = { + listTerminals: async () => { + throw new Error('selector_not_found') + } + } + // Any answer is the host answering. + await expect(observeTerminals(gone, 'id:a')()).resolves.toBeUndefined() + }) + + it('reads an unconfirmed sleep or close as unverifiable, and an observed result as settled', () => { + const sleep = { stopped: 1, stoppedPtyIds: [], livePtyIds: [] } + expect( + sleepOutcome({ + ...sleep, + postStopVerified: false, + postStopFailure: 'terminal_liveness_unavailable' + }) + ).toBe('unverifiable') + // A terminal observed still running is a definite answer. + expect( + sleepOutcome({ + ...sleep, + postStopVerified: false, + postStopFailure: 'terminal_worktree_sleep_still_live', + remainingLivePtyIds: ['pty-1'] + }) + ).toBe('settled') + expect( + sleepOutcome({ stopped: 1, stoppedPtyIds: [], livePtyIds: [], postStopVerified: true }) + ).toBe('settled') + const closed = { closed: 1, stopped: 1, retiredSurfaces: true as const } + expect(closeOutcome({ ...closed, ptyStopVerdict: 'unverifiable' })).toBe('unverifiable') + // A terminal observed still live is a definite answer, not an unconfirmed one. + expect(closeOutcome({ ...closed, ptyStopVerdict: 'live' })).toBe('settled') + }) +}) diff --git a/src/main/runtime/workspaces/workspace-lifecycle-commands.ts b/src/main/runtime/workspaces/workspace-lifecycle-commands.ts new file mode 100644 index 00000000000..bfb4fbb9c25 --- /dev/null +++ b/src/main/runtime/workspaces/workspace-lifecycle-commands.ts @@ -0,0 +1,230 @@ +import type { + RuntimeWorktreeTerminalCloseResult, + RuntimeWorktreeTerminalSleepResult +} from '../../../shared/runtime-types' +import { + LOCAL_EXECUTION_HOST_ID, + parseExecutionHostId, + type ExecutionHostId +} from '../../../shared/execution-host' +import { splitWorktreeId } from '../../../shared/worktree/id' +import { + isProvenLivePtyRemovalError, + isUnstoppedPtyRemovalError +} from '../../../shared/worktree/removal' +import { + getExplicitWorktreeIdSelector, + getRuntimeWorktreeRemovalOptionsKey +} from '../runtime-worktree-selection' + +import { hostSleepKey } from '../terminal/stop/worktree-terminal-sleep-key' +import { resolveWorktreeHostRouting, type LaunchHostRepo } from '../worktree-launch-host-repo' +import type { + LifecycleCommand, + LifecycleDeleteOptions, + LifecycleOutcome +} from './workspace-lifecycle-command' +import type { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue' + +type RemovalFlag = 'force' | 'runHooks' | 'allowUnverifiedPtyStop' | 'allowFailedArchiveHook' + +/** One key per host's copy of a workspace, the same one its terminals sleep under. */ +export function workspaceLifecycleKey(hostId: ExecutionHostId, worktreeId: string): string { + return hostSleepKey(hostId, worktreeId) +} + +type LifecycleRuntime = { + readonly workspaceLifecycle: WorkspaceLifecycleQueue | null + listRepos: () => readonly LaunchHostRepo[] + showManagedWorktree: ( + selector: string + ) => Promise<{ id: string; repoId: string; hostId?: string | null }> +} + +type ObservingRuntime = { + listTerminals: ( + selector: string, + limit: number, + opts: { requireFreshPtyLiveness?: boolean } + ) => Promise +} + +/** Runs `command` in its workspace's queue, or directly in a build with none. */ +export function runWorkspaceLifecycleCommand( + queue: WorkspaceLifecycleQueue | null, + key: string, + command: LifecycleCommand +): Promise { + return queue ? queue.submit(key, command) : command.run() +} + +const LIVENESS_UNAVAILABLE = 'terminal_liveness_unavailable' + +/** A thrown terminal command that lost the host's liveness answer; any other error is definite. */ +export function terminalLostContact(error: unknown): boolean { + return error instanceof Error && error.message === LIVENESS_UNAVAILABLE +} + +// The host could not say whether the stops took; a terminal observed still live is definite. +export function sleepOutcome(result: RuntimeWorktreeTerminalSleepResult): LifecycleOutcome { + return result.postStopVerified === false && result.postStopFailure === LIVENESS_UNAVAILABLE + ? 'unverifiable' + : 'settled' +} + +export function closeOutcome(result: RuntimeWorktreeTerminalCloseResult): LifecycleOutcome { + return result.ptyStopVerdict === 'unverifiable' ? 'unverifiable' : 'settled' +} + +/** A delete whose PTY stop went unproven (not seen live) lost contact; any other error is definite. */ +export function deleteLostContact(error: unknown): boolean { + if (!(error instanceof Error)) { + return false + } + return ( + error.message === LIVENESS_UNAVAILABLE || + (isUnstoppedPtyRemovalError(error.message) && !isProvenLivePtyRemovalError(error.message)) + ) +} + +/** Re-reads the workspace's terminal liveness; throws only while the host still cannot answer. */ +export function observeTerminals(runtime: ObservingRuntime, selector: string) { + return async (): Promise => { + try { + await runtime.listTerminals(selector, 1, { requireFreshPtyLiveness: true }) + } catch (error) { + // Any other answer (the workspace is gone, say) is the host answering. + if (terminalLostContact(error)) { + throw error + } + } + } +} + +/** The sync key of an `id:` selector routed by the repo rows; null when only a lookup can say. */ +function explicitKey( + repos: readonly LaunchHostRepo[], + selector: string, + hostId?: string | null +): string | null { + const worktreeId = getExplicitWorktreeIdSelector(selector) + const repoId = worktreeId ? splitWorktreeId(worktreeId)?.repoId : undefined + if (!worktreeId || !repoId) { + return null + } + const named = parseExecutionHostId(hostId ?? undefined)?.id + if (named) { + return workspaceLifecycleKey(named, worktreeId) + } + const routing = resolveWorktreeHostRouting(repos, { repoId }) + if (routing.kind === 'ambiguous') { + return null + } + return workspaceLifecycleKey( + routing.kind === 'resolved' ? routing.hostId : LOCAL_EXECUTION_HOST_ID, + worktreeId + ) +} + +/** The key a lookup finds: the workspace's own host, else as the repo rows route it. */ +async function lookedUpKey( + runtime: LifecycleRuntime, + selector: string, + hostId?: string +): Promise { + const worktree = await runtime.showManagedWorktree(selector) + const named = parseExecutionHostId(hostId ?? worktree.hostId ?? undefined)?.id + if (named) { + return workspaceLifecycleKey(named, worktree.id) + } + const routing = resolveWorktreeHostRouting(runtime.listRepos(), worktree) + if (routing.kind === 'ambiguous') { + throw new Error('worktree_execution_host_unresolved') + } + return workspaceLifecycleKey( + routing.kind === 'resolved' ? routing.hostId : LOCAL_EXECUTION_HOST_ID, + worktree.id + ) +} + +/** `terminal.sleep`, `terminal.closeAll` and `worktree.sleep` as queued commands, in arrival order. */ +export function queueTerminalWorkspaceCommand( + runtime: LifecycleRuntime, + selector: string, + command: LifecycleCommand +): Promise { + const queue = runtime.workspaceLifecycle + if (!queue) { + return command.run() + } + const key = explicitKey(runtime.listRepos(), selector) + return key + ? queue.submit(key, command) + : queue.submitResolving(() => lookedUpKey(runtime, selector), command) +} + +/** `worktree.sleep`: a phone's or the CLI's sleep of the whole workspace. */ +export function queueWorktreeSleep( + runtime: LifecycleRuntime & + ObservingRuntime & { + sleepManagedWorktree: (selector: string) => Promise<{ worktreeId: string }> + }, + selector: string +): Promise<{ worktreeId: string }> { + return queueTerminalWorkspaceCommand(runtime, selector, { + kind: 'sleep', + run: () => runtime.sleepManagedWorktree(selector), + lostContact: terminalLostContact, + observe: observeTerminals(runtime, selector) + }) +} + +/** + * `worktree.rm`: keyed by the workspace's resolved id, whatever selector named it, so a delete by + * name waits for that workspace's running sleep. A workspace no lookup finds runs unqueued. + */ +export function queueWorkspaceDelete( + runtime: LifecycleRuntime & ObservingRuntime, + selector: string, + hostId: string | undefined, + requested: Partial>, + /** Called with the normalized flags, plus the caller's host when it named one. */ + run: (removal: Record & { hostId?: string }) => Promise +): Promise { + const removal: Record = { + force: requested.force === true, + runHooks: requested.runHooks === true, + allowUnverifiedPtyStop: requested.allowUnverifiedPtyStop === true, + allowFailedArchiveHook: requested.allowFailedArchiveHook === true + } + const options: LifecycleDeleteOptions = { + key: getRuntimeWorktreeRemovalOptionsKey(removal), + force: removal.allowUnverifiedPtyStop, + label: getExplicitWorktreeIdSelector(selector) ?? selector + } + const command: LifecycleCommand = { + kind: 'delete', + run: () => run({ ...removal, ...(hostId ? { hostId } : {}) }), + lostContact: deleteLostContact, + observe: observeTerminals(runtime, selector), + delete: options + } + const queue = runtime.workspaceLifecycle + if (!queue) { + return command.run() + } + const key = explicitKey(runtime.listRepos(), selector, hostId) + return key + ? queue.submit(key, command) + : queue.submitResolving( + () => + lookedUpKey(runtime, selector, hostId).catch((error: unknown) => { + // A delete legitimately reaches a workspace Git no longer lists; nothing orders it. + if (error instanceof Error && error.message === 'selector_not_found') { + return null + } + throw error + }), + command + ) +} diff --git a/src/main/runtime/workspaces/workspace-lifecycle-creations.ts b/src/main/runtime/workspaces/workspace-lifecycle-creations.ts new file mode 100644 index 00000000000..6200ae89b69 --- /dev/null +++ b/src/main/runtime/workspaces/workspace-lifecycle-creations.ts @@ -0,0 +1,108 @@ +import { randomUUID } from 'node:crypto' +import type { + WorkspaceCreation, + WorkspaceCreationsSnapshot +} from '../../../shared/sidebar-list/workspace-creations' +import type { ProducerRevision } from '../sidebar-inputs/producer-revision' +import type { LifecycleMembership } from './workspace-lifecycle-command' + +export type WorkspaceCreationHandle = { + /** The workspace is persisted; called before the host announces it. */ + named: (worktreeId: string) => void + close: () => void +} + +/** How a create holds its named workspace's queue while it runs. */ +export type CreationQueuePort = { + revision: ProducerRevision + membership: () => LifecycleMembership + deletePending: (key: string) => boolean + bind: (key: string) => Token + finish: (key: string, token: Token) => void +} + +type Creation = WorkspaceCreation & { + key?: string + token?: Token + settling?: boolean + closed?: boolean +} + +/** The creates in flight: listed from open, bound to the workspace's queue once named. */ +export class WorkspaceLifecycleCreations { + private readonly creations = new Map>() + + constructor(private readonly port: CreationQueuePort) {} + + /** Opened once the project is known; a caller without an id gets one, so a CLI create is listed. */ + open(start: { + creationId?: string + repoId: string + /** The named workspace's queue key; commands sent to it then wait behind the create. */ + keyOf?: (worktreeId: string) => string + }): WorkspaceCreationHandle { + const { port } = this + const creationId = start.creationId ?? randomUUID() + const creation: Creation = { repoId: start.repoId } + this.creations.set(creationId, creation) + port.revision.bump() + // Identity, not id: a concurrent retry under the same id owns the listing from then on. + const owns = (): boolean => this.creations.get(creationId) === creation + return { + named: (worktreeId) => { + if (creation.closed || !owns() || creation.worktreeId === worktreeId) { + return + } + creation.worktreeId = worktreeId + const key = start.keyOf?.(worktreeId) + // Why not behind a delete: that delete is the previous occupant's, never this new one's. + if (key && !creation.key && !port.deletePending(key)) { + creation.key = key + creation.token = port.bind(key) + } + port.revision.bump() + }, + close: () => { + if (creation.closed) { + return + } + creation.closed = true + // Its own queue entry always ends here, owned or not, so the workspace's queue moves on. + if (creation.key && creation.token !== undefined) { + port.finish(creation.key, creation.token) + } + if (!owns()) { + return + } + const finish = (): void => { + if (owns()) { + this.creations.delete(creationId) + } + port.revision.bump() + } + if (creation.worktreeId === undefined) { + finish() + return + } + // Why held: the creation shows until a listing shows the workspace, in the same revision. + creation.settling = true + port.membership()(finish) + } + } + } + + snapshot(): WorkspaceCreationsSnapshot { + const creations: WorkspaceCreationsSnapshot['creations'] = {} + for (const [creationId, { repoId, worktreeId }] of this.creations) { + creations[creationId] = worktreeId === undefined ? { repoId } : { repoId, worktreeId } + } + return { revision: this.port.revision.value, creations } + } + + /** Queue keys of finished creates still waiting for a listing to show them. */ + settlingKeys(): string[] { + return [...this.creations.values()].flatMap(({ key, settling }) => + key && settling ? [key] : [] + ) + } +} diff --git a/src/main/runtime/workspaces/workspace-lifecycle-queue.test.ts b/src/main/runtime/workspaces/workspace-lifecycle-queue.test.ts new file mode 100644 index 00000000000..ac6324164ad --- /dev/null +++ b/src/main/runtime/workspaces/workspace-lifecycle-queue.test.ts @@ -0,0 +1,521 @@ +import { describe, expect, it } from 'vitest' +import { WorkspaceLifecycleQueue } from './workspace-lifecycle-queue' +import { WORKSPACE_DELETED_ERROR, type LifecycleCommand } from './workspace-lifecycle-command' + +const KEY = 'local|repo-1::/wt' + +function deferred() { + let resolve!: (value: T) => void + let reject!: (error: unknown) => void + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) + return { promise, resolve, reject } +} + +/** A command that records when it starts and finishes only when released. */ +function gated( + log: string[], + name: string, + kind: LifecycleCommand['kind'], + value: T, + extra: Partial> = {} +) { + const gate = deferred() + const command: LifecycleCommand = { + kind, + run: async () => { + log.push(`start ${name}`) + await gate.promise + log.push(`end ${name}`) + return value + }, + ...extra + } + return { command, release: () => gate.resolve() } +} + +const flush = () => new Promise((resolve) => setTimeout(resolve, 0)) + +const deleteOptions = (key: string, force = false) => ({ + delete: { key, force, label: 'repo-1::/wt' } +}) + +/** A queue whose retries after a silent owner wait for the test to release them. */ +function manualRetryQueue() { + const retries: (() => void)[] = [] + const queue = new WorkspaceLifecycleQueue( + () => new Promise((resolve) => retries.push(resolve)) + ) + return { queue, retry: () => retries.splice(0).forEach((resolve) => resolve()) } +} + +describe('WorkspaceLifecycleQueue', () => { + it('runs one command at a time, in the order accepted, without merging kinds', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const sleep = gated(log, 'sleep', 'sleep', 'slept') + const close = gated(log, 'close', 'close', 'closed') + const secondSleep = gated(log, 'sleep-2', 'sleep', 'slept-2') + const replies = [ + queue.submit(KEY, sleep.command), + queue.submit(KEY, close.command), + queue.submit(KEY, secondSleep.command) + ] + await flush() + expect(log).toEqual(['start sleep']) + expect(queue.hints()).toEqual({ [KEY]: { kind: 'sleep', queuedBehind: 2 } }) + + sleep.release() + await flush() + expect(log).toEqual(['start sleep', 'end sleep', 'start close']) + expect(queue.hints()).toEqual({ [KEY]: { kind: 'close', queuedBehind: 1 } }) + close.release() + await flush() + secondSleep.release() + expect(await Promise.all(replies)).toEqual(['slept', 'closed', 'slept-2']) + expect(log).toEqual([ + 'start sleep', + 'end sleep', + 'start close', + 'end close', + 'start sleep-2', + 'end sleep-2' + ]) + expect(queue.hints()).toEqual({}) + }) + + it('runs a repeated command again rather than merging it into the first', async () => { + const queue = new WorkspaceLifecycleQueue() + let runs = 0 + const sleep: LifecycleCommand = { kind: 'sleep', run: async () => ++runs } + expect(await Promise.all([queue.submit(KEY, sleep), queue.submit(KEY, sleep)])).toEqual([1, 2]) + }) + + it('keeps separate workspaces apart, including one id on two hosts', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const local = gated(log, 'local', 'sleep', 1) + const ssh = gated(log, 'ssh', 'sleep', 2) + void queue.submit(KEY, local.command) + void queue.submit('ssh:box|repo-1::/wt', ssh.command) + await flush() + expect(log).toEqual(['start local', 'start ssh']) + local.release() + ssh.release() + }) + + it('delete cancels what has not started and waits for what has', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const sleep = gated(log, 'sleep', 'sleep', 'slept') + const wake = gated(log, 'wake', 'wake', 'woken') + const remove = gated(log, 'delete', 'delete', 'deleted', deleteOptions('normal')) + const sleeping = queue.submit(KEY, sleep.command) + const waking = queue.submit(KEY, wake.command) + await flush() + const deleting = queue.submit(KEY, remove.command) + // Acknowledged at once: its hint replaces the queued ones. + expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0 } }) + await expect(waking).rejects.toThrow(WORKSPACE_DELETED_ERROR) + // A command after the delete is refused, not run on a workspace about to go. + await expect(queue.submit(KEY, wake.command)).rejects.toThrow(WORKSPACE_DELETED_ERROR) + expect(log).toEqual(['start sleep']) + + sleep.release() + await flush() + expect(log).toEqual(['start sleep', 'end sleep', 'start delete']) + remove.release() + expect(await sleeping).toBe('slept') + expect(await deleting).toBe('deleted') + expect(log).not.toContain('start wake') + }) + + it('joins a second delete with the same options, and refuses one with different options', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const remove = gated(log, 'delete', 'delete', 'deleted', deleteOptions('normal')) + const first = queue.submit(KEY, remove.command) + const second = queue.submit( + KEY, + gated(log, 'delete-2', 'delete', 'other', deleteOptions('normal')).command + ) + await expect( + queue.submit(KEY, gated(log, 'delete-3', 'delete', 'x', deleteOptions('hooks')).command) + ).rejects.toThrow('Worktree deletion already in progress: repo-1::/wt') + remove.release() + expect(await Promise.all([first, second])).toEqual(['deleted', 'deleted']) + expect(log).toEqual(['start delete', 'end delete']) + }) + + it('pauses on an unconfirmed outcome; re-observes when the owner answers, never replaying it', async () => { + const { queue, retry } = manualRetryQueue() + const log: string[] = [] + let reachable = false + const sleep: LifecycleCommand<{ postStopVerified: boolean }> = { + kind: 'sleep', + run: async () => { + log.push('sleep') + return { postStopVerified: reachable } + }, + outcome: (result) => (result.postStopVerified ? 'settled' : 'unverifiable'), + observe: async () => { + log.push('observe') + if (!reachable) { + throw new Error('terminal_liveness_unavailable') + } + } + } + // The requester gets the unconfirmed reply (today's toast); the workspace stays as observed. + expect(await queue.submit(KEY, sleep)).toEqual({ postStopVerified: false }) + expect(queue.isPaused(KEY)).toBe(true) + expect(queue.hints()).toEqual({ [KEY]: { kind: 'sleep', queuedBehind: 0, paused: true } }) + + // A command issued while paused waits behind the pause; it is neither refused nor run. + const woke: string[] = [] + const waking = queue.submit(KEY, { + kind: 'wake', + run: async () => { + woke.push('wake') + return 'woken' + } + }) + await flush() + expect(log).toEqual(['sleep', 'observe']) + expect(woke).toEqual([]) + expect(queue.hints()).toEqual({ [KEY]: { kind: 'wake', queuedBehind: 0, paused: true } }) + + // Still silent on the next ask; then the owner answers and the queue carries on. + retry() + await flush() + expect(log).toEqual(['sleep', 'observe', 'observe']) + reachable = true + retry() + expect(await waking).toBe('woken') + expect(log).toEqual(['sleep', 'observe', 'observe', 'observe']) + expect(queue.isPaused(KEY)).toBe(false) + expect(queue.hints()).toEqual({}) + }) + + it('re-runs a repeated command once contact is back', async () => { + const { queue } = manualRetryQueue() + let reachable = false + let sleeps = 0 + const sleep: LifecycleCommand = { + kind: 'sleep', + run: async () => { + sleeps += 1 + return reachable + }, + outcome: (ok) => (ok ? 'settled' : 'unverifiable'), + observe: async () => { + if (!reachable) { + throw new Error('lost') + } + } + } + expect(await queue.submit(KEY, sleep)).toBe(false) + reachable = true + expect(await queue.submit(KEY, sleep)).toBe(true) + expect(sleeps).toBe(2) + expect(queue.isPaused(KEY)).toBe(false) + }) + + it('pauses on lost contact mid-run, but not on a definite failure', async () => { + const queue = new WorkspaceLifecycleQueue() + const lost = (message: string): LifecycleCommand => ({ + kind: 'close', + run: async () => { + throw new Error(message) + }, + lostContact: (error) => error instanceof Error && error.message === 'lost' + }) + await expect(queue.submit(KEY, lost('refused'))).rejects.toThrow('refused') + expect(queue.isPaused(KEY)).toBe(false) + await expect(queue.submit(KEY, lost('lost'))).rejects.toThrow('lost') + expect(queue.isPaused(KEY)).toBe(true) + }) + + it('acknowledges a delete issued while paused and runs it once the owner answers', async () => { + const { queue, retry } = manualRetryQueue() + const log: string[] = [] + let reachable = false + await queue.submit(KEY, { + kind: 'sleep', + run: async () => false, + outcome: () => 'unverifiable', + observe: async () => { + if (!reachable) { + throw new Error('lost') + } + } + }) + const deleting = queue.submit(KEY, { + kind: 'delete', + run: async () => { + log.push('delete') + return 'deleted' + }, + ...deleteOptions('normal') + }) + await flush() + expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0, paused: true } }) + expect(log).toEqual([]) + reachable = true + retry() + expect(await deleting).toBe('deleted') + expect(queue.isPaused(KEY)).toBe(false) + }) + + it('runs a forced delete past the pause at once, in place of an unforced one waiting', async () => { + const { queue } = manualRetryQueue() + const log: string[] = [] + await queue.submit(KEY, { + kind: 'sleep', + run: async () => false, + outcome: () => 'unverifiable', + observe: () => new Promise(() => {}) + }) + const waiting = queue.submit(KEY, { + kind: 'delete', + run: async () => 'unforced', + ...deleteOptions('normal') + }) + await flush() + const forced = queue.submit(KEY, { + kind: 'delete', + run: async () => { + log.push('forced') + return 'deleted' + }, + ...deleteOptions('force', true) + }) + await expect(waiting).rejects.toThrow(WORKSPACE_DELETED_ERROR) + expect(await forced).toBe('deleted') + expect(log).toEqual(['forced']) + expect(queue.isPaused(KEY)).toBe(false) + }) + + it('starts commands in arrival order even when an earlier one is still looking up its workspace', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const lookup = deferred() + const first = queue.submitResolving(() => lookup.promise, { + kind: 'sleep', + run: async () => { + log.push('sleep (by name)') + return 'slept' + } + }) + const second = queue.submit(KEY, { + kind: 'wake', + run: async () => { + log.push('wake (by id)') + return 'woken' + } + }) + await flush() + expect(log).toEqual([]) + lookup.resolve(KEY) + expect(await Promise.all([first, second])).toEqual(['slept', 'woken']) + expect(log).toEqual(['sleep (by name)', 'wake (by id)']) + }) + + it('lets a lookup stuck on a dead host hold another workspace only briefly', async () => { + let clock = 0 + const queue = new WorkspaceLifecycleQueue( + async (ms) => { + clock += ms + }, + () => clock + ) + void queue.submitResolving(() => new Promise(() => {}), { + kind: 'sleep', + run: async () => 'never' + }) + const other = queue.submit('ssh:other|r::/w', { kind: 'wake', run: async () => 'woken' }) + expect(await other).toBe('woken') + expect(clock).toBe(500) + }) + + it('never cancels a command because its requester went away', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const sleep = gated(log, 'sleep', 'sleep', 'slept') + // The requester drops its reply (a phone disconnect); the command still runs to its end. + void queue.submit(KEY, sleep.command).catch(() => {}) + await flush() + sleep.release() + await flush() + expect(log).toEqual(['start sleep', 'end sleep']) + }) + + it('leaves nothing pending in a fresh queue, as after a restart', () => { + const queue = new WorkspaceLifecycleQueue() + expect(queue.hints()).toEqual({}) + expect(queue.creationsSnapshot().creations).toEqual({}) + }) +}) + +describe('WorkspaceLifecycleQueue creates', () => { + it('lists a creation once opened, names its workspace, and drops it on close', () => { + const queue = new WorkspaceLifecycleQueue() + const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' }) + expect(queue.creationsSnapshot().creations).toEqual({ c1: { repoId: 'repo-1' } }) + const opened = queue.creationsSnapshot().revision + creation.named('repo-1::/wt') + expect(queue.creationsSnapshot().creations).toEqual({ + c1: { repoId: 'repo-1', worktreeId: 'repo-1::/wt' } + }) + expect(queue.creationsSnapshot().revision).toBeGreaterThan(opened) + creation.close() + expect(queue.creationsSnapshot().creations).toEqual({}) + }) + + it('mints an id for a caller that sends none', () => { + const queue = new WorkspaceLifecycleQueue() + const creation = queue.openCreate({ repoId: 'repo-1' }) + const ids = Object.keys(queue.creationsSnapshot().creations) + expect(ids).toHaveLength(1) + expect(ids[0]).not.toBe('') + creation.close() + expect(queue.creationsSnapshot().creations).toEqual({}) + }) + + it('leaves a concurrent retry under the same id listed when the first create closes', () => { + const queue = new WorkspaceLifecycleQueue() + const first = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' }) + const second = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' }) + first.named('repo-1::/stale') + first.close() + expect(queue.creationsSnapshot().creations).toEqual({ c1: { repoId: 'repo-1' } }) + second.close() + expect(queue.creationsSnapshot().creations).toEqual({}) + }) + + it('frees the workspace queue when a create closes after a retry took over its id', async () => { + const queue = new WorkspaceLifecycleQueue() + const first = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY }) + first.named('repo-1::/wt') + const retry = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY }) + const sleeping = queue.submit(KEY, { kind: 'sleep', run: async () => 'slept' }) + first.close() + expect(await sleeping).toBe('slept') + retry.close() + }) + + it('closes once', () => { + const queue = new WorkspaceLifecycleQueue() + const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1' }) + creation.close() + const closed = queue.creationsSnapshot().revision + creation.close() + expect(queue.creationsSnapshot().revision).toBe(closed) + }) + + it('binds to the workspace once named, so commands on the Creating row wait behind it', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY }) + creation.named('repo-1::/wt') + expect(queue.hints()).toEqual({ [KEY]: { kind: 'create', queuedBehind: 0 } }) + const sleep = gated(log, 'sleep', 'sleep', 'slept') + const sleeping = queue.submit(KEY, sleep.command) + await flush() + expect(log).toEqual([]) + expect(queue.hints()).toEqual({ [KEY]: { kind: 'create', queuedBehind: 1 } }) + creation.close() + await flush() + expect(log).toEqual(['start sleep']) + sleep.release() + expect(await sleeping).toBe('slept') + }) + + it('lets a delete cancel what waits behind a create and run once the create ends', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY }) + creation.named('repo-1::/wt') + const sleeping = queue.submit(KEY, gated(log, 'sleep', 'sleep', 'slept').command) + const remove = gated(log, 'delete', 'delete', 'deleted') + const deleting = queue.submit(KEY, remove.command) + await expect(sleeping).rejects.toThrow(WORKSPACE_DELETED_ERROR) + await flush() + expect(log).toEqual([]) + creation.close() + await flush() + expect(log).toEqual(['start delete']) + remove.release() + expect(await deleting).toBe('deleted') + }) +}) + +describe('WorkspaceLifecycleQueue create fencing', () => { + it('never puts a new create behind the previous occupant’s delete', async () => { + const queue = new WorkspaceLifecycleQueue() + const log: string[] = [] + const sleep = gated(log, 'sleep', 'sleep', 'slept') + void queue.submit(KEY, sleep.command) + const deleting = queue.submit(KEY, gated(log, 'delete', 'delete', 'deleted').command) + const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY }) + creation.named('repo-1::/wt') + // The delete still waits on the sleep alone; the create is not one of its commands. + expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0 } }) + creation.close() + sleep.release() + await flush() + expect(log).toEqual(['start sleep', 'end sleep', 'start delete']) + void deleting + }) +}) + +describe('WorkspaceLifecycleQueue membership', () => { + /** A listing whose fresh scan the test applies by hand. */ + function listing() { + const waiters: (() => void)[] = [] + return { + membership: (done: () => void) => waiters.push(done), + applyFreshScan: () => waiters.splice(0).forEach((done) => done()) + } + } + + it('holds a finished create’s hint until a fresh listing, then drops it in that revision', () => { + const queue = new WorkspaceLifecycleQueue() + const rows = listing() + queue.setMembership(rows.membership) + const creation = queue.openCreate({ creationId: 'c1', repoId: 'repo-1', keyOf: () => KEY }) + creation.named('repo-1::/wt') + creation.close() + // A scan that started before the create ended cannot drop the row: the creation still shows. + expect(queue.creationsSnapshot().creations).toEqual({ + c1: { repoId: 'repo-1', worktreeId: 'repo-1::/wt' } + }) + expect(queue.hints()).toEqual({ [KEY]: { kind: 'create', queuedBehind: 0 } }) + const bumps: number[] = [] + queue.revision.onChange(() => bumps.push(queue.revision.value)) + rows.applyFreshScan() + expect(queue.creationsSnapshot().creations).toEqual({}) + expect(queue.hints()).toEqual({}) + // One synchronous bump, inside the listing's apply: the feed flushes both together. + expect(bumps).toHaveLength(1) + }) + + it('drops a failed create at once, with nothing to list', () => { + const queue = new WorkspaceLifecycleQueue() + const rows = listing() + queue.setMembership(rows.membership) + queue.openCreate({ creationId: 'c1', repoId: 'repo-1' }).close() + expect(queue.creationsSnapshot().creations).toEqual({}) + }) + + it('shows a finished delete until a fresh listing drops the row', async () => { + const queue = new WorkspaceLifecycleQueue() + const rows = listing() + queue.setMembership(rows.membership) + expect(await queue.submit(KEY, { kind: 'delete', run: async () => 'deleted' })).toBe('deleted') + expect(queue.hints()).toEqual({ [KEY]: { kind: 'delete', queuedBehind: 0 } }) + rows.applyFreshScan() + expect(queue.hints()).toEqual({}) + }) +}) diff --git a/src/main/runtime/workspaces/workspace-lifecycle-queue.ts b/src/main/runtime/workspaces/workspace-lifecycle-queue.ts new file mode 100644 index 00000000000..61990b369c7 --- /dev/null +++ b/src/main/runtime/workspaces/workspace-lifecycle-queue.ts @@ -0,0 +1,362 @@ +import { ProducerRevision } from '../sidebar-inputs/producer-revision' +import type { + WorkspaceLifecycleHint, + WorkspaceLifecycleKind +} from '../../../shared/sidebar-list/workspace-lifecycle-hints' +import type { WorkspaceCreationsSnapshot } from '../../../shared/sidebar-list/workspace-creations' +import { + WORKSPACE_DELETED_ERROR, + attemptLifecycleCommand, + type LifecycleAttempt, + type LifecycleCommand, + type LifecycleMembership +} from './workspace-lifecycle-command' +import { + WorkspaceLifecycleCreations, + type WorkspaceCreationHandle +} from './workspace-lifecycle-creations' + +type Entry = { + kind: WorkspaceLifecycleKind + /** Arrival order across every workspace, so a command waiting on its lookup keeps its place. */ + seq: number + state: 'queued' | 'running' + /** Null for a create, which runs outside the queue between open and close. */ + command: LifecycleCommand | null + reply: Promise | null + settle: (attempt: LifecycleAttempt) => void +} + +type Pause = { kind: WorkspaceLifecycleKind; observe: (() => Promise) | null } + +type WorkspaceQueue = { + entries: Entry[] + paused: Pause | null + /** A finished create or delete, held until a fresh listing shows its membership. */ + settling: number + draining: boolean + /** Ends a wait on the pause early: a forced delete arrived. */ + wake: (() => void) | null +} + +type Signal = { promise: Promise; notify: () => void } + +const OBSERVE_RETRY_MS = 5_000 +// Long enough for a lookup on a reachable host, so its command keeps its place in the order. +const LOOKUP_ORDER_WAIT_MS = 500 + +function signal(): Signal { + let notify!: () => void + const promise = new Promise((resolve) => (notify = resolve)) + return { promise, notify } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms).unref?.()) +} + +/** + * One FIFO queue per host-qualified workspace for create, sleep, wake, close and delete. Commands + * run one at a time in the order they arrived and are never merged. Delete cancels what has not + * started and waits for what has. An unconfirmed outcome pauses the queue until the owner answers + * again; main then re-observes and carries on, never replaying the paused command. Lives in + * memory: after a restart nothing is pending and readers show observed state. + */ +export class WorkspaceLifecycleQueue { + readonly revision = new ProducerRevision() + private readonly queues = new Map() + private membership: LifecycleMembership = (done) => done() + private nextSeq = 0 + /** + * Arrivals still looking up their workspace, by when they arrived. A later arrival waits for + * them only briefly: the lookup may name any workspace, and a dead host must not stall them all. + */ + private readonly resolving = new Map() + private resolved: Signal = signal() + private readonly creations = new WorkspaceLifecycleCreations({ + revision: this.revision, + membership: () => this.membership, + deletePending: (key) => this.deletePending(key), + bind: (key) => this.bindCreate(key), + finish: (key, entry) => this.finishEntry(key, entry) + }) + + constructor( + private readonly retryDelay: (ms: number) => Promise = sleep, + private readonly now: () => number = Date.now + ) {} + + /** How long `seq` still waits for an earlier arrival's lookup; 0 once none holds it. */ + private lookupWaitMs(seq: number): number { + const waits = [...this.resolving].flatMap(([earlier, arrivedAt]) => + earlier < seq ? [arrivedAt + LOOKUP_ORDER_WAIT_MS - this.now()] : [] + ) + return Math.max(0, ...waits) + } + + /** Where finished creates and deletes wait to be listed; until set they settle at once. */ + setMembership(membership: LifecycleMembership): void { + this.membership = membership + } + + submit(key: string, command: LifecycleCommand): Promise { + return this.enqueue(key, command, this.nextSeq++) + } + + /** + * For a command whose workspace is known only after a lookup: it takes its place on arrival. + * A lookup answering null finds no workspace to order against, so the command runs directly. + */ + async submitResolving( + resolveKey: () => Promise, + command: LifecycleCommand + ): Promise { + const seq = this.nextSeq++ + this.resolving.set(seq, this.now()) + let key: string | null + try { + key = await resolveKey() + } finally { + this.resolving.delete(seq) + const resolved = this.resolved + this.resolved = signal() + resolved.notify() + } + return key === null ? command.run() : this.enqueue(key, command, seq) + } + + openCreate( + start: Parameters['open']>[0] + ): WorkspaceCreationHandle { + return this.creations.open(start) + } + + /** The per-workspace hints the feed publishes. */ + hints(): Record { + const hints: Record = {} + for (const [key, queue] of this.queues) { + const deleting = queue.entries.some((entry) => entry.kind === 'delete') + const head = queue.entries[0] + const kind = deleting ? 'delete' : (head?.kind ?? queue.paused?.kind) + if (kind === undefined) { + // Only a settling delete is left: it shows until its row leaves the listing. + if (queue.settling > 0) { + hints[key] = { kind: 'delete', queuedBehind: 0 } + } + continue + } + hints[key] = { + kind, + queuedBehind: deleting ? 0 : Math.max(0, queue.entries.length - 1), + ...(queue.paused ? { paused: true as const } : {}) + } + } + for (const key of this.creations.settlingKeys()) { + hints[key] ??= { kind: 'create', queuedBehind: 0 } + } + return hints + } + + /** The creations projection the `creations` readers already read. */ + creationsSnapshot(): WorkspaceCreationsSnapshot { + return this.creations.snapshot() + } + + isPaused(key: string): boolean { + return this.queues.get(key)?.paused != null + } + + private enqueue(key: string, command: LifecycleCommand, seq: number): Promise { + const queue = this.queueFor(key) + const pendingDelete = queue.entries.find((entry) => entry.kind === 'delete') + if (pendingDelete) { + const pending = pendingDelete.command?.delete + if (command.kind !== 'delete') { + return Promise.reject(new Error(WORKSPACE_DELETED_ERROR)) + } + if (pendingDelete.reply && pending?.key === command.delete?.key) { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: an equal delete shares the pending delete's reply, as a removal in flight always has. + return pendingDelete.reply as Promise + } + // A forced delete replaces an unforced one still waiting (behind a pause); never a running one. + if (!command.delete?.force || pendingDelete.state === 'running') { + const label = command.delete?.label ?? key + return Promise.reject(new Error(`Worktree deletion already in progress: ${label}`)) + } + } + if (command.kind === 'delete') { + this.cancelQueued(queue) + } + let settle!: (attempt: LifecycleAttempt) => void + const reply = new Promise((resolve, reject) => { + settle = (result) => (result.threw ? reject(result.error) : resolve(result.value)) + }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: widened only to be stored; run's result is handed back to this command's own caller. + const stored = command as LifecycleCommand + const entry: Entry = { + kind: command.kind, + seq, + state: 'queued', + command: stored, + reply, + settle + } + // In arrival order among what has not started; a late lookup goes ahead of later arrivals. + const after = queue.entries.findIndex((queued) => queued.state === 'queued' && queued.seq > seq) + queue.entries.splice(after === -1 ? queue.entries.length : after, 0, entry) + this.revision.bump() + if (command.delete?.force) { + queue.wake?.() + } + void this.drain(key) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: settled only with this command's run result. + return reply as Promise + } + + private deletePending(key: string): boolean { + return this.queues.get(key)?.entries.some((entry) => entry.kind === 'delete') === true + } + + private queueFor(key: string): WorkspaceQueue { + let queue = this.queues.get(key) + if (!queue) { + queue = { entries: [], paused: null, settling: 0, draining: false, wake: null } + this.queues.set(key, queue) + } + return queue + } + + private bindCreate(key: string): Entry { + const queue = this.queueFor(key) + const entry: Entry = { + kind: 'create', + seq: this.nextSeq++, + state: 'running', + command: null, + reply: null, + settle: () => {} + } + // Ahead of everything not started: the create is already running. + const firstQueued = queue.entries.findIndex((queued) => queued.state === 'queued') + queue.entries.splice(firstQueued === -1 ? queue.entries.length : firstQueued, 0, entry) + return entry + } + + private finishEntry(key: string, entry: Entry): void { + const queue = this.queues.get(key) + const index = queue?.entries.indexOf(entry) ?? -1 + if (queue && index !== -1) { + queue.entries.splice(index, 1) + this.revision.bump() + void this.drain(key) + } + } + + private cancelQueued(queue: WorkspaceQueue): void { + const cancelled = queue.entries.filter((entry) => entry.state === 'queued') + queue.entries = queue.entries.filter((entry) => entry.state !== 'queued') + for (const entry of cancelled) { + entry.settle({ outcome: 'settled', error: new Error(WORKSPACE_DELETED_ERROR), threw: true }) + } + } + + private async drain(key: string): Promise { + const queue = this.queues.get(key) + if (!queue || queue.draining) { + return + } + queue.draining = true + try { + for (let head = queue.entries[0]; head; head = queue.entries[0]) { + if (head.state === 'running' || !head.command) { + // A create running outside the queue; its close drains again. + return + } + const waitMs = this.lookupWaitMs(head.seq) + if (waitMs > 0) { + await Promise.race([this.resolved.promise, this.retryDelay(waitMs)]) + continue + } + if (queue.paused && !head.command.delete?.force) { + await this.awaitOwner(queue, queue.paused) + continue + } + await this.runHead(key, queue, head, head.command) + } + } finally { + queue.draining = false + this.forgetIfIdle(key, queue) + } + } + + private async runHead( + key: string, + queue: WorkspaceQueue, + head: Entry, + command: LifecycleCommand + ): Promise { + head.state = 'running' + this.revision.bump() + const result = await attemptLifecycleCommand(command) + queue.entries.splice(queue.entries.indexOf(head), 1) + if (result.outcome === 'unverifiable') { + queue.paused = { kind: head.kind, observe: command.observe ?? null } + } else if (head.kind === 'delete' && !result.threw) { + // Deleted: nothing is left to observe. + queue.paused = null + queue.settling += 1 + this.membership(() => { + queue.settling -= 1 + this.revision.bump() + this.forgetIfIdle(key, queue) + }) + } + head.settle(result) + this.revision.bump() + } + + /** + * Waits for the paused command's owner to answer again, then re-observes the listing and lifts + * the pause. Returns early when a forced delete arrives; a silent owner is asked again later. + */ + private async awaitOwner(queue: WorkspaceQueue, paused: Pause): Promise { + if (!paused.observe) { + queue.paused = null + this.revision.bump() + return + } + const woken = new Promise<'woken'>((resolve) => (queue.wake = () => resolve('woken'))) + try { + const answered = await Promise.race([ + paused.observe().then( + () => true, + () => false + ), + woken + ]) + if (answered === true) { + await new Promise((done) => this.membership(done)) + if (queue.paused === paused) { + queue.paused = null + this.revision.bump() + } + } else if (answered === false) { + await Promise.race([this.retryDelay(OBSERVE_RETRY_MS), woken]) + } + } finally { + queue.wake = null + } + } + + private forgetIfIdle(key: string, queue: WorkspaceQueue): void { + if ( + queue.entries.length === 0 && + !queue.paused && + queue.settling === 0 && + !queue.draining && + this.queues.get(key) === queue + ) { + this.queues.delete(key) + } + } +} diff --git a/src/shared/sidebar-list/workspace-lifecycle-hints.ts b/src/shared/sidebar-list/workspace-lifecycle-hints.ts new file mode 100644 index 00000000000..8fb61c3dbf2 --- /dev/null +++ b/src/shared/sidebar-list/workspace-lifecycle-hints.ts @@ -0,0 +1,19 @@ +import { z } from 'zod' + +export const WORKSPACE_LIFECYCLE_KINDS = ['create', 'sleep', 'wake', 'close', 'delete'] as const + +export type WorkspaceLifecycleKind = (typeof WORKSPACE_LIFECYCLE_KINDS)[number] + +/** A lifecycle command main accepted for a workspace: a hint only, never saved, gone on restart. */ +const WorkspaceLifecycleHintSchema = z.object({ + kind: z.enum(WORKSPACE_LIFECYCLE_KINDS), + /** Commands accepted after this one and still waiting. */ + queuedBehind: z.number(), + /** The last command's outcome went unconfirmed by its host; the queue waits for proof. */ + paused: z.literal(true).optional() +}) + +/** Keyed by the workspace's host-qualified key (`hostId|id`). */ +export const WorkspaceLifecycleHintsSchema = z.record(z.string(), WorkspaceLifecycleHintSchema) + +export type WorkspaceLifecycleHint = z.infer