diff --git a/doc/plans/2026-11-07-online-git-source-fence.md b/doc/plans/2026-11-07-online-git-source-fence.md index 682725fcd97..91ede807b68 100644 --- a/doc/plans/2026-11-07-online-git-source-fence.md +++ b/doc/plans/2026-11-07-online-git-source-fence.md @@ -38,9 +38,9 @@ their existing reconstruction path. Preparation, lifecycle, runtime-service, expiration, and reopen owners retain the same physical keys before filesystem mutation or lifecycle exclusion. The importer holds all cohort task/project and lifecycle keys through publication. -Each cohort reserves its at most eight members in independent transactions in -parallel. It joins every reservation before starting capture or releasing the -source fence, including on failure. Partial journals keep the full exact member +Each cohort reserves its members in fully joined batches of at most eight +independent transactions. It joins every reservation before starting capture or +releasing the source fence, including on failure. Partial journals keep the full exact member and operation set for resume; publication stays serial in one transaction. It also holds the existing per-run physical finalizer keys for each exact native recovery proof. It re-inspects the complete source proof before capture and @@ -128,3 +128,28 @@ reacquires the source locks and captures current local bytes again. Read-only verifier callers do not opt into checkpoint publication. A change that strengthens immutable archive verification must advance the proof version so earlier verification cannot bypass the new checks. + + +A cohort with nine to 32 members is admitted only when every canonical resolved +physical root is exactly the same. Cohorts of at most eight members retain the +existing mixed-leaf rules. Source resolution, admission, reservation, recovery +checks, final readers, replay and cleanup remain bounded to eight peers per +stage. Capture concurrency remains at most eight, and all running peers join +before an error or cancellation can release the source fence. + +With shared initial and final child verification, 30 fresh aliases need one +initial and one final child pass instead of four pairs. For N distinct covered +children this changes 8N child checks to 2N. Thirty parent captures, uploads and +restores remain required. This is a work-count reduction; hosted wall-clock +improvement is not yet measured. Saved six- and eight-member cohorts retain their +exact membership. The canonical maintenance request accepts the same bounded +32-member list; single-workspace and capture-worker limits stay unchanged. +The journal shape is unchanged; older tools that enforce the +eight-member bound refuse larger cohorts and must not regroup them. + +Completed parent captures retain scratch until atomic publication. A larger +cohort can therefore retain up to four times the parent scratch of an +eight-member cohort, even with the same worker limit. Select a larger cohort +only after checking the actual parent capture footprint, its verification and +restore scratch, and current free capacity on the target filesystem. The worker +limit does not bound retained scratch. Keep the existing resource guards. diff --git a/server/src/maintenance/online-workspace-migration.test.ts b/server/src/maintenance/online-workspace-migration.test.ts index b68e8f56741..5ff91ad1dea 100644 --- a/server/src/maintenance/online-workspace-migration.test.ts +++ b/server/src/maintenance/online-workspace-migration.test.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { describe, expect, it, vi } from "vitest"; import { assertOnlineMigrationManifest, assertOnlineMigrationRuntime, assertOnlineMigrationSchema, onlineWorkspaceMigrationReceipt, parseOnlineWorkspaceMigrationRequest, runOnlineWorkspaceMigration } from "./online-workspace-migration.js"; import type { Db } from "@paperclipai/db"; @@ -52,7 +53,10 @@ describe("online workspace migration request", () => { it("refuses duplicate membership and bounds the exact cohort", () => { const cohort = { ...request(), mode: "cohort", concurrency: 2 }; expect(() => parseOnlineWorkspaceMigrationRequest({ ...cohort, workspaceIds: [cohort.workspaceIds[0], cohort.workspaceIds[0]] })).toThrow(); - expect(() => parseOnlineWorkspaceMigrationRequest({ ...cohort, workspaceIds: Array(9).fill(cohort.workspaceIds[0]) })).toThrow(); + const workspaceIds = Array.from({ length: 32 }, () => randomUUID()); + expect(parseOnlineWorkspaceMigrationRequest({ ...cohort, workspaceIds, concurrency: 8 }).workspaceIds).toHaveLength(32); + expect(() => parseOnlineWorkspaceMigrationRequest({ ...cohort, workspaceIds: [...workspaceIds, randomUUID()] })).toThrow(); + expect(() => parseOnlineWorkspaceMigrationRequest({ ...cohort, workspaceIds, concurrency: 9 })).toThrow(); }); }); diff --git a/server/src/maintenance/online-workspace-migration.ts b/server/src/maintenance/online-workspace-migration.ts index 5ff053d5be4..7459632cbc0 100644 --- a/server/src/maintenance/online-workspace-migration.ts +++ b/server/src/maintenance/online-workspace-migration.ts @@ -24,7 +24,7 @@ const requestSchema = z.object({ toolRoot: z.string().min(1).max(4096), source: z.object({ commit, deploymentId: z.string().uuid() }).strict(), companyId: z.string().uuid(), - workspaceIds: z.array(z.string().uuid()).min(1).max(8), + workspaceIds: z.array(z.string().uuid()).min(1).max(32), mode: z.enum(["inspect", "single", "cohort", "resume"]), operationId: z.string().uuid().optional(), snapshotId: z.string().uuid().optional(), diff --git a/server/src/services/remote-workspace-cohort-migration.test.ts b/server/src/services/remote-workspace-cohort-migration.test.ts index cc2c3ce9af4..719defc7b99 100644 --- a/server/src/services/remote-workspace-cohort-migration.test.ts +++ b/server/src/services/remote-workspace-cohort-migration.test.ts @@ -70,6 +70,15 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => } return { companyId, projectId, root, parent: members[0], leaves: members.slice(1) }; } + async function aliasMembers(f: Awaited>, base: typeof f.parent, count: number) { + const tasks = await db.insert(issues).values(Array.from({ length: count - 1 }, (_, index) => ({ + companyId: f.companyId, projectId: f.projectId, title: `Alias ${index}` }))).returning(); + const aliases = await db.insert(executionWorkspaces).values(tasks.map(task => ({ companyId: f.companyId, + projectId: f.projectId, sourceIssueId: task.id, name: "Alias", mode: base.mode, strategyType: base.strategyType, + providerType: base.providerType, cwd: base.cwd, providerRef: base.providerRef, repoUrl: base.repoUrl, branchName: base.branchName }))).returning(); + for (const alias of aliases) await db.update(issues).set({ executionWorkspaceId: alias.id }).where(eq(issues.id, alias.sourceIssueId!)); + return [base, ...aliases]; + } const run = (f: Awaited>, extra: Partial[0]> = {}) => migrateRemoteWorkspaceCohort({ db, storage, companyId: f.companyId, workspaceIds: f.leaves.map(row => row.id), apply: true, concurrency: 2, signal: AbortSignal.timeout(50_000), ...extra }); @@ -214,6 +223,123 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => await assertLocal(f); expect(await db.select().from(workspaceOperations)).toHaveLength(mode === "foreign" ? 3 : 2); }); + it.each(["fresh", "resume"] as const)("imports a larger same-root cohort with bounded work (%s)", async mode => { + const f = await fixture(), selected = await aliasMembers(f, f.parent, 10), workspaceIds = selected.map(row => row.id); + for (const child of f.leaves) await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id, + apply: true, signal: AbortSignal.timeout(50_000) }); + await fs.writeFile(path.join(f.root, "root-untracked.txt"), "parent retained work\n"); + const inspect = onlineWorkspaceSource.inspectOnlineWorkspaceSource, transaction = db.transaction.bind(db), put = storage.putFile.bind(storage); + let readers = 0, maximumReaders = 0, reservations = 0, maximumReservations = 0, uploads = 0, maximumUploads = 0; + vi.spyOn(onlineWorkspaceSource, "inspectOnlineWorkspaceSource").mockImplementation(async (...args) => { + readers++; maximumReaders = Math.max(maximumReaders, readers); + try { return await inspect(...args); } finally { readers--; } + }); + vi.spyOn(db, "transaction").mockImplementation(async (...args) => { + reservations++; maximumReservations = Math.max(maximumReservations, reservations); + try { return await transaction(...args); } finally { reservations--; } + }); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + let entered = 0, release!: () => void, interrupt = mode === "resume"; + let timer: ReturnType | undefined; + const barrier = new Promise(done => { release = done; }); + vi.spyOn(storage, "putFile").mockImplementation(async input => { + uploads++; maximumUploads = Math.max(maximumUploads, uploads); + try { + if (interrupt && ++entered === 2) throw new Error("large_cohort_upload_interrupted"); + if (!interrupt) { + if (++entered === 1) timer = setTimeout(() => release(), 5000); + if (entered === 8) release(); + await barrier; + } + return await put(input); + } finally { uploads--; } + }); + let savedOperations: Array<{ id: string; metadata: Record | null }> = [], savedSnapshotIds: string[] = []; + try { + if (mode === "resume") { + await expect(run(f, { workspaceIds, concurrency: 1 })).rejects.toThrow("large_cohort_upload_interrupted"); + savedOperations = (await db.select().from(workspaceOperations)).filter(row => workspaceIds.includes(row.executionWorkspaceId!)); + savedSnapshotIds = (await db.select().from(executionWorkspaceSnapshots)).filter(row => workspaceIds.includes(row.executionWorkspaceId)).map(row => row.id); + expect(savedOperations).toHaveLength(10); expect(savedSnapshotIds).toHaveLength(2); + interrupt = false; entered = 0; + } + await expect(run(f, { workspaceIds, concurrency: 8 })).resolves.toMatchObject({ applied: true }); + } finally { clearTimeout(timer); } + expect(maximumReaders).toBe(8); expect(maximumReservations).toBe(8); expect(maximumUploads).toBe(8); + expect(readers).toBe(0); expect(reservations).toBe(0); expect(uploads).toBe(0); + const operations = (await db.select().from(workspaceOperations)).filter(row => workspaceIds.includes(row.executionWorkspaceId!)); + expect(operations).toHaveLength(10); expect(operations.every(row => row.status === "succeeded")).toBe(true); + if (mode === "fresh") expect(capture.mock.calls.filter(([input]) => f.leaves.some(child => input.cwd === child.cwd) + && path.basename(input.archivePath) === "current.pcws")).toHaveLength(4); + else { + expect(operations.map(row => row.id).sort()).toEqual(savedOperations.map(row => row.id).sort()); + expect(operations.every(row => JSON.stringify(row.metadata!.cohortMigration) === JSON.stringify(savedOperations[0].metadata!.cohortMigration))).toBe(true); + expect((await db.select().from(executionWorkspaceSnapshots)).filter(row => savedSnapshotIds.includes(row.id))).toHaveLength(2); + } + await expect(run(f, { workspaceIds })).resolves.toMatchObject({ applied: true, replayed: true }); + await fs.rename(f.root, f.root + "-original"); + for (const workspace of selected) { + const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id)); + const ref = (await getWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: row.id, + snapshotId: readRemoteWorkspaceAuthority(row.metadata)!.snapshotId! }))!; + const archivePath = path.join(home, `${row.id}.pcws`), targetDir = path.join(home, `${row.id}-cold`); + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("committed base\n"); + expect(await fs.readFile(path.join(targetDir, "root-untracked.txt"), "utf8")).toBe("parent retained work\n"); + expect(await git(targetDir, "log", "-1", "--format=%s")).toBe("Original history"); + } + }, 90_000); + it.each(["failure", "cancel"] as const)("joins the first extended reader batch before stopping later members (%s)", async mode => { + const f = await fixture(), selected = await aliasMembers(f, f.parent, 10), ids = selected.map(row => row.id).sort(); + for (const child of f.leaves) await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id, + apply: true, signal: AbortSignal.timeout(50_000) }); + const operations = await db.select().from(workspaceOperations), snapshots = await db.select().from(executionWorkspaceSnapshots); + const inspect = onlineWorkspaceSource.inspectOnlineWorkspaceSource, calls = new Map(), started: string[] = []; + let bothEntered!: () => void, release!: () => void, failed!: () => void, settled = false; + const both = new Promise(done => { bothEntered = done; }), held = new Promise(done => { release = done; }); + const observed = new Promise(done => { failed = done; }), controller = new AbortController(); + vi.spyOn(onlineWorkspaceSource, "inspectOnlineWorkspaceSource").mockImplementation(async (...args) => { + const id = args[1].id, call = (calls.get(id) ?? 0) + 1; calls.set(id, call); + if (ids.includes(id) && call === 2) { + started.push(id); + if (id === ids[0]) { + await both; failed(); + if (mode === "failure") throw new Error("extended_reader_failure"); + controller.abort(new Error("extended_reader_cancelled")); + } else if (id === ids[1]) { bothEntered(); await held; } + } + return inspect(...args); + }); + const put = vi.spyOn(storage, "putFile"), other = createDb(database.connectionString, { maxConnections: 1 }); + const pending = run(f, { workspaceIds: ids, signal: controller.signal }) + .then(value => ({ value }), error => ({ error })).finally(() => { settled = true; }); + try { + await both; await observed; + expect(started).toContain(ids[0]); expect(started).toContain(ids[1]); + expect(started.every(id => ids.slice(0, 8).includes(id))).toBe(true); + expect(settled).toBe(false); + await expect(withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => true, true)) + .rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); + expect(put).not.toHaveBeenCalled(); + } finally { release(); await other.$client.end({ timeout: 1 }); } + expect(await pending).toMatchObject({ error: expect.objectContaining({ message: mode === "failure" + ? "extended_reader_failure" : "extended_reader_cancelled" }) }); + expect(started.sort()).toEqual(ids.slice(0, 8)); + expect(calls.get(ids[8])).toBe(1); expect(calls.get(ids[9])).toBe(1); + expect(await db.select().from(workspaceOperations)).toEqual(operations); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(snapshots); + expect(put).not.toHaveBeenCalled(); + }, 60_000); + it("refuses more than eight members across different canonical leaf roots", async () => { + const f = await fixture(), aliases = await aliasMembers(f, f.leaves[0], 8); + const workspaces = [f.leaves[1], ...aliases], noStorage = new Proxy(storage, { get() { throw new Error("storage_must_not_be_used"); } }); + await expect(run(f, { storage: noStorage, workspaceIds: workspaces.map(row => row.id) })) + .rejects.toThrow("workspace_cohort_same_root_required"); + expect(await db.select().from(workspaceOperations)).toHaveLength(0); + expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0); + await assertLocal(f); + }); it("resumes the same uploaded cohort after an archived companion is independently migrated", async () => { const f = await fixture("cohort"), held = await recoverable(f), cwd = path.join(f.root, ".paperclip", "worktrees", "retired"), closedAt = new Date(); await git(f.root, "worktree", "add", "-qb", "retired", cwd); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index f512dc5552d..5b9965abd6e 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -422,13 +422,12 @@ async function lockMigrationRecovery(lock: MigrationTransaction, plans: Migratio sources: online.flatMap(({ reserved, onlineSource }) => [...reserved.legacyRecoverySources, ...(onlineSource?.companions.flatMap(proof => proof.sources) ?? [])]), holds: online.flatMap(({ reserved, onlineSource }) => [...reserved.onlineRecoveryHolds, ...(onlineSource?.companions.flatMap(proof => proof.holds) ?? [])]) })) throw new Error("remote_workspace_migration_blocked:remote_copyback_owned"); - const readers = await Promise.allSettled(online.map(async ({ input, onlineSource, reserved }) => { + await mapCohortMembers(online, async ({ input, onlineSource, reserved }) => { const current = await inspect(input.db, { ...input, onlineSource, operationId: reserved.operation!.id }, cohort); if (current.reasons.length || canonical(current.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources) || canonical(current.onlineRecoveryHolds) !== canonical(reserved.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_recovery_source_changed"); - })); - for (const reader of readers) if (reader.status === "rejected") throw reader.reason; + }, online[0].input.signal); } function createMigrationChildVerifier(input: MigrationInput, onlineSource: OnlineWorkspaceSource | undefined, sourceLock: MigrationTransaction, cohort = false) { @@ -700,15 +699,29 @@ export async function migrateRemoteWorkspace(input: MigrationInput) { const cohortSchema = z.object({ schema: z.literal("paperclip.remote-workspace-cohort.v1"), id: z.string().uuid(), companyId: z.string().uuid(), sourceProof: z.string().regex(/^[a-f0-9]{64}$/), - members: z.array(z.object({ workspaceId: z.string().uuid(), operationId: z.string().uuid() }).strict()).min(1).max(8), + members: z.array(z.object({ workspaceId: z.string().uuid(), operationId: z.string().uuid() }).strict()).min(1).max(32), }).strict(); +async function mapCohortMembers(members: readonly T[], read: (member: T, index: number) => Promise, signal?: AbortSignal): Promise { + const results: R[] = [], cancellation = members.length > 8 ? signal : undefined; + for (let start = 0; start < members.length; start += 8) { + cancellation?.throwIfAborted(); + const readers = await Promise.allSettled(members.slice(start, start + 8).map((member, index) => read(member, start + index))); + for (const reader of readers) { + if (reader.status === "rejected") throw reader.reason; + results.push(reader.value); + } + cancellation?.throwIfAborted(); + } + return results; +} + export async function migrateRemoteWorkspaceCohort(input: { db: Db; storage: StorageService; companyId: string; workspaceIds: readonly string[]; apply?: boolean; concurrency?: number; signal: AbortSignal; sourceLockWaitMs?: number; }) { const companyId = z.string().uuid().parse(input.companyId).toLowerCase(); - const workspaceIds = z.array(z.string().uuid()).min(1).max(8).parse(input.workspaceIds).map(id => id.toLowerCase()).sort(); + const workspaceIds = z.array(z.string().uuid()).min(1).max(32).parse(input.workspaceIds).map(id => id.toLowerCase()).sort(); const concurrency = z.number().int().min(1).max(8).parse(input.concurrency ?? 2); if (new Set(workspaceIds).size !== workspaceIds.length) throw new Error("workspace_cohort_duplicate_member"); input.signal.throwIfAborted(); @@ -737,20 +750,22 @@ export async function migrateRemoteWorkspaceCohort(input: { || cohort.members.find(member => member.workspaceId === snapshot.executionWorkspaceId)?.operationId !== operation.id || canonical(operation.metadata?.cohortMigration) !== canonical(cohort))) throw new Error("workspace_cohort_membership_changed"); return { companyId, cohortId: cohort.id, applied: Boolean(input.apply), replayed: true, - members: await Promise.all(workspaceIds.map(workspaceId => inventoryRemoteWorkspaceMigration(input.db, { companyId, workspaceId }))) }; + members: await mapCohortMembers(workspaceIds, workspaceId => inventoryRemoteWorkspaceMigration(input.db, { companyId, workspaceId }), input.signal) }; } if (imported.some(Boolean)) throw new Error("workspace_cohort_membership_changed"); const stop = new AbortController(), signal = AbortSignal.any([input.signal, stop.signal]); - const members = await Promise.all(workspaceIds.map(async workspaceId => { + const members = await mapCohortMembers(workspaceIds, async workspaceId => { const memberInput: MigrationInput = { db: input.db, storage: input.storage, companyId, workspaceId, apply: true, signal }; const onlineSource = await resolveMigrationSource(memberInput); if (!onlineSource) throw new Error("workspace_cohort_online_source_required"); return { input: memberInput, onlineSource }; - })); + }, signal); const scopeIdentity = (scope: OnlineWorkspaceSource) => canonical({ kind: scope.kind, projectIds: scope.projectIds, issueIds: scope.issueIds, workspaceIds: scope.workspaceIds }); if (members.some(member => scopeIdentity(member.onlineSource) !== scopeIdentity(members[0].onlineSource))) throw new Error("workspace_cohort_source_scope_mismatch"); + if (members.length > 8 && members.some(member => member.onlineSource.root !== members[0].onlineSource.root)) + throw new Error("workspace_cohort_same_root_required"); for (const member of members) for (const other of members) { if (member !== other && member.onlineSource.root !== other.onlineSource.root && member.onlineSource.root.startsWith(other.onlineSource.root + path.sep)) throw new Error("workspace_cohort_leaf_members_required"); @@ -779,13 +794,12 @@ export async function migrateRemoteWorkspaceCohort(input: { || new Set(cohort.members.map(member => member.operationId)).size !== workspaceIds.length || pending.some(marker => canonical(marker) !== canonical(cohort))) throw new Error("workspace_cohort_membership_changed"); const preflight = async (db: ReadDb) => { - const readers = await Promise.allSettled(members.map(async member => { + await mapCohortMembers(members, async member => { const current = await inspect(db, { ...member.input, onlineSource: member.onlineSource }, cohort); if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); if (fingerprint(current.workspace) !== fingerprint(initial.find(workspace => workspace.id === member.input.workspaceId)!)) throw new Error("workspace_cohort_source_changed"); - })); - for (const reader of readers) if (reader.status === "rejected") throw reader.reason; + }, signal); }; if (!input.apply) { await preflight(input.db); return { companyId, cohortId: cohort.id, applied: false, eligible: true, workspaceIds }; } const publications: Parameters[0][] = []; @@ -795,16 +809,12 @@ export async function migrateRemoteWorkspaceCohort(input: { if (!await tryAcquireExecutionWorkspaceLifecycleLock(lock, workspaceId)) throw new IsolatedRepositorySourceBusy("workspace_source_busy"); } await preflight(input.db); - const reservations = await Promise.allSettled(members.map(async member => { + const plans: MigrationPlan[] = await mapCohortMembers(members, async member => { const reserved = await reserveMigration(member.input, member.onlineSource, cohort, fingerprint(initial.find(workspace => workspace.id === member.input.workspaceId)!)); if (reserved.authority) throw new Error("workspace_cohort_membership_changed"); return { ...member, reserved }; - })); - const plans: MigrationPlan[] = reservations.map(reservation => { - if (reservation.status === "rejected") throw reservation.reason; - return reservation.value; - }); + }, signal); await lockMigrationRecovery(lock, plans, cohort); if (refreshSourceProof) { const refreshedCohort = { ...cohort, sourceProof }; @@ -857,15 +867,14 @@ export async function migrateRemoteWorkspaceCohort(input: { for (const peer of peers) if (peer.status === "rejected") failure ??= peer.reason; if (failure) throw failure; signal.throwIfAborted(); - const nestedReaders = await Promise.allSettled(plans.map(async (plan, index) => { + await mapCohortMembers(plans, async (plan, index) => { const captured = captures[index]!; if (!captured.nested.length) return; const scratch = path.join(captured.scratch, "verify-nested"); await fs.mkdir(scratch); const verified = await (childVerifier?.finalize ?? prepareNestedWorkspaceImports)({ ...plan.input, workspace: plan.reserved.workspace, scratch, recordedChildren: captured.nested, gitSourceRoots: plan.onlineSource?.gitSourceRoots }); if (canonical(verified) !== canonical(captured.nested)) throw new Error("nested_workspace_source_changed"); - })); - for (const reader of nestedReaders) if (reader.status === "rejected") throw reader.reason; + }, signal); await lock.execute(sql`select 1`); await lock.transaction(async tx => { await lockMigrationParents(tx, companyId, null, plans.flatMap(plan => [ @@ -891,9 +900,9 @@ export async function migrateRemoteWorkspaceCohort(input: { } catch (error) { failure = error; throw error; } finally { await childVerifier?.close(); - const cleanup = await Promise.allSettled(captures.filter((capture): capture is CapturedMigration => Boolean(capture)) - .map(capture => removeTaskDirectoryTree(capture.scratch))); - const errors = cleanup.flatMap(outcome => outcome.status === "rejected" ? [outcome.reason] : []); + const cleanup = await mapCohortMembers(captures.filter((capture): capture is CapturedMigration => Boolean(capture)), + capture => removeTaskDirectoryTree(capture.scratch).then(() => [], error => [error])); + const errors = cleanup.flat(); if (errors.length) throw new AggregateError(failure ? [failure, ...errors] : errors, "workspace_cohort_cleanup_failed"); if (!failure) signal.throwIfAborted(); }