From 76486790e1ba0630f3022388f1fca53922407605 Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 13:13:00 -0700 Subject: [PATCH 01/12] Add abortable clone and storage preparation that run in process groups (#51 item 5) createTaskCloneAsync and prepareTaskFilesystemsAsync take an AbortSignal and run every Git or Docker call as the leader of its own process group (runInProcessGroup in agents/process-group.ts). Each group is reported through onProcessGroup in the same turn as the spawn, so F can save it before anything else runs. On abort or at the deadline the group gets SIGTERM, then SIGKILL after 5 s, and the call settles only once the leader and every other member of its group have exited. An aborted clone removes its partial directory; an aborted allocation removes what it created with uncancelled cleanup. The clone and allocation are now step generators shared by both variants: the synchronous helpers keep their API and behaviour, and long file walks pause regularly so the asynchronous variants keep the event loop free. Co-Authored-By: Claude Opus 5.5 --- agents/container/storage.ts | 204 +++++++++++++++++------- agents/process-group.ts | 129 +++++++++++++++ docs/implementation/agent-isolation.md | 6 +- docs/implementation/runner-lifecycle.md | 4 +- git/clone.ts | 130 +++++++++++---- test/agent-clone-async.test.ts | 82 ++++++++++ test/agent-container.test.ts | 24 ++- test/agent-process-group.test.ts | 79 +++++++++ test/agent-storage-async.test.ts | 128 +++++++++++++++ 9 files changed, 692 insertions(+), 94 deletions(-) create mode 100644 agents/process-group.ts create mode 100644 test/agent-clone-async.test.ts create mode 100644 test/agent-process-group.test.ts create mode 100644 test/agent-storage-async.test.ts diff --git a/agents/container/storage.ts b/agents/container/storage.ts index b6747eeb..cfe90b13 100644 --- a/agents/container/storage.ts +++ b/agents/container/storage.ts @@ -1,4 +1,4 @@ -import { execFileSync, spawnSync } from 'node:child_process'; +import { spawnSync } from 'node:child_process'; import { randomUUID } from 'node:crypto'; import { lstatSync, opendirSync, readlinkSync } from 'node:fs'; import { dirname, isAbsolute, join, relative } from 'node:path'; @@ -6,7 +6,8 @@ import type { TaskClone } from '../contract.ts'; import { assertTaskClone } from '../../git/clone.ts'; import { assertBuiltAgentImage } from './image.ts'; import { createOutcomeUnknown, DOCKER_ID } from '../client-outcome.ts'; -import { runDocker } from '../docker.ts'; +import { DockerError, pause, runDocker, type DockerOutcome } from '../docker.ts'; +import { runInProcessGroup, type ProcessGroup } from '../process-group.ts'; import { ALLOCATION_IN_USE, allocationListCommands, assertResourceOwner, claimAllocationId, hasOwnerLabels, ownerLabelArgs, releaseAllocationId, type ResourceOwner, UUID_V4 } from '../labels.ts'; @@ -55,33 +56,95 @@ const createDeadline = (timeoutMs: number) => { return value; }; }; -const docker = (args: readonly string[], timeoutMs: number) => execFileSync('docker', [...args], { - encoding: 'utf8', timeout: timeoutMs, killSignal: 'SIGKILL', env: dockerEnvironment(), - stdio: ['ignore', 'pipe', 'pipe'], -}).trim(); -const absent = (result: ReturnType) => result.status !== 0 && !result.error - && /No such (?:object|container|volume)/i.test(`${result.stdout ?? ''}\n${result.stderr ?? ''}`); +/** One Docker call a storage step needs run, a wait, or a point where a long walk lets other work (and an abort) in. */ +type StorageStep = { readonly args: readonly string[]; readonly timeoutMs: number; readonly cancellable: boolean } + | { readonly sleepMs: number } | typeof PAUSE; +type Steps = Generator; +const PAUSE = Symbol('pause'); +// Entries walked between pauses, so the asynchronous variant never holds the event loop for a whole checkout. +const PAUSE_EVERY = 1_000; +/** Ask the driver to run one Docker call. Allocation calls are cancellable; cleanup calls never are. */ +function* run(args: readonly string[], timeoutMs: number, cancellable: boolean): Steps { + return (yield { args, timeoutMs, cancellable })!; +} +/** Run one Docker call that must succeed; its failure throws a `DockerError` that `createOutcomeUnknown` reads. */ +function* must(args: readonly string[], timeoutMs: number): Steps { + const outcome = yield* run(args, timeoutMs, true); + if (outcome.status !== 0) throw new DockerError(args, outcome); + return outcome.stdout.trim(); +} + +// The blocking driver, for the synchronous API. +const sleep = (ms: number) => Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); +function runSteps(steps: Steps): T { + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) next = steps.next(); + else if ('sleepMs' in step) { sleep(step.sleepMs); next = steps.next(); } + else { + const result = spawnSync('docker', [...step.args], { encoding: 'utf8', timeout: step.timeoutMs, killSignal: 'SIGKILL', + env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'], maxBuffer: 16 * 1024 * 1024 }); + const error = result.error ?? (result.status === null + ? new Error(`docker ${step.args[0] ?? ''} was killed by ${result.signal}.`) : undefined); + next = steps.next({ status: error ? null : result.status, stdout: String(result.stdout ?? ''), + stderr: String(result.stderr ?? ''), ...(error ? { error } : {}) }); + } + } + return next.value; +} + +/** How the asynchronous variant runs its Docker calls. */ +export interface PreparationOptions { + /** Aborting stops the running allocation call's process group; cleanup of anything created still runs, uncancelled. */ + readonly signal?: AbortSignal; + /** Called in the same turn as each Docker spawn with its process group, so the caller can record it durably. */ + readonly onProcessGroup?: (group: ProcessGroup) => void; +} +// The non-blocking driver: every Docker call runs in its own process group, and the promise settles only after it has +// exited. An abort reaches the generator as an error at its next cancellable call or pause, so its own cleanup runs. +async function runStepsAsync(steps: Steps, options: PreparationOptions): Promise { + const aborted = () => Object.assign(new Error('Task storage allocation was cancelled.'), + { name: 'AbortError', code: 'ABORT_ERR' }); + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) { + await new Promise(resolve => setImmediate(resolve)); + next = options.signal?.aborted ? steps.throw(aborted()) : steps.next(); + } else if ('sleepMs' in step) { + await pause(step.sleepMs); + next = steps.next(); + } else { + next = steps.next(await runInProcessGroup('docker', step.args, { env: dockerEnvironment(), + timeoutMs: step.timeoutMs, signal: step.cancellable ? options.signal : undefined, + onProcessGroup: options.onProcessGroup })); + } + } + return next.value; +} + +const absent = (result: DockerOutcome) => result.status !== 0 && result.status !== null && !result.error + && /No such (?:object|container|volume)/i.test(`${result.stdout}\n${result.stderr}`); /** How long an object whose create client was killed may still materialize in the daemon. */ const CREATE_SETTLE_MS = 10_000; -const sleep = (ms: number) => Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); // Remove one storage object found by the name this allocation gave it, once it carries all three owner labels. A // container is then removed and confirmed by the ID just inspected, so a same-named replacement created after the // inspect is never touched. A volume's name is its only identity and Docker has no conditional remove, so a volume // replaced between the inspect and the remove could still be deleted; that needs someone to have deleted ours first. -const remove = (object: 'container' | 'volume', name: string, remaining: () => number, kind: string, - owner: ResourceOwner, settleBy = 0) => { - const inspectArgs = [object, 'inspect', name]; - let before: ReturnType; +// Cleanup is never cancelled: it runs to its own deadline so nothing is dropped. +function* remove(object: 'container' | 'volume', name: string, remaining: () => number, kind: string, + owner: ResourceOwner, settleBy = 0): Steps { + let before: DockerOutcome; for (;;) { - before = spawnSync('docker', [...inspectArgs], { encoding: 'utf8', timeout: remaining(), killSignal: 'SIGKILL', - env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); + before = yield* run([object, 'inspect', name], remaining(), false); if (before.status === 0) break; if (!absent(before)) throw new Error(`Failed to establish ownership of ${kind}.`); // A killed create may still land; only absence after the settle window counts. if (performance.now() >= settleBy) return; - sleep(250); + yield { sleepMs: 250 }; } - const inspected = JSON.parse(String(before.stdout || '[]'))[0] as + const inspected = JSON.parse(before.stdout || '[]')[0] as { Id?: string; Labels?: Record; Config?: { Labels?: Record } } | undefined; const labels = inspected?.Labels ?? inspected?.Config?.Labels; // All three owner labels must match: the allocation ID alone is caller-chosen and could be reused elsewhere. @@ -91,27 +154,26 @@ const remove = (object: 'container' | 'volume', name: string, remaining: () => n if (!inspected?.Id || !DOCKER_ID.test(inspected.Id)) throw new Error(`Failed to establish the identity of ${kind}.`); target = inspected.Id; } - const result = spawnSync('docker', object === 'container' ? ['rm', '--force', target] : ['volume', 'rm', '--force', target], - { encoding: 'utf8', timeout: remaining(), killSignal: 'SIGKILL', env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); + const result = yield* run(object === 'container' ? ['rm', '--force', target] : ['volume', 'rm', '--force', target], + remaining(), false); if (result.status === 0) return; - const inspect = spawnSync('docker', [object, 'inspect', target], { encoding: 'utf8', timeout: remaining(), - killSignal: 'SIGKILL', env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); - if (!absent(inspect)) throw new Error(`Failed to confirm removal of ${kind}.`); -}; -const cleanup = (containers: readonly string[], volumes: readonly string[], owner: ResourceOwner, - unsettled: ReadonlySet = new Set(), timeoutMs = 30_000) => { + if (!absent(yield* run([object, 'inspect', target], remaining(), false))) + throw new Error(`Failed to confirm removal of ${kind}.`); +} +function* cleanup(containers: readonly string[], volumes: readonly string[], owner: ResourceOwner, + unsettled: ReadonlySet = new Set(), timeoutMs = 30_000): Steps { const remaining = createDeadline(timeoutMs + (unsettled.size ? CREATE_SETTLE_MS : 0)), failures: unknown[] = []; const settleBy = (name: string) => unsettled.has(name) ? performance.now() + CREATE_SETTLE_MS : 0; for (const container of containers) { - try { remove('container', container, remaining, 'task container', owner, settleBy(container)); } + try { yield* remove('container', container, remaining, 'task container', owner, settleBy(container)); } catch (error) { failures.push(error); } } for (const volume of volumes) { - try { remove('volume', volume, remaining, 'task volume', owner, settleBy(volume)); } + try { yield* remove('volume', volume, remaining, 'task volume', owner, settleBy(volume)); } catch (error) { failures.push(error); } } if (failures.length) throw new AggregateError(failures, 'Task filesystem cleanup did not settle.'); -}; +} export function assertTaskFilesystems(filesystems: TaskFilesystems, clone?: TaskClone): void { const identity = allocations.get(filesystems); @@ -172,12 +234,13 @@ const linkStaysInside = (staging: string, link: string) => { * checkout (for example process environments that hold vendor credentials). Worktree links that stay inside, including * loops and not-yet-existing targets, are allowed. */ -const assertContainedLinks = (staging: string, remaining: () => number) => { +function* containedLinks(staging: string, remaining: () => number): Steps { const metadata = join(staging, '.git'), pending = [staging]; let count = 0; while (pending.length) { remaining(); count++; + if (count % PAUSE_EVERY === 0) yield PAUSE; const path = pending.pop()!, stat = lstatSync(path); if (stat.isSymbolicLink()) { const name = JSON.stringify(relative(staging, path)); @@ -198,27 +261,22 @@ const assertContainedLinks = (staging: string, remaining: () => number) => { } } finally { directory.closeSync(); } } -}; +} // Fails closed: an unanswered list cannot prove the ID is unused. Throws when more than `expected` objects carry it. -const assertAllocationObjects = (allocationId: string, expected: number, remaining: () => number) => { +function* allocationObjects(allocationId: string, expected: number, remaining: () => number): Steps { let found = 0; for (const command of allocationListCommands(allocationId)) { - const result = spawnSync('docker', [...command], { encoding: 'utf8', timeout: remaining(), killSignal: 'SIGKILL', - env: dockerEnvironment(), stdio: ['ignore', 'pipe', 'pipe'] }); + const result = yield* run(command, remaining(), true); if (result.status !== 0) throw new Error('Could not confirm that allocationId is unused.'); - found += String(result.stdout).split('\n').filter(line => line.trim()).length; + found += result.stdout.split('\n').filter(line => line.trim()).length; if (found > expected) throw new Error(ALLOCATION_IN_USE); } -}; +} -/** - * Copy a staging clone into bounded, engine-owned task storage and keep it mounted. `owner` is recorded by the caller - * before this call: every volume and container is labelled with its runner, attempt and allocation, so recovery after - * a crash finds exactly this storage. An allocation ID is used once; a reused ID is refused before anything is created. - */ -export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimits, - imageId: string, owner: ResourceOwner, timeoutMs = 60_000): TaskFilesystems { +/** The allocation, as a sequence of Docker calls. Both variants run these same steps; only how each call runs differs. */ +function* allocation(clone: TaskClone, limits: TaskStorageLimits, imageId: string, owner: ResourceOwner, + timeoutMs: number): Steps { owner = assertResourceOwner(owner); const labels = ownerLabelArgs(owner); for (const [name, value] of Object.entries(limits)) validLimit(value, name); @@ -227,10 +285,10 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi const staging = assertTaskClone(clone), remaining = createDeadline(timeoutMs); if (/[\n,]/.test(staging)) throw new Error('Staging path cannot be represented as a Docker mount.'); if (!lstatSync(`${staging}/.git`).isDirectory()) throw new Error('Staging clone must contain standalone Git metadata.'); - assertContainedLinks(staging, remaining); + yield* containedLinks(staging, remaining); const allocationId = claimAllocationId(owner.allocationId); // Nothing created yet: a reused ID (still labelling objects from any earlier process) is refused here. - try { assertAllocationObjects(allocationId, 0, remaining); } + try { yield* allocationObjects(allocationId, 0, remaining); } catch (error) { releaseAllocationId(allocationId); throw error; } // Live from before the first create: a failed allocation whose cleanup does not settle still owns what it made. liveAllocations.set(allocationId, owner.runnerOwner); @@ -240,13 +298,13 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi // create the daemon refused made nothing, and its name may belong to someone else. const made = new Set(), unsettled = new Set(); const mine = (names: readonly string[]) => names.filter(name => made.has(name) || unsettled.has(name)); - // Run one allocation step; a client killed by its deadline leaves the daemon outcome for `name` unknown. A pure - // create the daemon refused made nothing. A `docker run` is different: the daemon can create the container and then - // fail to start it, so after any failure of a run whose client started, `name` may exist and cleanup checks it. - const allocate = (name: string, args: readonly string[]) => { - const timeout = remaining(); + // Run one allocation step; a client killed by its deadline or an abort leaves the daemon outcome for `name` unknown. + // A pure create the daemon refused made nothing. A `docker run` is different: the daemon can create the container + // and then fail to start it, so after any failure of a run whose client started, `name` may exist and cleanup + // checks it. + function* allocate(name: string, args: readonly string[]): Steps { try { - const output = docker(args, timeout); + const output = yield* must(args, remaining()); made.add(name); return output; } catch (error) { @@ -254,16 +312,16 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi else if (args[0] === 'run' && typeof (error as { status?: unknown }).status === 'number') made.add(name); throw error; } - }; + } try { for (const [kind, name, bytes, inodes] of [['work', workVolume, limits.workBytes, limits.workInodes], ['metadata', metadataVolume, limits.metadataBytes, limits.metadataInodes]] as const) { - allocate(name, ['volume', 'create', '--driver', 'local', '--opt', 'type=tmpfs', '--opt', 'device=tmpfs', + yield* allocate(name, ['volume', 'create', '--driver', 'local', '--opt', 'type=tmpfs', '--opt', 'device=tmpfs', '--opt', `o=size=${bytes},nr_inodes=${inodes},uid=10001,gid=10001,mode=0755,nosuid,nodev`, '--label', `io.codeboost.task-storage=${kind}`, ...labels, name]); // The check above and this create are not atomic across processes. Once our first object exists, it must be // the only one carrying the ID: of two racing allocations, the later check sees both and backs out. - if (kind === 'work') assertAllocationObjects(allocationId, 1, remaining); + if (kind === 'work') yield* allocationObjects(allocationId, 1, remaining); } // Copy metadata straight to its own volume so the work allocation never holds both at once. const seed = ['set -eu', @@ -272,14 +330,14 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi 'cp -a --no-preserve=ownership,timestamps /run/codeboost-staging/.git/. /metadata/', 'mkdir -p /work/.git', 'chown -R 10001:10001 /work /metadata'].join('; '); // Create and start separately: once the create returns, the keeper is ours by ID even if its start fails. - const keeperId = allocate(keeper, ['create', '--name', keeper, '--read-only', '--user', '10001:10001', '--network=none', + const keeperId = yield* allocate(keeper, ['create', '--name', keeper, '--read-only', '--user', '10001:10001', '--network=none', '--cap-drop=ALL', '--security-opt=no-new-privileges', '--security-opt=seccomp=builtin', '--runtime=runc', '--pids-limit=32', '--memory=128m', '--cpus=.25', '--mount', `type=volume,source=${workVolume},target=/work`, '--mount', `type=volume,source=${metadataVolume},target=/metadata`, '--label', 'io.codeboost.task-storage=keeper', ...labels, '--entrypoint', 'sleep', imageId, 'infinity']); if (!DOCKER_ID.test(keeperId)) throw new Error('Docker did not return the created keeper ID.'); - docker(['start', keeperId], remaining()); - allocate(seeder, ['run', '--rm', '--name', seeder, '--label', 'io.codeboost.task-storage=seeder', ...labels, + yield* must(['start', keeperId], remaining()); + yield* allocate(seeder, ['run', '--rm', '--name', seeder, '--label', 'io.codeboost.task-storage=seeder', ...labels, '--read-only', '--user', '0:0', '--network=none', '--cap-drop=ALL', '--cap-add=CHOWN', '--cap-add=DAC_OVERRIDE', '--cap-add=FOWNER', '--security-opt=no-new-privileges', '--security-opt=seccomp=builtin', '--runtime=runc', '--pids-limit=32', '--memory=128m', '--cpus=.25', '--mount', `type=bind,source=${staging},target=/run/codeboost-staging,readonly`, @@ -294,7 +352,7 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi releaseAllocationId(allocationId); return filesystems; } catch (error) { - try { cleanup(mine([seeder, keeper]), mine([metadataVolume, workVolume]), owner, unsettled); } + try { yield* cleanup(mine([seeder, keeper]), mine([metadataVolume, workVolume]), owner, unsettled); } // Still live: this process owns whatever cleanup could not remove, and reports it through the error. catch (cleanupError) { throw new AggregateError([error, cleanupError], 'Task allocation failed and cleanup did not settle.'); } liveAllocations.delete(allocationId); @@ -303,6 +361,30 @@ export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimi } } +/** + * Copy a staging clone into bounded, engine-owned task storage and keep it mounted. `owner` is recorded by the caller + * before this call: every volume and container is labelled with its runner, attempt and allocation, so recovery after + * a crash finds exactly this storage. An allocation ID is used once; a reused ID is refused before anything is created. + * Blocks the event loop while it runs; the runner uses `prepareTaskFilesystemsAsync`. + */ +export function prepareTaskFilesystems(clone: TaskClone, limits: TaskStorageLimits, + imageId: string, owner: ResourceOwner, timeoutMs = 60_000): TaskFilesystems { + return runSteps(allocation(clone, limits, imageId, owner, timeoutMs)); +} + +/** + * `prepareTaskFilesystems` without blocking the event loop. Each Docker call runs in its own process group, reported + * through `onProcessGroup`. An abort stops the running call's group (SIGTERM, then SIGKILL after 5 s); the allocation + * then removes what it created, uncancelled and within its own cleanup deadline, and the promise rejects only after + * every group has exited. A create cut short by the abort counts as possibly created, as for a killed client. + */ +export async function prepareTaskFilesystemsAsync(clone: TaskClone, limits: TaskStorageLimits, imageId: string, + owner: ResourceOwner, options: PreparationOptions & { readonly timeoutMs?: number } = {}): Promise { + if (options.signal?.aborted) + throw Object.assign(new Error('Task storage allocation was cancelled.'), { name: 'AbortError', code: 'ABORT_ERR' }); + return runStepsAsync(allocation(clone, limits, imageId, owner, options.timeoutMs ?? 60_000), options); +} + /** * Task storage that `recoverLeftovers` found after a restart, when the `TaskFilesystems` value that allocated it is * gone. A crash during allocation can leave any subset of the parts, so each is optional. D issues a handle only after @@ -355,7 +437,7 @@ export async function adoptRecoveredTaskStorage(owner: ResourceOwner, parts: Tas const target = object === 'container' && keeperId ? keeperId : name; const inspect = await runDocker([object, 'inspect', target], { timeoutMs: remaining() }); if (inspect.status !== 0) { - if (absent(inspect as unknown as ReturnType)) + if (absent(inspect)) throw new RecoveredStorageRejected(`Recovered task ${kind} is gone.`); throw new Error(`Recovered task ${kind} could not be inspected.`); } @@ -392,15 +474,15 @@ export function removeTaskFilesystems(filesystems: TaskFilesystems | RecoveredTa const recovered = recoveredStorage.get(filesystems as RecoveredTaskStorage); if (recovered) { const handle = filesystems as RecoveredTaskStorage; - cleanup(recovered.keeperId ? [recovered.keeperId] : [], - [handle.metadataVolume, handle.workVolume].filter((name): name is string => name !== undefined), recovered.owner); + runSteps(cleanup(recovered.keeperId ? [recovered.keeperId] : [], + [handle.metadataVolume, handle.workVolume].filter((name): name is string => name !== undefined), recovered.owner)); recoveredStorage.delete(handle); return; } const allocated = filesystems as TaskFilesystems; assertTaskFilesystems(allocated); const owner = taskFilesystemOwner(allocated); - cleanup([allocated.keeper], [allocated.metadataVolume, allocated.workVolume], owner); + runSteps(cleanup([allocated.keeper], [allocated.metadataVolume, allocated.workVolume], owner)); allocations.delete(allocated); liveAllocations.delete(owner.allocationId); } diff --git a/agents/process-group.ts b/agents/process-group.ts new file mode 100644 index 00000000..9fb639a1 --- /dev/null +++ b/agents/process-group.ts @@ -0,0 +1,129 @@ +import { spawn } from 'node:child_process'; +import { NOT_STARTED, type DockerOutcome } from './docker.ts'; + +/** A subprocess's process group, reported as soon as it exists so the caller can record it durably. */ +export interface ProcessGroup { + /** The group ID, which is the leader's PID. */ + readonly pgid: number; + /** `Date.now()` right after the spawn; with the ID, it tells this group from a later one that reuses the ID. */ + readonly startedAt: number; +} +export interface ProcessGroupOptions { + readonly env: NodeJS.ProcessEnv; + readonly timeoutMs: number; + readonly cwd?: string; + /** Aborting stops the group (SIGTERM, then SIGKILL after the grace period); the call still settles only after exit. */ + readonly signal?: AbortSignal; + /** Called synchronously, in the same turn as the spawn, with the new group. */ + readonly onProcessGroup?: (group: ProcessGroup) => void; + /** How long a stopped group gets between SIGTERM and SIGKILL. Default 5 s. */ + readonly graceMs?: number; + /** Bytes kept of each of stdout and stderr; more stops the group. Default 16 MiB. */ + readonly maxBuffer?: number; +} + +const DEFAULT_GRACE_MS = 5_000; +const DEFAULT_MAX_BUFFER = 16 * 1024 * 1024; +// After the leader exits, how long to keep killing and polling other members of its group before giving up. +const DRAIN_LIMIT_MS = 10_000; +const pause = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); +const notStarted = (message: string): DockerOutcome => + ({ status: null, stdout: '', stderr: '', error: Object.assign(new Error(message), { code: NOT_STARTED }) }); + +// Signal every process in the group; false once the group no longer exists. +const signalGroup = (pgid: number, signal: NodeJS.Signals | 0) => { + try { process.kill(-pgid, signal); return true; } + catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ESRCH') return false; + throw error; + } +}; + +// The leader can exit while other members of its group still run (for example a child it forked). A group ID stays +// taken while any member exists, so until the group is empty the ID cannot name someone else's group. +const drainGroup = async (pgid: number) => { + const giveUpAt = performance.now() + DRAIN_LIMIT_MS; + while (signalGroup(pgid, 0)) { + signalGroup(pgid, 'SIGKILL'); + if (performance.now() >= giveUpAt) return false; + await pause(20); + } + return true; +}; + +/** + * Run one command as the leader of a new process group. The group is reported through `onProcessGroup` before this + * returns control to the event loop. On abort or at the deadline the whole group gets SIGTERM, then SIGKILL after the + * grace period, so a child that ignores SIGTERM is still bounded. The promise settles only after the leader has + * exited and every other member of its group is gone. Never rejects: failures are in the outcome, shaped like + * `runDocker`'s, so `status` is a number only when the command ran to completion. + */ +export function runInProcessGroup(file: string, args: readonly string[], + options: ProcessGroupOptions): Promise { + if (!Number.isSafeInteger(options.timeoutMs) || options.timeoutMs < 1) + return Promise.resolve(notStarted('Process deadline must be a positive integer.')); + if (options.signal?.aborted) return Promise.resolve(notStarted(`${file} was cancelled before it started.`)); + const graceMs = options.graceMs ?? DEFAULT_GRACE_MS, maxBuffer = options.maxBuffer ?? DEFAULT_MAX_BUFFER; + return new Promise(resolve => { + // `detached` makes the child the leader of a new process group (setsid), so signals to -pgid reach all it starts. + const child = spawn(file, [...args], { cwd: options.cwd, env: options.env, detached: true, + stdio: ['ignore', 'pipe', 'pipe'] }); + const pgid = child.pid; + if (pgid === undefined) { + // The spawn itself failed (for example ENOENT); nothing started, and 'error' carries the cause. + child.once('error', error => resolve({ status: null, stdout: '', stderr: '', error })); + return; + } + options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); + const out: Buffer[] = [], err: Buffer[] = []; + let outBytes = 0, errBytes = 0, stopped: 'cancelled' | 'timeout' | 'output-limit' | undefined; + let graceTimer: ReturnType | undefined; + const stop = (reason: NonNullable) => { + if (stopped) return; + stopped = reason; + signalGroup(pgid, 'SIGTERM'); + graceTimer = setTimeout(() => signalGroup(pgid, 'SIGKILL'), graceMs); + }; + const collect = (chunks: Buffer[], add: (bytes: number) => number) => (chunk: Buffer) => { + if (add(chunk.length) > maxBuffer) stop('output-limit'); + else chunks.push(chunk); + }; + child.stdout.on('data', collect(out, bytes => (outBytes += bytes))); + child.stderr.on('data', collect(err, bytes => (errBytes += bytes))); + const deadline = setTimeout(() => stop('timeout'), options.timeoutMs); + const onAbort = () => stop('cancelled'); + options.signal?.addEventListener('abort', onAbort, { once: true }); + let spawnError: Error | undefined; + child.once('error', error => { spawnError = error; }); + child.once('close', (code, signal) => { + clearTimeout(deadline); + options.signal?.removeEventListener('abort', onAbort); + void drainGroup(pgid).then(drained => { + clearTimeout(graceTimer); + const stdout = Buffer.concat(out).toString('utf8'), stderr = Buffer.concat(err).toString('utf8'); + if (!drained) { + resolve({ status: null, stdout, stderr, error: Object.assign( + new Error(`${file} left processes in its group that did not exit after SIGKILL.`), { code: 'EGROUPALIVE' }) }); + return; + } + if (stopped === 'cancelled') { + resolve({ status: null, stdout, stderr, error: Object.assign(new Error(`${file} ${args[0] ?? ''} was cancelled.`), + { name: 'AbortError', code: 'ABORT_ERR' }) }); + return; + } + if (stopped) { + resolve({ status: null, stdout, stderr, error: Object.assign(new Error(stopped === 'timeout' + ? `${file} ${args[0] ?? ''} ETIMEDOUT after ${options.timeoutMs} ms.` + : `${file} ${args[0] ?? ''} exceeded its output limit.`), { code: stopped === 'timeout' ? 'ETIMEDOUT' : 'ENOBUFS' }) }); + return; + } + if (spawnError || code === null) { + resolve({ status: null, stdout, stderr, + error: spawnError ?? Object.assign(new Error(`${file} ${args[0] ?? ''} was killed by ${signal}.`), { signal }) }); + return; + } + resolve({ status: code, stdout, stderr }); + }); + }); + }); +} diff --git a/docs/implementation/agent-isolation.md b/docs/implementation/agent-isolation.md index 0cfc7752..3186b6b7 100644 --- a/docs/implementation/agent-isolation.md +++ b/docs/implementation/agent-isolation.md @@ -57,8 +57,10 @@ metadata, read-only isolation and capacity probes had this defect. Use only these entry points to run an agent: -1. `createTaskClone` creates a committed, standalone staging clone. -2. `prepareTaskFilesystems` copies that clone into bounded task storage, labelled with the owner you pass: your +1. `createTaskClone` creates a committed, standalone staging clone. The runner uses `createTaskCloneAsync`, which + takes an `AbortSignal` and reports each Git process group through `onProcessGroup` (#51 item 5). +2. `prepareTaskFilesystems` (or `prepareTaskFilesystemsAsync`, which takes `signal` and `onProcessGroup` in the same + way) copies that clone into bounded task storage, labelled with the owner you pass: your runner token, the attempt ID, and an allocation ID (a lowercase UUID v4) you record first. Use each allocation ID once: before creating anything, D refuses an ID that another allocation in this process holds, or that any container, volume or network still carries (so a reuse after a restart is caught too). That check and the first diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index 70a44270..ae2fbde7 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -16,7 +16,7 @@ - the order of steps when the program shuts down; - the feedback events that the learning lane (J) will read. -**When implementation starts.** The plan starts F implementation after D5 merges, and D5 merged as #50. F1 code can start now. The attempt lifecycle itself uses only the D interface already on `main` (`agents/contract.ts`), which D4 (#47) did not change. Preparation, shutdown and startup recovery need D changes. Bounded settlement (#51 item 1) and asynchronous launch (#51 item 2) are done. Owner labels (#51 item 3) are done: `runnerOwner` on `InvocationInput`, and an owner on every allocator that runs before an invocation. Scoped recovery (#51 item 4) is done: `recoverLeftovers` (`agents/recovery.ts`) returns authenticated storage handles that `removeTaskFilesystems` accepts. This does **not** exist yet: asynchronous, abortable versions of both preparation helpers, `createTaskClone` and `prepareTaskFilesystems` (#51 item 5), listed as a prerequisite under "Shutdown". The F1 implementation may start against fakes, but it must not merge until #51's pre-F1 items land. +**When implementation starts.** The plan starts F implementation after D5 merges, and D5 merged as #50. F1 code can start now. The attempt lifecycle itself uses only the D interface already on `main` (`agents/contract.ts`), which D4 (#47) did not change. Preparation, shutdown and startup recovery need D changes. Bounded settlement (#51 item 1) and asynchronous launch (#51 item 2) are done. Owner labels (#51 item 3) are done: `runnerOwner` on `InvocationInput`, and an owner on every allocator that runs before an invocation. Scoped recovery (#51 item 4) is done: `recoverLeftovers` (`agents/recovery.ts`) returns authenticated storage handles that `removeTaskFilesystems` accepts. Abortable preparation (#51 item 5) is done: `createTaskCloneAsync` (`git/clone.ts`) and `prepareTaskFilesystemsAsync` (`agents/container/storage.ts`) take an `AbortSignal`, run every subprocess in its own process group, report each group through `onProcessGroup` in the same turn as the spawn, and settle only after it has exited. All of #51's pre-F1 items have landed. ## Summary @@ -260,7 +260,7 @@ This belongs to D (D5 or a D follow-up), not F. It changes `agents/contract.ts`, **Also before F1 merges: asynchronous launch.** Done in #51 item 2 (see "Launch setup does not block the server"). -**Third prerequisite before F1 merges: abortable preparation helpers.** Both preparation helpers are synchronous today. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. +**Third prerequisite before F1 merges: abortable preparation helpers.** Done in #51 item 5: `createTaskCloneAsync` and `prepareTaskFilesystemsAsync` take `signal` and `onProcessGroup`. Each subprocess runs as the leader of a new process group (`agents/process-group.ts`); `onProcessGroup` receives `{ pgid, startedAt }` synchronously after the spawn, so F can save it as `preparation_pgid` and `preparation_started_at` before anything else runs. Every Git or Docker call is a separate group, so F updates the saved group on each call. On abort or at the deadline the group gets `SIGTERM`, then `SIGKILL` after 5 seconds; the call settles only after the leader and every other member of its group have exited. An aborted clone removes its partial directory; an aborted storage allocation removes what it created, with cleanup that is not cancelled. The synchronous helpers remain for existing callers and share the same steps. What follows is the original requirement. Both preparation helpers were synchronous. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. **Prerequisite before F2's writable attempts: a bounded task-volume export.** D must add a bounded diff export over the task volumes (see "Partial output and interrupted rebases"). F1 does not need it: F1 runs only read-only attempts, and their task storage is removed without an export. diff --git a/git/clone.ts b/git/clone.ts index edece0c2..2b5a8bc5 100644 --- a/git/clone.ts +++ b/git/clone.ts @@ -3,6 +3,7 @@ import { randomUUID } from 'node:crypto'; import { lstatSync, mkdtempSync, opendirSync, realpathSync, rmSync } from 'node:fs'; import { isAbsolute, join, relative, resolve } from 'node:path'; import type { TaskClone } from '../agents/contract.ts'; +import { runInProcessGroup, type ProcessGroup } from '../agents/process-group.ts'; interface DirectoryIdentity { readonly dev: number; readonly ino: number } interface CloneIdentity { readonly directory: string; readonly root: DirectoryIdentity; readonly metadata: DirectoryIdentity } @@ -25,15 +26,29 @@ export function assertTaskClone(clone: TaskClone): string { return identity.directory; } +/** One Git call the clone steps need run, or a point where a long file walk lets other work (and an abort) in. */ +type CloneStep = { readonly cwd: string; readonly args: readonly string[]; readonly timeoutMs: number } | typeof PAUSE; +const PAUSE = Symbol('pause'); +/** What a Git call returned: `error` is set whenever it did not exit 0. */ +interface GitOutcome { readonly status: number | null; readonly stdout: string; readonly stderr: string; readonly error?: Error } +// Entries walked between pauses, so the asynchronous variant never holds the event loop for a whole object store. +const PAUSE_EVERY = 1_000; +const GIT_OPTIONS = ['--no-pager', '--no-replace-objects', '-c', 'core.hooksPath=/dev/null', '-c', 'init.templateDir=', + '-c', 'protocol.allow=never', '-c', 'submodule.recurse=false']; +// Deliberately do not inherit Git variables or credential/config environment. +const gitEnvironment = () => ({ PATH: process.env.PATH, GIT_CONFIG_NOSYSTEM: '1', GIT_CONFIG_GLOBAL: '/dev/null', + GIT_TERMINAL_PROMPT: '0', GIT_NO_LAZY_FETCH: '1', GIT_GRAFT_FILE: '/dev/null' }); + +export interface CloneOptions { + readonly source: string; readonly parent: string; readonly taskId: string; readonly head: string; + readonly timeoutMs?: number; +} + /** - * Prepare an independent committed snapshot. This is trusted staging, not the - * writable execution filesystem: D2 must reserve bounded storage and separate - * metadata before mounting it. Source must stay quiescent during this operation. - * No hooks, filters from user config, credentials, submodules or network access. + * The clone, as a sequence of Git calls. Both variants run these same steps; only how each call runs differs. Every + * failure after the clone directory exists, an abort included, removes it before the error leaves. */ -export function createTaskClone(options: { - source: string; parent: string; taskId: string; head: string; timeoutMs?: number; -}): TaskClone { +function* cloneSteps(options: CloneOptions): Generator { if (!options.taskId || options.taskId.includes('\0')) throw new Error('Task identity is required.'); if (!/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/.test(options.head)) throw new Error('A full committed head is required.'); const timeout = options.timeoutMs ?? 30_000; @@ -50,21 +65,15 @@ export function createTaskClone(options: { return rel === '' || (!isAbsolute(rel) && rel !== '..' && !rel.startsWith('../')); }; if (within(source, parent)) throw new Error('Task storage must be outside the source repository.'); - // Deliberately do not inherit Git variables or credential/config environment. - const env = { PATH: process.env.PATH, GIT_CONFIG_NOSYSTEM: '1', GIT_CONFIG_GLOBAL: '/dev/null', - GIT_TERMINAL_PROMPT: '0', GIT_NO_LAZY_FETCH: '1', GIT_GRAFT_FILE: '/dev/null' }; - const run = (cwd: string, ...args: string[]) => { - const result = execFileSync('git', [ - '--no-pager', '--no-replace-objects', '-c', 'core.hooksPath=/dev/null', '-c', 'init.templateDir=', - '-c', 'protocol.allow=never', '-c', 'submodule.recurse=false', ...args, - ], { cwd, env, timeout: remaining(), killSignal: 'SIGKILL', maxBuffer: 1024 * 1024, - stdio: ['ignore', 'pipe', 'pipe'] }); + function* run(cwd: string, ...args: string[]): Generator { + const outcome = yield { cwd, args, timeoutMs: remaining() }; + if (!outcome || outcome.status !== 0) throw outcome?.error ?? new Error(`git ${args[0] ?? ''} failed.`); remaining(); - return result.toString().trim(); - }; - const common = realpathSync(resolve(source, run(source, 'rev-parse', '--git-common-dir'))); + return outcome.stdout.trim(); + } + const common = realpathSync(resolve(source, yield* run(source, 'rev-parse', '--git-common-dir'))); if (within(common, parent)) throw new Error('Task storage must be outside source metadata.'); - function audit(metadata: string, independent: boolean) { + function* audit(metadata: string, independent: boolean): Generator { for (const name of ['shallow', 'info/grafts', 'objects/info/alternates', 'objects/info/http-alternates']) { if (lstatSync(join(metadata, name), { throwIfNoEntry: false })) throw new Error(`Unsupported Git storage: ${name}`); } @@ -73,6 +82,7 @@ export function createTaskClone(options: { while (pending.length) { remaining(); if (++count > 100_000) throw new Error('Object storage exceeds inspection limit.'); + if (count % PAUSE_EVERY === 0) yield PAUSE; const path = pending.pop()!, stat = lstatSync(path); if (stat.isSymbolicLink() || (!stat.isDirectory() && !stat.isFile())) throw new Error('Unsupported object entry.'); if (independent && stat.isFile() && stat.nlink !== 1) throw new Error('Task objects must not be hard-linked.'); @@ -88,18 +98,21 @@ export function createTaskClone(options: { } } } - audit(common, false); - if (run(source, 'for-each-ref', '--format=%(refname)', 'refs/replace')) throw new Error('Replacement objects are unsupported.'); - if (run(source, 'rev-parse', '--verify', `${options.head}^{commit}`) !== options.head) throw new Error('Head is not a commit.'); + yield* audit(common, false); + if (yield* run(source, 'for-each-ref', '--format=%(refname)', 'refs/replace')) + throw new Error('Replacement objects are unsupported.'); + if ((yield* run(source, 'rev-parse', '--verify', `${options.head}^{commit}`)) !== options.head) + throw new Error('Head is not a commit.'); const directory = mkdtempSync(join(parent, 'codeboost-task-')); try { - run(parent, '-c', 'protocol.file.allow=always', 'clone', '--local', '--no-hardlinks', '--no-checkout', '--', source, directory); + yield* run(parent, '-c', 'protocol.file.allow=always', 'clone', '--local', '--no-hardlinks', '--no-checkout', '--', + source, directory); const metadata = join(directory, '.git'); if (!lstatSync(metadata).isDirectory()) throw new Error('Task requires standalone Git metadata.'); - audit(metadata, true); - run(directory, 'remote', 'remove', 'origin'); - run(directory, 'checkout', '--detach', options.head); - if (run(directory, 'rev-parse', 'HEAD') !== options.head) throw new Error('Task head changed during clone.'); + yield* audit(metadata, true); + yield* run(directory, 'remote', 'remove', 'origin'); + yield* run(directory, 'checkout', '--detach', options.head); + if ((yield* run(directory, 'rev-parse', 'HEAD')) !== options.head) throw new Error('Task head changed during clone.'); const clone = Object.freeze({ id: randomUUID(), taskId: options.taskId, directory, head: options.head }); const root = directoryIdentity(directory), cloneMetadata = directoryIdentity(metadata); if (!root || !cloneMetadata) throw new Error('Task clone directory is not a real directory.'); @@ -110,3 +123,64 @@ export function createTaskClone(options: { throw error; } } + +/** + * Prepare an independent committed snapshot. This is trusted staging, not the + * writable execution filesystem: D2 must reserve bounded storage and separate + * metadata before mounting it. Source must stay quiescent during this operation. + * No hooks, filters from user config, credentials, submodules or network access. + * Blocks the event loop while it runs; the runner uses `createTaskCloneAsync`. + */ +export function createTaskClone(options: CloneOptions): TaskClone { + const env = gitEnvironment(), steps = cloneSteps(options); + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) { next = steps.next(); continue; } + let outcome: GitOutcome; + try { + const stdout = execFileSync('git', [...GIT_OPTIONS, ...step.args], { cwd: step.cwd, env, timeout: step.timeoutMs, + killSignal: 'SIGKILL', maxBuffer: 1024 * 1024, stdio: ['ignore', 'pipe', 'pipe'] }); + outcome = { status: 0, stdout: stdout.toString(), stderr: '' }; + } catch (error) { + const failed = error as { status?: number | null; stdout?: Buffer; stderr?: Buffer }; + outcome = { status: typeof failed.status === 'number' ? failed.status : null, + stdout: String(failed.stdout ?? ''), stderr: String(failed.stderr ?? ''), error: error as Error }; + } + next = steps.next(outcome); + } + return next.value; +} + +export interface AsyncCloneOptions extends CloneOptions { + /** Aborting stops the running Git call's process group and removes the partial clone before the promise rejects. */ + readonly signal?: AbortSignal; + /** Called in the same turn as each Git spawn with its process group, so the caller can record it durably. */ + readonly onProcessGroup?: (group: ProcessGroup) => void; +} + +/** + * `createTaskClone` without blocking the event loop. Each Git call runs in its own process group, reported through + * `onProcessGroup`; an abort sends the group SIGTERM, then SIGKILL after 5 s, and the promise settles only after the + * group has exited and the partial clone is removed. + */ +export async function createTaskCloneAsync(options: AsyncCloneOptions): Promise { + const env = gitEnvironment(), steps = cloneSteps(options), signal = options.signal; + const aborted = () => Object.assign(new Error('Task clone was cancelled.'), { name: 'AbortError', code: 'ABORT_ERR' }); + if (signal?.aborted) throw aborted(); + let next = steps.next(); + while (!next.done) { + const step = next.value; + if (step === PAUSE) { + await new Promise(resolve => setImmediate(resolve)); + next = signal?.aborted ? steps.throw(aborted()) : steps.next(); + continue; + } + const outcome = await runInProcessGroup('git', [...GIT_OPTIONS, ...step.args], { cwd: step.cwd, env, + timeoutMs: step.timeoutMs, signal, onProcessGroup: options.onProcessGroup, maxBuffer: 1024 * 1024 }); + const cancelled = (outcome.error as NodeJS.ErrnoException | undefined)?.code === 'ABORT_ERR'; + next = steps.next(outcome.status === 0 ? outcome : { ...outcome, error: cancelled ? aborted() : outcome.error + ?? new Error(`git ${step.args[0] ?? ''} failed (exit ${outcome.status}): ${outcome.stderr.trim().slice(0, 512)}`) }); + } + return next.value; +} diff --git a/test/agent-clone-async.test.ts b/test/agent-clone-async.test.ts new file mode 100644 index 00000000..9e1fb84f --- /dev/null +++ b/test/agent-clone-async.test.ts @@ -0,0 +1,82 @@ +import { execFileSync } from 'node:child_process'; +import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { assertTaskClone, createTaskCloneAsync } from '../git/clone.ts'; +import type { ProcessGroup } from '../agents/process-group.ts'; + +// The asynchronous clone (#51 item 5): every Git call runs in its own process group, and an abort settles only after +// that group has exited and the partial clone is gone. +const roots: string[] = []; +const git = (cwd: string, ...args: string[]) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', ...args], + { cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }).trim(); +function fixture() { + const root = mkdtempSync(join(tmpdir(), 'clone-async-')); roots.push(root); + const source = join(root, 'source'), parent = join(root, 'tasks'); + mkdirSync(source); mkdirSync(parent); + git(source, 'init'); git(source, 'config', 'user.name', 'Test'); git(source, 'config', 'user.email', 'test@example.com'); + writeFileSync(join(source, 'file.txt'), 'trusted\n'); git(source, 'add', '.'); git(source, 'commit', '-m', 'baseline'); + return { root, source, parent, head: git(source, 'rev-parse', 'HEAD'), taskId: 'task-1' }; +} +const groupAlive = (pgid: number) => { + try { process.kill(-pgid, 0); return true; } catch { return false; } +}; +/** Put a `git` first on PATH whose `clone` ignores SIGTERM, starts a background child, and never finishes. */ +const withStubbornClone = async (root: string, started: string, run: () => Promise) => { + const shim = join(root, 'bin'); mkdirSync(shim); + const realGit = execFileSync('sh', ['-c', 'command -v git'], { encoding: 'utf8' }).trim(); + writeFileSync(join(shim, 'git'), ['#!/bin/sh', + 'for arg in "$@"; do if [ "$arg" = clone ]; then', + ` trap '' TERM; sleep 60 & touch '${started}'; while :; do sleep 1; done`, + 'fi; done', + `exec '${realGit}' "$@"`].join('\n'), { mode: 0o755 }); + const path = process.env.PATH; + process.env.PATH = `${shim}:${path}`; + try { return await run(); } finally { process.env.PATH = path; } +}; +afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); + +describe('asynchronous task clone', () => { + it('clones like the synchronous helper and reports each Git process group, all gone once it settles', async () => { + const input = fixture(), groups: ProcessGroup[] = []; + const clone = await createTaskCloneAsync({ ...input, onProcessGroup: group => { groups.push(group); } }); + expect(assertTaskClone(clone)).toBe(clone.directory); + expect(readFileSync(join(clone.directory, 'file.txt'), 'utf8')).toBe('trusted\n'); + expect(git(clone.directory, 'rev-parse', 'HEAD')).toBe(input.head); + expect(groups.length).toBeGreaterThanOrEqual(5); + for (const group of groups) { + expect(group.pgid).toBeGreaterThan(1); + expect(groupAlive(group.pgid)).toBe(false); + } + }); + + it('on abort, kills a clone that ignores SIGTERM, waits for its group, and removes the partial clone', async () => { + const input = fixture(), started = join(input.root, 'clone-started'), groups: ProcessGroup[] = []; + const controller = new AbortController(); + // The event loop stays free while Git runs: this keeps ticking until the clone settles. + let ticks = 0; + const ticker = setInterval(() => { ticks += 1; }, 50); + const waitForClone = setInterval(() => { if (existsSync(started)) controller.abort(); }, 20); + const began = performance.now(); + try { + const error = await withStubbornClone(input.root, started, () => + createTaskCloneAsync({ ...input, timeoutMs: 60_000, signal: controller.signal, + onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught)); + expect((error as Error).message).toContain('cancelled'); + } finally { clearInterval(ticker); clearInterval(waitForClone); } + // SIGTERM was ignored, so settlement waited for the SIGKILL after the 5 s grace period. + expect(performance.now() - began).toBeGreaterThanOrEqual(5_000); + expect(ticks).toBeGreaterThan(50); + for (const group of groups) expect(groupAlive(group.pgid)).toBe(false); + expect(readdirSync(input.parent)).toEqual([]); + }, 30_000); + + it('spawns nothing and creates nothing when the signal is already aborted', async () => { + const input = fixture(), groups: ProcessGroup[] = []; + await expect(createTaskCloneAsync({ ...input, signal: AbortSignal.abort(), + onProcessGroup: group => { groups.push(group); } })).rejects.toThrow('cancelled'); + expect(groups).toEqual([]); + expect(readdirSync(input.parent)).toEqual([]); + }); +}); diff --git a/test/agent-container.test.ts b/test/agent-container.test.ts index f6c21bac..60e73161 100644 --- a/test/agent-container.test.ts +++ b/test/agent-container.test.ts @@ -12,7 +12,7 @@ import { createValidatedContainer, disposeValidatedContainer, prepareTaskFilesys startValidatedContainer, hasExactOptions, validateContainer } from '../agents/container/run.ts'; import { createTaskClone } from '../git/clone.ts'; import { hasOwnerLabels } from '../agents/labels.ts'; -import { isRecoveredTaskStorage, taskFilesystemOwner } from '../agents/container/storage.ts'; +import { isRecoveredTaskStorage, prepareTaskFilesystemsAsync, taskFilesystemOwner } from '../agents/container/storage.ts'; import { recoverLeftovers } from '../agents/recovery.ts'; import { createVendorNetwork, removeVendorNetwork, VendorNetworkCreationCleanupError, type VendorNetwork } from '../agents/network/network.ts'; @@ -715,6 +715,28 @@ describe('real Docker agent isolation', () => { expect(byAllocation(owner.allocationId)).toEqual([]); }, 60_000); + it('aborts storage allocation while a Docker call ignores SIGTERM, and removes what it created before settling', async () => { + const data = fixture(), owner = testOwner('abort-allocation'), marker = join(data.root, 'start-began'); + // The keeper really starts, then the client ignores SIGTERM and never returns. + const stubborn = [ + 'result = run(args);', + `require('node:fs').writeFileSync(${JSON.stringify(marker)}, '');`, + "process.on('SIGTERM', () => {});", + 'Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0);', + ].join('\n'); + const controller = new AbortController(), pgids: number[] = []; + const waitForStart = setInterval(() => { if (existsSync(marker)) controller.abort(); }, 20); + const began = performance.now(); + try { + await withDockerShim(['start'], stubborn, () => expect(prepareTaskFilesystemsAsync(data.clone, storageLimits, + imageId, owner, { signal: controller.signal, onProcessGroup: group => { pgids.push(group.pgid); } })) + .rejects.toThrow('cancelled')); + } finally { clearInterval(waitForStart); } + expect(performance.now() - began).toBeGreaterThanOrEqual(5_000); + expect(byAllocation(owner.allocationId)).toEqual([]); + for (const pgid of pgids) expect(spawnSync('kill', ['-0', `-${pgid}`]).status).not.toBe(0); + }, 90_000); + it('keeps a failed allocation whose cleanup did not settle live, so recovery in this process refuses its runner', async () => { const data = fixture(), runnerOwner = randomBytes(16).toString('hex'); const owner = { runnerOwner, attemptId: 'unsettled-allocation', allocationId: randomUUID() }; diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts new file mode 100644 index 00000000..330e9417 --- /dev/null +++ b/test/agent-process-group.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, it } from 'vitest'; +import { runInProcessGroup, type ProcessGroup } from '../agents/process-group.ts'; +import { NOT_STARTED } from '../agents/docker.ts'; +import { createOutcomeUnknown } from '../agents/client-outcome.ts'; + +// Preparation subprocesses run in their own process group, and settle only once the whole group has exited (#51 item 5). +const env = { PATH: process.env.PATH }; +const groupAlive = (pgid: number) => { + try { process.kill(-pgid, 0); return true; } catch { return false; } +}; + +describe('runInProcessGroup', () => { + it('reports the group in the same turn as the spawn, and returns a normal exit', async () => { + let reported: ProcessGroup | undefined; + const pending = runInProcessGroup('sh', ['-c', 'echo out; echo err >&2; exit 3'], + { env, timeoutMs: 5_000, onProcessGroup: group => { reported = group; } }); + // Synchronously, before anything else can run: the caller can record it before a crash could lose it. + expect(reported?.pgid).toBeGreaterThan(1); + expect(Math.abs(reported!.startedAt - Date.now())).toBeLessThan(5_000); + expect(await pending).toMatchObject({ status: 3, stdout: 'out\n', stderr: 'err\n' }); + expect(groupAlive(reported!.pgid)).toBe(false); + }); + + it('kills what the leader left in its group before settling, even after a normal exit', async () => { + let group: ProcessGroup | undefined; + const began = performance.now(); + const outcome = await runInProcessGroup('sh', ['-c', 'sleep 60 >/dev/null 2>&1 & echo started'], + { env, timeoutMs: 30_000, onProcessGroup: reported => { group = reported; } }); + expect(outcome).toMatchObject({ status: 0, stdout: 'started\n' }); + expect(groupAlive(group!.pgid)).toBe(false); + expect(performance.now() - began).toBeLessThan(10_000); + }); + + it('bounds a group that ignores SIGTERM: SIGKILL after the grace period, then settles as a timeout', async () => { + let group: ProcessGroup | undefined; + const began = performance.now(); + const outcome = await runInProcessGroup('sh', ['-c', "trap '' TERM; sleep 60 & while :; do sleep 1; done"], + { env, timeoutMs: 200, graceMs: 400, onProcessGroup: reported => { group = reported; } }); + const elapsed = performance.now() - began; + expect(outcome.status).toBeNull(); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ETIMEDOUT'); + expect(createOutcomeUnknown({ status: outcome.status, cause: outcome.error })).toBe(true); + expect(elapsed).toBeGreaterThanOrEqual(600); + expect(groupAlive(group!.pgid)).toBe(false); + }); + + it('stops the group on abort and reports it as cancelled', async () => { + const controller = new AbortController(); + let group: ProcessGroup | undefined; + setTimeout(() => controller.abort(), 100); + const outcome = await runInProcessGroup('sleep', ['60'], + { env, timeoutMs: 30_000, signal: controller.signal, onProcessGroup: reported => { group = reported; } }); + expect(outcome.status).toBeNull(); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ABORT_ERR'); + expect(groupAlive(group!.pgid)).toBe(false); + }); + + it('stops a group whose output exceeds the limit', async () => { + const outcome = await runInProcessGroup('sh', ['-c', 'while :; do echo xxxxxxxxxxxxxxxx; done'], + { env, timeoutMs: 30_000, maxBuffer: 1024 }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ENOBUFS'); + expect(outcome.stdout.length).toBeLessThanOrEqual(1024); + }); + + it('starts nothing for an aborted signal, an invalid deadline, or a missing program', async () => { + let reported = false; + const onProcessGroup = () => { reported = true; }; + for (const outcome of [ + await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 5_000, signal: AbortSignal.abort(), onProcessGroup }), + await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 0, onProcessGroup }), + ]) { + expect((outcome.error as NodeJS.ErrnoException).code).toBe(NOT_STARTED); + expect(createOutcomeUnknown({ status: outcome.status, cause: outcome.error })).toBe(false); + } + const missing = await runInProcessGroup('codeboost-no-such-program', [], { env, timeoutMs: 5_000, onProcessGroup }); + expect((missing.error as NodeJS.ErrnoException).code).toBe('ENOENT'); + expect(reported).toBe(false); + }); +}); diff --git a/test/agent-storage-async.test.ts b/test/agent-storage-async.test.ts new file mode 100644 index 00000000..b3430af2 --- /dev/null +++ b/test/agent-storage-async.test.ts @@ -0,0 +1,128 @@ +import { execFileSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { existsSync, mkdirSync, mkdtempSync, readdirSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ProcessGroup } from '../agents/process-group.ts'; + +// The asynchronous storage allocation (#51 item 5), against a fake Docker CLI that keeps its objects as files, so the +// abort path runs without a daemon. The real-Docker suite covers the same path against Docker itself. +vi.mock('../agents/container/image.ts', async importOriginal => ({ + ...await importOriginal(), assertBuiltAgentImage: () => {}, +})); +const { prepareTaskFilesystemsAsync, removeTaskFilesystems } = await import('../agents/container/storage.ts'); +const { createTaskClone } = await import('../git/clone.ts'); + +const RUNNER = '0123456789abcdef0123456789abcdef'; +const IMAGE = `sha256:${'a'.repeat(64)}`; +const LIMITS = { workBytes: 16 * 1024 * 1024, workInodes: 512, metadataBytes: 16 * 1024 * 1024, metadataInodes: 512 }; +let root = '', state = '', path: string | undefined; +// Objects are files named by name, holding { id, labels }. `docker start` hangs ignoring SIGTERM when told to. +const FAKE_DOCKER = String.raw`#!${process.execPath} +const fs = require('node:fs'), path = require('node:path'), crypto = require('node:crypto'); +const state = __STATE__, args = process.argv.slice(2); +const objects = () => fs.readdirSync(state).filter(name => name.endsWith('.json')) + .map(file => ({ name: file.slice(0, -5), ...JSON.parse(fs.readFileSync(path.join(state, file), 'utf8')) })); +const find = ref => objects().find(object => object.name === ref || object.id === ref); +const labels = () => Object.fromEntries(args.flatMap((arg, i) => arg === '--label' ? [args[i + 1].split(/=(.*)/s).slice(0, 2)] : [])); +const save = (name, object) => fs.writeFileSync(path.join(state, name + '.json'), JSON.stringify(object)); +const [a, b] = args; +if (a === 'ps' || b === 'ls') process.exit(0); +if (a === 'volume' && b === 'create') { save(args.at(-1), { kind: 'volume', labels: labels() }); console.log(args.at(-1)); process.exit(0); } +if (a === 'create') { + const id = crypto.randomBytes(32).toString('hex'); + save(args[args.indexOf('--name') + 1], { kind: 'container', id, labels: labels() }); console.log(id); process.exit(0); +} +if (a === 'start' && fs.existsSync(path.join(state, 'hang-start'))) { + fs.writeFileSync(path.join(state, 'start-began'), String(process.pid)); + process.on('SIGTERM', () => {}); + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0); +} +if (a === 'start') process.exit(0); +if (a === 'run') process.exit(0); +if (b === 'inspect') { + const object = find(args[2]); + if (!object) { console.error('Error: No such object: ' + args[2]); process.exit(1); } + console.log(JSON.stringify([object.kind === 'container' + ? { Id: object.id, Name: '/' + object.name, Config: { Labels: object.labels } } + : { Name: object.name, Labels: object.labels }])); + process.exit(0); +} +if (a === 'rm' || (a === 'volume' && b === 'rm')) { + const object = find(args.at(-1)); + if (object) fs.rmSync(path.join(state, object.name + '.json')); + process.exit(0); +} +console.error('fake docker: unsupported ' + args.join(' ')); process.exit(2); +`; + +const git = (cwd: string, ...args: string[]) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', ...args], + { cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }).trim(); +const clone = () => { + const source = join(root, `source-${randomUUID()}`), parent = join(root, `staging-${randomUUID()}`); + mkdirSync(source); mkdirSync(parent); + git(source, 'init'); git(source, 'config', 'user.name', 'Test'); git(source, 'config', 'user.email', 'test@example.com'); + writeFileSync(join(source, 'file.txt'), 'trusted\n'); git(source, 'add', '.'); git(source, 'commit', '-m', 'baseline'); + return createTaskClone({ source, parent, taskId: 'task-1', head: git(source, 'rev-parse', 'HEAD') }); +}; +const owner = () => ({ runnerOwner: RUNNER, attemptId: `attempt-${randomUUID()}`, allocationId: randomUUID() }); +const stored = () => readdirSync(state).filter(name => name.endsWith('.json')); +const groupAlive = (pgid: number) => { + try { process.kill(-pgid, 0); return true; } catch { return false; } +}; + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'storage-async-')); state = join(root, 'state'); mkdirSync(state); + const bin = join(root, 'bin'); mkdirSync(bin); + // Storage passes Docker only an allowlisted environment, so the fake's state path is written into the script. + writeFileSync(join(bin, 'docker'), FAKE_DOCKER.replace('__STATE__', JSON.stringify(state)), { mode: 0o755 }); + path = process.env.PATH; + process.env.PATH = `${bin}:${path}`; +}); +afterEach(() => { + process.env.PATH = path; + rmSync(root, { recursive: true, force: true }); +}); + +describe('asynchronous task storage allocation', () => { + it('allocates the same storage as the synchronous helper, reporting every Docker process group', async () => { + const groups: ProcessGroup[] = []; + const filesystems = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { onProcessGroup: group => { groups.push(group); } }); + expect(stored().sort()).toEqual([`${filesystems.keeper}.json`, `${filesystems.metadataVolume}.json`, + `${filesystems.workVolume}.json`].sort()); + expect(groups.length).toBeGreaterThanOrEqual(8); + for (const group of groups) expect(groupAlive(group.pgid)).toBe(false); + removeTaskFilesystems(filesystems); + expect(stored()).toEqual([]); + }); + + it('on abort, kills a Docker call that ignores SIGTERM, then removes everything it created before settling', async () => { + writeFileSync(join(state, 'hang-start'), ''); + const groups: ProcessGroup[] = [], controller = new AbortController(); + let ticks = 0; + const ticker = setInterval(() => { ticks += 1; }, 50); + const waitForStart = setInterval(() => { if (existsSync(join(state, 'start-began'))) controller.abort(); }, 20); + const began = performance.now(); + try { + const error = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { signal: controller.signal, onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught); + expect((error as Error).message).toContain('cancelled'); + } finally { clearInterval(ticker); clearInterval(waitForStart); } + // SIGTERM was ignored, so the call was killed only after the 5 s grace period. + expect(performance.now() - began).toBeGreaterThanOrEqual(5_000); + expect(ticks).toBeGreaterThan(50); + // The volumes and the keeper it created are gone, and no group it started is still running. + expect(stored()).toEqual([]); + for (const group of groups) expect(groupAlive(group.pgid)).toBe(false); + }, 30_000); + + it('creates nothing when the signal is already aborted', async () => { + const groups: ProcessGroup[] = []; + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { signal: AbortSignal.abort(), onProcessGroup: group => { groups.push(group); } })).rejects.toThrow('cancelled'); + expect(groups).toEqual([]); + expect(stored()).toEqual([]); + }); +}); From 357e1495bf873ee8beff03d76390af42c2875a14 Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 13:25:45 -0700 Subject: [PATCH 02/12] Kill a group the caller cannot record; never throw from group signals; report early aborts as cancelled runInProcessGroup ran onProcessGroup unguarded, so a throwing callback left the child running and unwatched. It now kills the group at once and settles only after it exits, with EUNRECORDED and the callback's error as the cause. signalGroup no longer throws (only ESRCH means gone), so a group member that cannot be signalled cannot crash the runner from a timer or hang the drain. An allocation ID check whose client was stopped or never started now reports why (for example cancelled) instead of a generic message. The real-Docker abort test checks groups with process.kill, which the kill utility cannot stand in for. Co-Authored-By: Claude Opus 5.5 --- agents/container/storage.ts | 2 ++ agents/process-group.ts | 24 +++++++++++++++++------- test/agent-container.test.ts | 4 +++- test/agent-process-group.test.ts | 10 ++++++++++ test/agent-storage-async.test.ts | 15 +++++++++++++++ 5 files changed, 47 insertions(+), 8 deletions(-) diff --git a/agents/container/storage.ts b/agents/container/storage.ts index cfe90b13..7db1682c 100644 --- a/agents/container/storage.ts +++ b/agents/container/storage.ts @@ -268,6 +268,8 @@ function* allocationObjects(allocationId: string, expected: number, remaining: ( let found = 0; for (const command of allocationListCommands(allocationId)) { const result = yield* run(command, remaining(), true); + // A client that was stopped (an abort or its deadline) or never started says why; a daemon answer does not. + if (result.status === null) throw new DockerError(command, result); if (result.status !== 0) throw new Error('Could not confirm that allocationId is unused.'); found += result.stdout.split('\n').filter(line => line.trim()).length; if (found > expected) throw new Error(ALLOCATION_IN_USE); diff --git a/agents/process-group.ts b/agents/process-group.ts index 9fb639a1..6ab320ed 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -30,13 +30,12 @@ const pause = (ms: number) => new Promise(resolve => setTimeout(resolve, m const notStarted = (message: string): DockerOutcome => ({ status: null, stdout: '', stderr: '', error: Object.assign(new Error(message), { code: NOT_STARTED }) }); -// Signal every process in the group; false once the group no longer exists. +// Signal every process in the group; false once the group no longer exists. Never throws: it runs in timers and +// listeners, where a throw would crash the runner. Any failure other than ESRCH (such as EPERM, for a member that can +// no longer be signalled) counts as still alive, so draining gives up at its limit and reports it. const signalGroup = (pgid: number, signal: NodeJS.Signals | 0) => { try { process.kill(-pgid, signal); return true; } - catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ESRCH') return false; - throw error; - } + catch (error) { return (error as NodeJS.ErrnoException).code !== 'ESRCH'; } }; // The leader can exit while other members of its group still run (for example a child it forked). A group ID stays @@ -74,9 +73,13 @@ export function runInProcessGroup(file: string, args: readonly string[], child.once('error', error => resolve({ status: null, stdout: '', stderr: '', error })); return; } - options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); + // If the caller cannot record the group, the child must not outlive this call: it is killed at once below, and + // the call still settles only after the group has exited, with the caller's error. + let unrecorded: unknown; + try { options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); } + catch (error) { unrecorded = error; } const out: Buffer[] = [], err: Buffer[] = []; - let outBytes = 0, errBytes = 0, stopped: 'cancelled' | 'timeout' | 'output-limit' | undefined; + let outBytes = 0, errBytes = 0, stopped: 'cancelled' | 'timeout' | 'output-limit' | 'unrecorded' | undefined; let graceTimer: ReturnType | undefined; const stop = (reason: NonNullable) => { if (stopped) return; @@ -95,6 +98,7 @@ export function runInProcessGroup(file: string, args: readonly string[], options.signal?.addEventListener('abort', onAbort, { once: true }); let spawnError: Error | undefined; child.once('error', error => { spawnError = error; }); + if (unrecorded !== undefined) { stopped = 'unrecorded'; signalGroup(pgid, 'SIGKILL'); } child.once('close', (code, signal) => { clearTimeout(deadline); options.signal?.removeEventListener('abort', onAbort); @@ -106,6 +110,12 @@ export function runInProcessGroup(file: string, args: readonly string[], new Error(`${file} left processes in its group that did not exit after SIGKILL.`), { code: 'EGROUPALIVE' }) }); return; } + if (stopped === 'unrecorded') { + resolve({ status: null, stdout, stderr, error: Object.assign(new Error( + `${file} was stopped because its process group could not be recorded.`, { cause: unrecorded }), + { code: 'EUNRECORDED' }) }); + return; + } if (stopped === 'cancelled') { resolve({ status: null, stdout, stderr, error: Object.assign(new Error(`${file} ${args[0] ?? ''} was cancelled.`), { name: 'AbortError', code: 'ABORT_ERR' }) }); diff --git a/test/agent-container.test.ts b/test/agent-container.test.ts index 60e73161..cff086c4 100644 --- a/test/agent-container.test.ts +++ b/test/agent-container.test.ts @@ -734,7 +734,9 @@ describe('real Docker agent isolation', () => { } finally { clearInterval(waitForStart); } expect(performance.now() - began).toBeGreaterThanOrEqual(5_000); expect(byAllocation(owner.allocationId)).toEqual([]); - for (const pgid of pgids) expect(spawnSync('kill', ['-0', `-${pgid}`]).status).not.toBe(0); + // process.kill with a negative ID signals the group; the kill utility would read "-" as an option. + const groupAlive = (pgid: number) => { try { process.kill(-pgid, 0); return true; } catch { return false; } }; + for (const pgid of pgids) expect(groupAlive(pgid)).toBe(false); }, 90_000); it('keeps a failed allocation whose cleanup did not settle live, so recovery in this process refuses its runner', async () => { diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts index 330e9417..e4381a93 100644 --- a/test/agent-process-group.test.ts +++ b/test/agent-process-group.test.ts @@ -55,6 +55,16 @@ describe('runInProcessGroup', () => { expect(groupAlive(group!.pgid)).toBe(false); }); + it('kills the group and still waits for it when the caller cannot record it', async () => { + let group: ProcessGroup | undefined; + const outcome = await runInProcessGroup('sh', ['-c', "trap '' TERM; sleep 60 & while :; do sleep 1; done"], + { env, timeoutMs: 30_000, onProcessGroup: reported => { group = reported; throw new Error('database is locked'); } }); + expect(outcome.status).toBeNull(); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('EUNRECORDED'); + expect(((outcome.error as Error).cause as Error).message).toBe('database is locked'); + expect(groupAlive(group!.pgid)).toBe(false); + }); + it('stops a group whose output exceeds the limit', async () => { const outcome = await runInProcessGroup('sh', ['-c', 'while :; do echo xxxxxxxxxxxxxxxx; done'], { env, timeoutMs: 30_000, maxBuffer: 1024 }); diff --git a/test/agent-storage-async.test.ts b/test/agent-storage-async.test.ts index b3430af2..9caf39bf 100644 --- a/test/agent-storage-async.test.ts +++ b/test/agent-storage-async.test.ts @@ -28,6 +28,10 @@ const find = ref => objects().find(object => object.name === ref || object.id == const labels = () => Object.fromEntries(args.flatMap((arg, i) => arg === '--label' ? [args[i + 1].split(/=(.*)/s).slice(0, 2)] : [])); const save = (name, object) => fs.writeFileSync(path.join(state, name + '.json'), JSON.stringify(object)); const [a, b] = args; +if ((a === 'ps' || b === 'ls') && fs.existsSync(path.join(state, 'hang-ls'))) { + fs.writeFileSync(path.join(state, 'ls-began'), ''); + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0); +} if (a === 'ps' || b === 'ls') process.exit(0); if (a === 'volume' && b === 'create') { save(args.at(-1), { kind: 'volume', labels: labels() }); console.log(args.at(-1)); process.exit(0); } if (a === 'create') { @@ -118,6 +122,17 @@ describe('asynchronous task storage allocation', () => { for (const group of groups) expect(groupAlive(group.pgid)).toBe(false); }, 30_000); + it('reports an abort during the allocation ID check as cancelled, having created nothing', async () => { + writeFileSync(join(state, 'hang-ls'), ''); + const controller = new AbortController(); + const waitForList = setInterval(() => { if (existsSync(join(state, 'ls-began'))) controller.abort(); }, 20); + try { + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), { signal: controller.signal })) + .rejects.toThrow('cancelled'); + } finally { clearInterval(waitForList); } + expect(stored()).toEqual([]); + }, 30_000); + it('creates nothing when the signal is already aborted', async () => { const groups: ProcessGroup[] = []; await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), From c913259a42ac6646dfca48d32e22b01a1e302587 Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 13:30:49 -0700 Subject: [PATCH 03/12] Catch an abort raised during onProcessGroup; settle on the leader's exit, not its pipes An abort raised inside the onProcessGroup callback came before the listener existed and was never seen, so the child ran to its deadline. The helper now checks the signal right after attaching the listener. Settlement waited for 'close', which also needs every holder of the output pipes; a process that left the group with setsid but kept the pipe survived the group kill and hung the call forever. It now settles from the leader's exit: drain the group, wait up to 1 s for the pipes, then destroy them. Co-Authored-By: Claude Opus 5.5 --- agents/process-group.ts | 26 +++++++++++++++++++++----- test/agent-process-group.test.ts | 31 +++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/agents/process-group.ts b/agents/process-group.ts index 6ab320ed..bf4fbd8e 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -26,6 +26,9 @@ const DEFAULT_GRACE_MS = 5_000; const DEFAULT_MAX_BUFFER = 16 * 1024 * 1024; // After the leader exits, how long to keep killing and polling other members of its group before giving up. const DRAIN_LIMIT_MS = 10_000; +// Once the group is gone, how long to wait for stdout and stderr to reach end of file. Only a process outside the group +// (one that called setsid but kept the inherited pipe) can hold them open after that, and it must not hold this call. +const STDIO_CLOSE_MS = 1_000; const pause = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); const notStarted = (message: string): DockerOutcome => ({ status: null, stdout: '', stderr: '', error: Object.assign(new Error(message), { code: NOT_STARTED }) }); @@ -96,13 +99,23 @@ export function runInProcessGroup(file: string, args: readonly string[], const deadline = setTimeout(() => stop('timeout'), options.timeoutMs); const onAbort = () => stop('cancelled'); options.signal?.addEventListener('abort', onAbort, { once: true }); - let spawnError: Error | undefined; - child.once('error', error => { spawnError = error; }); if (unrecorded !== undefined) { stopped = 'unrecorded'; signalGroup(pgid, 'SIGKILL'); } - child.once('close', (code, signal) => { + // An abort raised during onProcessGroup came before the listener existed, and an aborted signal never fires again. + else if (options.signal?.aborted) stop('cancelled'); + let spawnError: Error | undefined; + const closed = new Promise(resolveClosed => child.once('close', () => resolveClosed())); + // Settle from the leader's exit, not from 'close': 'close' also waits for every holder of the pipes, which can + // include a process that left the group and so survives the group kill. + let finished = false; + const finish = (code: number | null, signal: NodeJS.Signals | null) => { + if (finished) return; + finished = true; clearTimeout(deadline); options.signal?.removeEventListener('abort', onAbort); - void drainGroup(pgid).then(drained => { + void drainGroup(pgid).then(async drained => { + await Promise.race([closed, pause(STDIO_CLOSE_MS)]); + child.stdout.destroy(); + child.stderr.destroy(); clearTimeout(graceTimer); const stdout = Buffer.concat(out).toString('utf8'), stderr = Buffer.concat(err).toString('utf8'); if (!drained) { @@ -134,6 +147,9 @@ export function runInProcessGroup(file: string, args: readonly string[], } resolve({ status: code, stdout, stderr }); }); - }); + }; + child.once('exit', finish); + // A spawn that fails after the child had a PID emits 'error' and may never emit 'exit'. + child.once('error', error => { spawnError = error; finish(null, null); }); }); } diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts index e4381a93..2583da50 100644 --- a/test/agent-process-group.test.ts +++ b/test/agent-process-group.test.ts @@ -1,3 +1,6 @@ +import { mkdtempSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { describe, expect, it } from 'vitest'; import { runInProcessGroup, type ProcessGroup } from '../agents/process-group.ts'; import { NOT_STARTED } from '../agents/docker.ts'; @@ -65,6 +68,34 @@ describe('runInProcessGroup', () => { expect(groupAlive(group!.pgid)).toBe(false); }); + it('stops at once when the signal is aborted during onProcessGroup', async () => { + const controller = new AbortController(); + let group: ProcessGroup | undefined; + const began = performance.now(); + const outcome = await runInProcessGroup('sleep', ['60'], { env, timeoutMs: 30_000, signal: controller.signal, + onProcessGroup: reported => { group = reported; controller.abort(); } }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ABORT_ERR'); + expect(performance.now() - began).toBeLessThan(10_000); + expect(groupAlive(group!.pgid)).toBe(false); + }); + + it('settles when a process that left the group still holds its output pipe', async () => { + const dir = mkdtempSync(join(tmpdir(), 'escaped-')), pidFile = join(dir, 'pid'); + // The leader starts `sleep` in a new session (outside the group) that inherits stdout, then exits. + const escape = `const c = require('node:child_process').spawn('sleep', ['60'], { detached: true, ` + + `stdio: ['ignore', 'inherit', 'inherit'] }); require('node:fs').writeFileSync(${JSON.stringify(pidFile)}, ` + + `String(c.pid)); c.unref(); console.log('leader done');`; + try { + const began = performance.now(); + const outcome = await runInProcessGroup(process.execPath, ['-e', escape], { env, timeoutMs: 30_000 }); + expect(outcome).toMatchObject({ status: 0, stdout: 'leader done\n' }); + expect(performance.now() - began).toBeLessThan(10_000); + } finally { + try { process.kill(Number(readFileSync(pidFile, 'utf8')), 'SIGKILL'); } catch { /* already gone */ } + rmSync(dir, { recursive: true, force: true }); + } + }, 30_000); + it('stops a group whose output exceeds the limit', async () => { const outcome = await runInProcessGroup('sh', ['-c', 'while :; do echo xxxxxxxxxxxxxxxx; done'], { env, timeoutMs: 30_000, maxBuffer: 1024 }); From 4728f266a25698d5ab59b1dc263e25feb8dcb715 Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 13:32:58 -0700 Subject: [PATCH 04/12] Clear the SIGKILL timer once the group is empty; release the pipe wait; document settlement bounds The grace-period SIGKILL timer was cleared only after the up-to-1-second pipe wait, so a sub-second grace could fire it at an ID the group no longer held. It is now cleared as soon as the drain ends. The pipe-wait timer is cleared when the pipes close, so a finished call never keeps the process alive. The lifecycle doc now states the EGROUPALIVE and 1-second pipe bounds. Co-Authored-By: Claude Opus 5.5 --- agents/process-group.ts | 8 ++++++-- docs/implementation/runner-lifecycle.md | 2 +- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/agents/process-group.ts b/agents/process-group.ts index bf4fbd8e..3c8d4ea2 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -113,10 +113,14 @@ export function runInProcessGroup(file: string, args: readonly string[], clearTimeout(deadline); options.signal?.removeEventListener('abort', onAbort); void drainGroup(pgid).then(async drained => { - await Promise.race([closed, pause(STDIO_CLOSE_MS)]); + // The group is empty (or given up on): a later SIGKILL could reach a new group that reuses the ID. + clearTimeout(graceTimer); + let stdioTimer: ReturnType | undefined; + await Promise.race([closed, new Promise(done => { stdioTimer = setTimeout(done, STDIO_CLOSE_MS); })]); + // Cleared so a finished call never keeps the process alive. + clearTimeout(stdioTimer); child.stdout.destroy(); child.stderr.destroy(); - clearTimeout(graceTimer); const stdout = Buffer.concat(out).toString('utf8'), stderr = Buffer.concat(err).toString('utf8'); if (!drained) { resolve({ status: null, stdout, stderr, error: Object.assign( diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index ae2fbde7..40da7fe6 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -260,7 +260,7 @@ This belongs to D (D5 or a D follow-up), not F. It changes `agents/contract.ts`, **Also before F1 merges: asynchronous launch.** Done in #51 item 2 (see "Launch setup does not block the server"). -**Third prerequisite before F1 merges: abortable preparation helpers.** Done in #51 item 5: `createTaskCloneAsync` and `prepareTaskFilesystemsAsync` take `signal` and `onProcessGroup`. Each subprocess runs as the leader of a new process group (`agents/process-group.ts`); `onProcessGroup` receives `{ pgid, startedAt }` synchronously after the spawn, so F can save it as `preparation_pgid` and `preparation_started_at` before anything else runs. Every Git or Docker call is a separate group, so F updates the saved group on each call. On abort or at the deadline the group gets `SIGTERM`, then `SIGKILL` after 5 seconds; the call settles only after the leader and every other member of its group have exited. An aborted clone removes its partial directory; an aborted storage allocation removes what it created, with cleanup that is not cancelled. The synchronous helpers remain for existing callers and share the same steps. What follows is the original requirement. Both preparation helpers were synchronous. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. +**Third prerequisite before F1 merges: abortable preparation helpers.** Done in #51 item 5: `createTaskCloneAsync` and `prepareTaskFilesystemsAsync` take `signal` and `onProcessGroup`. Each subprocess runs as the leader of a new process group (`agents/process-group.ts`); `onProcessGroup` receives `{ pgid, startedAt }` synchronously after the spawn, so F can save it as `preparation_pgid` and `preparation_started_at` before anything else runs. Every Git or Docker call is a separate group, so F updates the saved group on each call. On abort or at the deadline the group gets `SIGTERM`, then `SIGKILL` after 5 seconds; the call settles only after the leader and every other member of its group have exited. Two bounds apply: a member still alive 10 seconds after the leader exits (repeated `SIGKILL` included) settles the call with `EGROUPALIVE`, which means the group may still be running and F must treat the outcome as unknown; and an output pipe still open 1 second after the group is gone (held by a process that left the group) is closed, so settlement never waits on it. An aborted clone removes its partial directory; an aborted storage allocation removes what it created, with cleanup that is not cancelled. The synchronous helpers remain for existing callers and share the same steps. What follows is the original requirement. Both preparation helpers were synchronous. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. **Prerequisite before F2's writable attempts: a bounded task-volume export.** D must add a bounded diff export over the task volumes (see "Partial output and interrupted rebases"). F1 does not need it: F1 runs only read-only attempts, and their task storage is removed without an export. From 5b1d1cc0bcd0dfa13a9e197e59de92619da77c33 Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 15:48:56 -0700 Subject: [PATCH 05/12] Pause while reading a large directory; start the deadline before the group is recorded The object audit and link check counted only paths taken off the queue, so reading one huge directory held the event loop (and an abort) for all of it. They now count every entry read too, and pause every 1,000 either way. The process deadline was armed only after onProcessGroup returned, so a slow durable write let the child run past timeoutMs; it is now armed at the spawn. Co-Authored-By: Claude Opus 5.5 --- agents/container/storage.ts | 6 ++- agents/process-group.ts | 13 ++--- git/clone.ts | 6 ++- test/agent-process-group.test.ts | 9 ++++ test/agent-walk-pauses.test.ts | 81 ++++++++++++++++++++++++++++++++ 5 files changed, 105 insertions(+), 10 deletions(-) create mode 100644 test/agent-walk-pauses.test.ts diff --git a/agents/container/storage.ts b/agents/container/storage.ts index 7db1682c..d8613319 100644 --- a/agents/container/storage.ts +++ b/agents/container/storage.ts @@ -236,11 +236,12 @@ const linkStaysInside = (staging: string, link: string) => { */ function* containedLinks(staging: string, remaining: () => number): Steps { const metadata = join(staging, '.git'), pending = [staging]; - let count = 0; + // `walked` counts every entry touched, including each one read from a directory, so one huge directory pauses too. + let count = 0, walked = 0; while (pending.length) { remaining(); count++; - if (count % PAUSE_EVERY === 0) yield PAUSE; + if (++walked % PAUSE_EVERY === 0) yield PAUSE; const path = pending.pop()!, stat = lstatSync(path); if (stat.isSymbolicLink()) { const name = JSON.stringify(relative(staging, path)); @@ -258,6 +259,7 @@ function* containedLinks(staging: string, remaining: () => number): Steps if (count + pending.length >= LINK_INSPECTION_LIMIT) throw new Error('Repository checkout exceeds the link inspection limit.'); pending.push(join(path, entry.name)); + if (++walked % PAUSE_EVERY === 0) yield PAUSE; } } finally { directory.closeSync(); } } diff --git a/agents/process-group.ts b/agents/process-group.ts index 3c8d4ea2..f4019556 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -76,11 +76,6 @@ export function runInProcessGroup(file: string, args: readonly string[], child.once('error', error => resolve({ status: null, stdout: '', stderr: '', error })); return; } - // If the caller cannot record the group, the child must not outlive this call: it is killed at once below, and - // the call still settles only after the group has exited, with the caller's error. - let unrecorded: unknown; - try { options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); } - catch (error) { unrecorded = error; } const out: Buffer[] = [], err: Buffer[] = []; let outBytes = 0, errBytes = 0, stopped: 'cancelled' | 'timeout' | 'output-limit' | 'unrecorded' | undefined; let graceTimer: ReturnType | undefined; @@ -90,13 +85,19 @@ export function runInProcessGroup(file: string, args: readonly string[], signalGroup(pgid, 'SIGTERM'); graceTimer = setTimeout(() => signalGroup(pgid, 'SIGKILL'), graceMs); }; + // Armed at the spawn, before the caller records the group: time spent recording counts against the deadline. + const deadline = setTimeout(() => stop('timeout'), options.timeoutMs); + // If the caller cannot record the group, the child must not outlive this call: it is killed at once below, and + // the call still settles only after the group has exited, with the caller's error. + let unrecorded: unknown; + try { options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); } + catch (error) { unrecorded = error; } const collect = (chunks: Buffer[], add: (bytes: number) => number) => (chunk: Buffer) => { if (add(chunk.length) > maxBuffer) stop('output-limit'); else chunks.push(chunk); }; child.stdout.on('data', collect(out, bytes => (outBytes += bytes))); child.stderr.on('data', collect(err, bytes => (errBytes += bytes))); - const deadline = setTimeout(() => stop('timeout'), options.timeoutMs); const onAbort = () => stop('cancelled'); options.signal?.addEventListener('abort', onAbort, { once: true }); if (unrecorded !== undefined) { stopped = 'unrecorded'; signalGroup(pgid, 'SIGKILL'); } diff --git a/git/clone.ts b/git/clone.ts index 2b5a8bc5..4b7c2a0d 100644 --- a/git/clone.ts +++ b/git/clone.ts @@ -78,11 +78,12 @@ function* cloneSteps(options: CloneOptions): Generator 100_000) throw new Error('Object storage exceeds inspection limit.'); - if (count % PAUSE_EVERY === 0) yield PAUSE; + if (++walked % PAUSE_EVERY === 0) yield PAUSE; const path = pending.pop()!, stat = lstatSync(path); if (stat.isSymbolicLink() || (!stat.isDirectory() && !stat.isFile())) throw new Error('Unsupported object entry.'); if (independent && stat.isFile() && stat.nlink !== 1) throw new Error('Task objects must not be hard-linked.'); @@ -93,6 +94,7 @@ function* cloneSteps(options: CloneOptions): Generator= 100_000) throw new Error('Object storage exceeds inspection limit.'); pending.push(join(path, entry.name)); + if (++walked % PAUSE_EVERY === 0) yield PAUSE; } } finally { directory.closeSync(); } } diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts index 2583da50..611c5144 100644 --- a/test/agent-process-group.test.ts +++ b/test/agent-process-group.test.ts @@ -96,6 +96,15 @@ describe('runInProcessGroup', () => { } }, 30_000); + it('counts time spent recording the group against the deadline', async () => { + const began = performance.now(); + const outcome = await runInProcessGroup('sleep', ['60'], { env, timeoutMs: 1_000, graceMs: 100, + // A slow durable write: the deadline must already be running while it blocks. + onProcessGroup: () => { const until = performance.now() + 1_000; while (performance.now() < until); } }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ETIMEDOUT'); + expect(performance.now() - began).toBeLessThan(1_700); + }); + it('stops a group whose output exceeds the limit', async () => { const outcome = await runInProcessGroup('sh', ['-c', 'while :; do echo xxxxxxxxxxxxxxxx; done'], { env, timeoutMs: 30_000, maxBuffer: 1024 }); diff --git a/test/agent-walk-pauses.test.ts b/test/agent-walk-pauses.test.ts new file mode 100644 index 00000000..8834be02 --- /dev/null +++ b/test/agent-walk-pauses.test.ts @@ -0,0 +1,81 @@ +import { execFileSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +// The asynchronous clone and storage allocation pause every 1,000 entries they walk, including entries read from one +// huge directory, so the event loop (and an abort) is never held for a whole directory (#51 item 5). +const trace = vi.hoisted(() => ({ run: 0, longest: 0, total: 0 })); +vi.mock('node:fs', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, opendirSync: ((...args: Parameters) => { + const directory = actual.opendirSync(...args), read = directory.readSync.bind(directory); + directory.readSync = () => { + const entry = read(); + if (entry) { trace.run += 1; trace.total += 1; trace.longest = Math.max(trace.longest, trace.run); } + return entry; + }; + return directory; + }) as typeof actual.opendirSync }; +}); +vi.mock('../agents/container/image.ts', async importOriginal => ({ + ...await importOriginal(), assertBuiltAgentImage: () => {}, +})); +const { createTaskClone, createTaskCloneAsync } = await import('../git/clone.ts'); +const { prepareTaskFilesystemsAsync } = await import('../agents/container/storage.ts'); + +const ENTRIES = 2_500; +let root = ''; +const realSetImmediate = globalThis.setImmediate; +const git = (cwd: string, ...args: string[]) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', ...args], + { cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }).trim(); +const repository = () => { + const source = join(root, 'source'), parent = join(root, 'staging'); + mkdirSync(source); mkdirSync(parent); + git(source, 'init'); git(source, 'config', 'user.name', 'Test'); git(source, 'config', 'user.email', 'test@example.com'); + writeFileSync(join(source, 'file.txt'), 'trusted\n'); git(source, 'add', '.'); git(source, 'commit', '-m', 'baseline'); + return { source, parent, head: git(source, 'rev-parse', 'HEAD'), taskId: 'task-1' }; +}; +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'walk-pauses-')); + trace.run = 0; trace.longest = 0; trace.total = 0; + // Every pause of the asynchronous drivers goes through setImmediate: the run of entries read restarts there. + vi.spyOn(globalThis, 'setImmediate').mockImplementation(((callback: (...args: unknown[]) => void, ...args: unknown[]) => { + trace.run = 0; + return realSetImmediate(callback, ...args); + }) as typeof setImmediate); +}); +afterEach(() => { vi.restoreAllMocks(); rmSync(root, { recursive: true, force: true }); }); + +describe('pauses while walking one large directory', () => { + it('in the asynchronous clone, reading object storage', async () => { + const input = repository(); + // One object-storage directory with far more entries than one pause interval. + const info = join(input.source, '.git', 'objects', 'info'); + for (let index = 0; index < ENTRIES; index++) writeFileSync(join(info, `entry-${index}`), ''); + await createTaskCloneAsync(input); + expect(trace.total).toBeGreaterThanOrEqual(ENTRIES); + expect(trace.longest).toBeLessThanOrEqual(1_000); + }, 60_000); + + it('in the asynchronous storage allocation, checking the checkout for links', async () => { + const input = repository(); + const clone = createTaskClone(input); + const wide = join(clone.directory, 'wide'); + mkdirSync(wide); + for (let index = 0; index < ENTRIES; index++) writeFileSync(join(wide, `entry-${index}`), ''); + trace.run = 0; trace.longest = 0; trace.total = 0; + // No Docker on PATH: the allocation fails at its first Docker call, after the link walk has finished. + const path = process.env.PATH; + process.env.PATH = '/nonexistent'; + try { + await expect(prepareTaskFilesystemsAsync(clone, { workBytes: 1 << 24, workInodes: 512, metadataBytes: 1 << 24, + metadataInodes: 512 }, `sha256:${'a'.repeat(64)}`, { runnerOwner: '0'.repeat(32), attemptId: 'walk', + allocationId: randomUUID() })).rejects.toThrow(); + } finally { process.env.PATH = path; } + expect(trace.total).toBeGreaterThanOrEqual(ENTRIES); + expect(trace.longest).toBeLessThanOrEqual(1_000); + }, 60_000); +}); From 4fb2c41773013cb5db8655413eb34811d267213d Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 15:57:23 -0700 Subject: [PATCH 06/12] Add review rules from #73: reread whole functions, verify stated claims, independent review Copilot found three issues on #73 that several author self-reviews missed: a pause cadence claimed in comments but broken by an inner directory loop, and a deadline armed only after a synchronous callback. New review-readiness rules require rereading every changed function in each pass, verifying every behavioural claim with a failing-if-false test, and an independent review for concurrency, process and cleanup code. Two new required race cases cover synchronous lifecycle hooks and descendants that hold output pipes. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 587d940c..f077e2dc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,12 +39,17 @@ Before opening or updating a PR for asynchronous behavior, test every applicable - An abort error fires, then subprocess close arrives later. - Submit starts, then the user edits the composer or switches items, then the response returns. - Referenced code is reassigned or the snapshot changes, then retry or rendering occurs. +- A synchronous caller hook inside a lifecycle (such as a callback that records a spawned process group) throws, blocks past the deadline, or aborts the signal, then the lifecycle continues. Timers and listeners that bound the lifecycle must already be armed when the hook runs, and a signal already aborted when its listener is attached must still take effect. +- A child process exits, but a descendant outside its process group still holds its output pipes. Every reproduced race requires a failing-before and passing-after regression. Assert both the visible result and the durable state when they can diverge. ## Review readiness - Before requesting or re-requesting an automated Copilot review, self-review the full current diff, fix every issue found, and repeat the self-review and fix cycle until a complete pass finds no new issues. Re-run the relevant validation after fixes; only then request Copilot review. +- Each self-review pass rereads every changed function in full against the base, not only the lines changed since the previous round. Code unchanged since the first commit of the PR still gets reviewed in every pass. +- Treat every behavioural claim the change makes, in code comments, the PR body or docs (for example "pauses every 1,000 entries", "settles only after exit", "never throws", "bounded by N seconds"), as something to verify. Trace each claim through every path that can break it, including nested loops, callbacks, error paths and early returns, and give it a test that fails if the claim is false. +- The author's self-review is not enough for concurrency, process, subprocess, timer or resource-cleanup code. Before requesting Copilot review, also run an independent review that does not share the author's context: `/codex review`, a separate review agent, or `/code-review` at `high` effort or above. Fix its findings like any other. - Run final validation against the exact pushed head after the last change. - Report current test counts separately from historical milestone counts. - Before requesting automated review, report the current head, CI state, mergeability, unresolved threads, and deferred follow-up issues. From d0cff551686a2bb892fe198558e06226f10aa44b Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 16:01:36 -0700 Subject: [PATCH 07/12] Count a Git call as a yield in the clone walk-pause test The two object checks run before and after the clone call, which is awaited and so yields the event loop. The test counted reads across that boundary as one run and failed at 1,001 on Linux's object layout. It now resets the run at each Git spawn; it still fails at 2,506 on the pre-fix clone code. Co-Authored-By: Claude Opus 5.5 --- test/agent-walk-pauses.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/agent-walk-pauses.test.ts b/test/agent-walk-pauses.test.ts index 8834be02..52a802bd 100644 --- a/test/agent-walk-pauses.test.ts +++ b/test/agent-walk-pauses.test.ts @@ -55,7 +55,9 @@ describe('pauses while walking one large directory', () => { // One object-storage directory with far more entries than one pause interval. const info = join(input.source, '.git', 'objects', 'info'); for (let index = 0; index < ENTRIES; index++) writeFileSync(join(info, `entry-${index}`), ''); - await createTaskCloneAsync(input); + // Each Git call is awaited, which yields the event loop too; the object checks before and after the clone call are + // separate walks, so a run only counts entries read without any yield in between. + await createTaskCloneAsync({ ...input, onProcessGroup: () => { trace.run = 0; } }); expect(trace.total).toBeGreaterThanOrEqual(ENTRIES); expect(trace.longest).toBeLessThanOrEqual(1_000); }, 60_000); From 523992a75b6657f033014d1966dd1764f03a61df Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 16:19:23 -0700 Subject: [PATCH 08/12] Report synchronous spawn failures as outcomes; stop nothing once the leader has exited spawn throws, rather than emitting 'error', for spawn failures such as ENOMEM or E2BIG. The throw rejected runInProcessGroup, so the step drivers skipped the generators' cleanup: an allocation left its volumes, stayed live and kept its claim. It is now reported as an outcome. stop() no longer runs after the leader exits, so output read during the pipe wait cannot signal a drained, possibly reused group ID or re-arm the SIGKILL timer. Co-Authored-By: Claude Opus 5.5 --- agents/process-group.ts | 26 +++++++++++++++++++------- test/agent-process-group.test.ts | 10 ++++++++++ test/agent-storage-async.test.ts | 21 ++++++++++++++++++++- 3 files changed, 49 insertions(+), 8 deletions(-) diff --git a/agents/process-group.ts b/agents/process-group.ts index f4019556..7beaf5e6 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -68,8 +68,16 @@ export function runInProcessGroup(file: string, args: readonly string[], const graceMs = options.graceMs ?? DEFAULT_GRACE_MS, maxBuffer = options.maxBuffer ?? DEFAULT_MAX_BUFFER; return new Promise(resolve => { // `detached` makes the child the leader of a new process group (setsid), so signals to -pgid reach all it starts. - const child = spawn(file, [...args], { cwd: options.cwd, env: options.env, detached: true, - stdio: ['ignore', 'pipe', 'pipe'] }); + // spawn throws, rather than emitting 'error', for invalid arguments and for spawn failures other than ENOENT, + // EACCES, EAGAIN, EMFILE and ENFILE (for example ENOMEM or E2BIG). No child exists then; report it, never reject. + let child: ReturnType; + try { + child = spawn(file, [...args], { cwd: options.cwd, env: options.env, detached: true, + stdio: ['ignore', 'pipe', 'pipe'] }); + } catch (error) { + resolve({ status: null, stdout: '', stderr: '', error: error as Error }); + return; + } const pgid = child.pid; if (pgid === undefined) { // The spawn itself failed (for example ENOENT); nothing started, and 'error' carries the cause. @@ -79,8 +87,10 @@ export function runInProcessGroup(file: string, args: readonly string[], const out: Buffer[] = [], err: Buffer[] = []; let outBytes = 0, errBytes = 0, stopped: 'cancelled' | 'timeout' | 'output-limit' | 'unrecorded' | undefined; let graceTimer: ReturnType | undefined; + // Set once the leader has exited: from then on the group is drained, not stopped, and its ID may be reused. + let exited = false; const stop = (reason: NonNullable) => { - if (stopped) return; + if (stopped || exited) return; stopped = reason; signalGroup(pgid, 'SIGTERM'); graceTimer = setTimeout(() => signalGroup(pgid, 'SIGKILL'), graceMs); @@ -92,12 +102,13 @@ export function runInProcessGroup(file: string, args: readonly string[], let unrecorded: unknown; try { options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); } catch (error) { unrecorded = error; } + // Output past the limit stops a running group; once the leader has exited, the excess is only dropped. const collect = (chunks: Buffer[], add: (bytes: number) => number) => (chunk: Buffer) => { if (add(chunk.length) > maxBuffer) stop('output-limit'); else chunks.push(chunk); }; - child.stdout.on('data', collect(out, bytes => (outBytes += bytes))); - child.stderr.on('data', collect(err, bytes => (errBytes += bytes))); + child.stdout!.on('data', collect(out, bytes => (outBytes += bytes))); + child.stderr!.on('data', collect(err, bytes => (errBytes += bytes))); const onAbort = () => stop('cancelled'); options.signal?.addEventListener('abort', onAbort, { once: true }); if (unrecorded !== undefined) { stopped = 'unrecorded'; signalGroup(pgid, 'SIGKILL'); } @@ -111,6 +122,7 @@ export function runInProcessGroup(file: string, args: readonly string[], const finish = (code: number | null, signal: NodeJS.Signals | null) => { if (finished) return; finished = true; + exited = true; clearTimeout(deadline); options.signal?.removeEventListener('abort', onAbort); void drainGroup(pgid).then(async drained => { @@ -120,8 +132,8 @@ export function runInProcessGroup(file: string, args: readonly string[], await Promise.race([closed, new Promise(done => { stdioTimer = setTimeout(done, STDIO_CLOSE_MS); })]); // Cleared so a finished call never keeps the process alive. clearTimeout(stdioTimer); - child.stdout.destroy(); - child.stderr.destroy(); + child.stdout!.destroy(); + child.stderr!.destroy(); const stdout = Buffer.concat(out).toString('utf8'), stderr = Buffer.concat(err).toString('utf8'); if (!drained) { resolve({ status: null, stdout, stderr, error: Object.assign( diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts index 611c5144..36fa768b 100644 --- a/test/agent-process-group.test.ts +++ b/test/agent-process-group.test.ts @@ -112,6 +112,16 @@ describe('runInProcessGroup', () => { expect(outcome.stdout.length).toBeLessThanOrEqual(1024); }); + it('resolves, never rejects, when spawn itself throws', async () => { + let reported = false; + // A NUL byte in an argument makes spawn throw synchronously, as ENOMEM or E2BIG from the spawn itself would. + const outcome = await runInProcessGroup('sh', ['-c', 'echo a\0b'], { env, timeoutMs: 5_000, + onProcessGroup: () => { reported = true; } }); + expect(outcome.status).toBeNull(); + expect(outcome.error).toBeInstanceOf(Error); + expect(reported).toBe(false); + }); + it('starts nothing for an aborted signal, an invalid deadline, or a missing program', async () => { let reported = false; const onProcessGroup = () => { reported = true; }; diff --git a/test/agent-storage-async.test.ts b/test/agent-storage-async.test.ts index 9caf39bf..f7aecf6e 100644 --- a/test/agent-storage-async.test.ts +++ b/test/agent-storage-async.test.ts @@ -8,10 +8,20 @@ import type { ProcessGroup } from '../agents/process-group.ts'; // The asynchronous storage allocation (#51 item 5), against a fake Docker CLI that keeps its objects as files, so the // abort path runs without a daemon. The real-Docker suite covers the same path against Docker itself. +// spawn passes through unless a test makes it throw, as it does synchronously for errors such as ENOMEM. +const spawnFault = vi.hoisted(() => ({ on: undefined as undefined | ((args: readonly string[]) => boolean) })); +vi.mock('node:child_process', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, spawn: ((file: string, args: readonly string[], options: object) => { + if (spawnFault.on?.(args)) throw Object.assign(new Error('spawn ENOMEM'), { code: 'ENOMEM', syscall: 'spawn' }); + return actual.spawn(file, args, options); + }) as typeof actual.spawn }; +}); vi.mock('../agents/container/image.ts', async importOriginal => ({ ...await importOriginal(), assertBuiltAgentImage: () => {}, })); -const { prepareTaskFilesystemsAsync, removeTaskFilesystems } = await import('../agents/container/storage.ts'); +const { hasLiveTaskStorage, prepareTaskFilesystemsAsync, removeTaskFilesystems } = + await import('../agents/container/storage.ts'); const { createTaskClone } = await import('../git/clone.ts'); const RUNNER = '0123456789abcdef0123456789abcdef'; @@ -85,6 +95,7 @@ beforeEach(() => { process.env.PATH = `${bin}:${path}`; }); afterEach(() => { + spawnFault.on = undefined; process.env.PATH = path; rmSync(root, { recursive: true, force: true }); }); @@ -133,6 +144,14 @@ describe('asynchronous task storage allocation', () => { expect(stored()).toEqual([]); }, 30_000); + it('removes what it created and releases the allocation when a later Docker call cannot be spawned', async () => { + // Both volumes are created; spawning the keeper's `docker create` then fails synchronously. + spawnFault.on = args => args[0] === 'create'; + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner())).rejects.toThrow('ENOMEM'); + expect(stored()).toEqual([]); + expect(hasLiveTaskStorage(RUNNER)).toBe(false); + }); + it('creates nothing when the signal is already aborted', async () => { const groups: ProcessGroup[] = []; await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), From ecb2fb37aea49b6d03096dcfcdb0a05266f9537e Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 16:21:51 -0700 Subject: [PATCH 09/12] Report output held open past the group as an error, not a success When a process outside the group held the pipes past the 1 s wait, the call returned the leader's exit status with output that may be incomplete. It now settles with ESTDIOHELD and keeps what was read. Before deciding, it gives I/O one more event-loop turn, since after a long block the timer can fire before output that is already waiting is read. Co-Authored-By: Claude Opus 5.5 --- agents/process-group.ts | 14 +++++++++++++- test/agent-process-group.test.ts | 4 +++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/agents/process-group.ts b/agents/process-group.ts index 7beaf5e6..3f82efa0 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -115,7 +115,8 @@ export function runInProcessGroup(file: string, args: readonly string[], // An abort raised during onProcessGroup came before the listener existed, and an aborted signal never fires again. else if (options.signal?.aborted) stop('cancelled'); let spawnError: Error | undefined; - const closed = new Promise(resolveClosed => child.once('close', () => resolveClosed())); + let pipesClosed = false; + const closed = new Promise(resolveClosed => child.once('close', () => { pipesClosed = true; resolveClosed(); })); // Settle from the leader's exit, not from 'close': 'close' also waits for every holder of the pipes, which can // include a process that left the group and so survives the group kill. let finished = false; @@ -132,6 +133,9 @@ export function runInProcessGroup(file: string, args: readonly string[], await Promise.race([closed, new Promise(done => { stdioTimer = setTimeout(done, STDIO_CLOSE_MS); })]); // Cleared so a finished call never keeps the process alive. clearTimeout(stdioTimer); + // Timers run before I/O in each event-loop turn: after a long block the timer can win while the rest of the + // output is already waiting. One more turn lets that I/O (and the end of the pipes) be read first. + if (!pipesClosed) await new Promise(resolveTurn => setImmediate(resolveTurn)); child.stdout!.destroy(); child.stderr!.destroy(); const stdout = Buffer.concat(out).toString('utf8'), stderr = Buffer.concat(err).toString('utf8'); @@ -140,6 +144,14 @@ export function runInProcessGroup(file: string, args: readonly string[], new Error(`${file} left processes in its group that did not exit after SIGKILL.`), { code: 'EGROUPALIVE' }) }); return; } + if (!pipesClosed) { + // Only a process outside the group can still hold the pipes; the output may be incomplete, so this is not a + // successful result even if the leader exited 0. + resolve({ status: null, stdout, stderr, error: Object.assign(new Error( + `${file} exited, but a process outside its group still holds its output; the output may be incomplete.`), + { code: 'ESTDIOHELD' }) }); + return; + } if (stopped === 'unrecorded') { resolve({ status: null, stdout, stderr, error: Object.assign(new Error( `${file} was stopped because its process group could not be recorded.`, { cause: unrecorded }), diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts index 36fa768b..f7bb2ea3 100644 --- a/test/agent-process-group.test.ts +++ b/test/agent-process-group.test.ts @@ -88,7 +88,9 @@ describe('runInProcessGroup', () => { try { const began = performance.now(); const outcome = await runInProcessGroup(process.execPath, ['-e', escape], { env, timeoutMs: 30_000 }); - expect(outcome).toMatchObject({ status: 0, stdout: 'leader done\n' }); + // The output may be incomplete, so it is not reported as a success; what was read is kept. + expect(outcome).toMatchObject({ status: null, stdout: 'leader done\n' }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ESTDIOHELD'); expect(performance.now() - began).toBeLessThan(10_000); } finally { try { process.kill(Number(readFileSync(pidFile, 'utf8')), 'SIGKILL'); } catch { /* already gone */ } From 5fa6c1880e95d81bb282cb9f1cb08d5645257d29 Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 16:22:28 -0700 Subject: [PATCH 10/12] Document ESTDIOHELD and synchronous spawn failures in the lifecycle bounds Co-Authored-By: Claude Opus 5.5 --- docs/implementation/runner-lifecycle.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index 40da7fe6..8dbcf4bb 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -260,7 +260,7 @@ This belongs to D (D5 or a D follow-up), not F. It changes `agents/contract.ts`, **Also before F1 merges: asynchronous launch.** Done in #51 item 2 (see "Launch setup does not block the server"). -**Third prerequisite before F1 merges: abortable preparation helpers.** Done in #51 item 5: `createTaskCloneAsync` and `prepareTaskFilesystemsAsync` take `signal` and `onProcessGroup`. Each subprocess runs as the leader of a new process group (`agents/process-group.ts`); `onProcessGroup` receives `{ pgid, startedAt }` synchronously after the spawn, so F can save it as `preparation_pgid` and `preparation_started_at` before anything else runs. Every Git or Docker call is a separate group, so F updates the saved group on each call. On abort or at the deadline the group gets `SIGTERM`, then `SIGKILL` after 5 seconds; the call settles only after the leader and every other member of its group have exited. Two bounds apply: a member still alive 10 seconds after the leader exits (repeated `SIGKILL` included) settles the call with `EGROUPALIVE`, which means the group may still be running and F must treat the outcome as unknown; and an output pipe still open 1 second after the group is gone (held by a process that left the group) is closed, so settlement never waits on it. An aborted clone removes its partial directory; an aborted storage allocation removes what it created, with cleanup that is not cancelled. The synchronous helpers remain for existing callers and share the same steps. What follows is the original requirement. Both preparation helpers were synchronous. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. +**Third prerequisite before F1 merges: abortable preparation helpers.** Done in #51 item 5: `createTaskCloneAsync` and `prepareTaskFilesystemsAsync` take `signal` and `onProcessGroup`. Each subprocess runs as the leader of a new process group (`agents/process-group.ts`); `onProcessGroup` receives `{ pgid, startedAt }` synchronously after the spawn, so F can save it as `preparation_pgid` and `preparation_started_at` before anything else runs. Every Git or Docker call is a separate group, so F updates the saved group on each call. On abort or at the deadline the group gets `SIGTERM`, then `SIGKILL` after 5 seconds; the call settles only after the leader and every other member of its group have exited. Two bounds apply: a member still alive 10 seconds after the leader exits (repeated `SIGKILL` included) settles the call with `EGROUPALIVE`, which means the group may still be running and F must treat the outcome as unknown; and an output pipe still open 1 second after the group is gone (held by a process that left the group) is closed, so settlement never waits on it, and the call settles with `ESTDIOHELD`: its output may be incomplete, so it is never reported as a success. A spawn that fails outright (for example `ENOMEM`) settles as not started, and the helper never rejects. An aborted clone removes its partial directory; an aborted storage allocation removes what it created, with cleanup that is not cancelled. The synchronous helpers remain for existing callers and share the same steps. What follows is the original requirement. Both preparation helpers were synchronous. `createTaskClone` runs git with `execFileSync`, and `prepareTaskFilesystems` runs `docker` with `execFileSync` and `spawnSync` (`agents/container/storage.ts`). Either would block the event loop, and cancel or shutdown could not abort it. D must add asynchronous variants of both that take an `AbortSignal`, spawn their subprocesses in their own process group, expose the group ID, and settle only after those processes have exited. Storage allocation must also label its volumes with the runner owner and attempt ID before it returns (see "Recovered storage handles"), so that storage allocated just before a crash is found by recovery. **Prerequisite before F2's writable attempts: a bounded task-volume export.** D must add a bounded diff export over the task volumes (see "Partial output and interrupted rebases"). F1 does not need it: F1 runs only read-only attempts, and their task storage is removed without an export. From 2f45273375b60b4f27986495120c308817e78713 Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 16:26:40 -0700 Subject: [PATCH 11/12] Settle truncated output as ENOBUFS; report every post-abort failure as a cancel Output crossing the limit after the leader exited was dropped while the call still reported success; any crossing now settles as ENOBUFS. An abort landing between two Git or Docker calls made the next call refuse to start, so the clone or allocation rejected with a plain error instead of an AbortError; both now rethrow any failure after the signal aborted as an AbortError with the original as its cause (an unsettled cleanup stays an AggregateError). Found by an independent review agent after the author's passes. Co-Authored-By: Claude Opus 5.5 --- agents/container/storage.ts | 10 +++++++++- agents/process-group.ts | 11 +++++++++-- git/clone.ts | 10 ++++++++++ test/agent-clone-async.test.ts | 29 ++++++++++++++++++++++++++--- test/agent-storage-async.test.ts | 22 +++++++++++++++++++--- 5 files changed, 73 insertions(+), 9 deletions(-) diff --git a/agents/container/storage.ts b/agents/container/storage.ts index d8613319..acdd9b2f 100644 --- a/agents/container/storage.ts +++ b/agents/container/storage.ts @@ -386,7 +386,15 @@ export async function prepareTaskFilesystemsAsync(clone: TaskClone, limits: Task owner: ResourceOwner, options: PreparationOptions & { readonly timeoutMs?: number } = {}): Promise { if (options.signal?.aborted) throw Object.assign(new Error('Task storage allocation was cancelled.'), { name: 'AbortError', code: 'ABORT_ERR' }); - return runStepsAsync(allocation(clone, limits, imageId, owner, options.timeoutMs ?? 60_000), options); + try { return await runStepsAsync(allocation(clone, limits, imageId, owner, options.timeoutMs ?? 60_000), options); } + catch (error) { + // However the abort surfaced, it is a cancel. A cleanup that did not settle stays an AggregateError: the caller + // must still learn what may be left. + if (options.signal?.aborted && !(error instanceof AggregateError) && (error as Error | undefined)?.name !== 'AbortError') + throw Object.assign(new Error('Task storage allocation was cancelled.', { cause: error }), + { name: 'AbortError', code: 'ABORT_ERR' }); + throw error; + } } /** diff --git a/agents/process-group.ts b/agents/process-group.ts index 3f82efa0..836677a5 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -102,9 +102,11 @@ export function runInProcessGroup(file: string, args: readonly string[], let unrecorded: unknown; try { options.onProcessGroup?.(Object.freeze({ pgid, startedAt: Date.now() })); } catch (error) { unrecorded = error; } - // Output past the limit stops a running group; once the leader has exited, the excess is only dropped. + // Output past the limit stops a running group; once the leader has exited it only drops the excess. Either way the + // output is incomplete, so the call settles as ENOBUFS, never as a success. + let truncated = false; const collect = (chunks: Buffer[], add: (bytes: number) => number) => (chunk: Buffer) => { - if (add(chunk.length) > maxBuffer) stop('output-limit'); + if (add(chunk.length) > maxBuffer) { truncated = true; stop('output-limit'); } else chunks.push(chunk); }; child.stdout!.on('data', collect(out, bytes => (outBytes += bytes))); @@ -163,6 +165,11 @@ export function runInProcessGroup(file: string, args: readonly string[], { name: 'AbortError', code: 'ABORT_ERR' }) }); return; } + if (truncated && !stopped) { + resolve({ status: null, stdout, stderr, error: Object.assign( + new Error(`${file} ${args[0] ?? ''} exceeded its output limit.`), { code: 'ENOBUFS' }) }); + return; + } if (stopped) { resolve({ status: null, stdout, stderr, error: Object.assign(new Error(stopped === 'timeout' ? `${file} ${args[0] ?? ''} ETIMEDOUT after ${options.timeoutMs} ms.` diff --git a/git/clone.ts b/git/clone.ts index 4b7c2a0d..39877c4d 100644 --- a/git/clone.ts +++ b/git/clone.ts @@ -167,6 +167,16 @@ export interface AsyncCloneOptions extends CloneOptions { * group has exited and the partial clone is removed. */ export async function createTaskCloneAsync(options: AsyncCloneOptions): Promise { + const signal = options.signal; + try { return await runCloneSteps(options); } + catch (error) { + // However the abort surfaced (during a call, at a pause, or as the next call refusing to start), it is a cancel. + if (signal?.aborted && (error as Error | undefined)?.name !== 'AbortError') + throw Object.assign(new Error('Task clone was cancelled.', { cause: error }), { name: 'AbortError', code: 'ABORT_ERR' }); + throw error; + } +} +async function runCloneSteps(options: AsyncCloneOptions): Promise { const env = gitEnvironment(), steps = cloneSteps(options), signal = options.signal; const aborted = () => Object.assign(new Error('Task clone was cancelled.'), { name: 'AbortError', code: 'ABORT_ERR' }); if (signal?.aborted) throw aborted(); diff --git a/test/agent-clone-async.test.ts b/test/agent-clone-async.test.ts index 9e1fb84f..a2d2d181 100644 --- a/test/agent-clone-async.test.ts +++ b/test/agent-clone-async.test.ts @@ -2,8 +2,19 @@ import { execFileSync } from 'node:child_process'; import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { afterEach, describe, expect, it } from 'vitest'; -import { assertTaskClone, createTaskCloneAsync } from '../git/clone.ts'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +// spawn passes through; a test can act one turn after a chosen Git call exits, before the next one starts. +const spawnHook = vi.hoisted(() => ({ afterExit: undefined as undefined | ((args: readonly string[]) => void) })); +vi.mock('node:child_process', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, spawn: ((file: string, args: readonly string[], options: object) => { + const child = actual.spawn(file, args, options), afterExit = spawnHook.afterExit; + if (afterExit) child.once('exit', () => setImmediate(() => afterExit(args))); + return child; + }) as typeof actual.spawn }; +}); +const { assertTaskClone, createTaskCloneAsync } = await import('../git/clone.ts'); import type { ProcessGroup } from '../agents/process-group.ts'; // The asynchronous clone (#51 item 5): every Git call runs in its own process group, and an abort settles only after @@ -35,7 +46,10 @@ const withStubbornClone = async (root: string, started: string, run: () => Pr process.env.PATH = `${shim}:${path}`; try { return await run(); } finally { process.env.PATH = path; } }; -afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); +afterEach(() => { + spawnHook.afterExit = undefined; + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); describe('asynchronous task clone', () => { it('clones like the synchronous helper and reports each Git process group, all gone once it settles', async () => { @@ -72,6 +86,15 @@ describe('asynchronous task clone', () => { expect(readdirSync(input.parent)).toEqual([]); }, 30_000); + it('reports an abort that lands between two Git calls as a cancel, and removes the partial clone', async () => { + const input = fixture(), controller = new AbortController(); + // Aborted right after `git clone` exits: the next Git call never starts. + spawnHook.afterExit = args => { if (args.includes('clone')) controller.abort(); }; + const error = await createTaskCloneAsync({ ...input, signal: controller.signal }).then(() => undefined, caught => caught); + expect(error).toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }); + expect(readdirSync(input.parent)).toEqual([]); + }); + it('spawns nothing and creates nothing when the signal is already aborted', async () => { const input = fixture(), groups: ProcessGroup[] = []; await expect(createTaskCloneAsync({ ...input, signal: AbortSignal.abort(), diff --git a/test/agent-storage-async.test.ts b/test/agent-storage-async.test.ts index f7aecf6e..82dd64e6 100644 --- a/test/agent-storage-async.test.ts +++ b/test/agent-storage-async.test.ts @@ -9,12 +9,17 @@ import type { ProcessGroup } from '../agents/process-group.ts'; // The asynchronous storage allocation (#51 item 5), against a fake Docker CLI that keeps its objects as files, so the // abort path runs without a daemon. The real-Docker suite covers the same path against Docker itself. // spawn passes through unless a test makes it throw, as it does synchronously for errors such as ENOMEM. -const spawnFault = vi.hoisted(() => ({ on: undefined as undefined | ((args: readonly string[]) => boolean) })); +const spawnFault = vi.hoisted(() => ({ on: undefined as undefined | ((args: readonly string[]) => boolean), + afterExit: undefined as undefined | ((args: readonly string[]) => void) })); vi.mock('node:child_process', async importOriginal => { const actual = await importOriginal(); return { ...actual, spawn: ((file: string, args: readonly string[], options: object) => { if (spawnFault.on?.(args)) throw Object.assign(new Error('spawn ENOMEM'), { code: 'ENOMEM', syscall: 'spawn' }); - return actual.spawn(file, args, options); + const child = actual.spawn(file, args, options); + // One turn after the exit, once the helper has finished with this call and before the next one starts. + const afterExit = spawnFault.afterExit; + if (afterExit) child.once('exit', () => setImmediate(() => afterExit(args))); + return child; }) as typeof actual.spawn }; }); vi.mock('../agents/container/image.ts', async importOriginal => ({ @@ -95,7 +100,7 @@ beforeEach(() => { process.env.PATH = `${bin}:${path}`; }); afterEach(() => { - spawnFault.on = undefined; + spawnFault.on = undefined; spawnFault.afterExit = undefined; process.env.PATH = path; rmSync(root, { recursive: true, force: true }); }); @@ -152,6 +157,17 @@ describe('asynchronous task storage allocation', () => { expect(hasLiveTaskStorage(RUNNER)).toBe(false); }); + it('reports an abort that lands between two Docker calls as a cancel, and removes what it created', async () => { + const controller = new AbortController(); + // Aborted right after the work volume's create exits: the next call (the reuse check) never starts. + spawnFault.afterExit = args => { if (args[0] === 'volume' && args[1] === 'create') controller.abort(); }; + const error = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), { signal: controller.signal }) + .then(() => undefined, caught => caught); + expect(error).toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }); + expect(stored()).toEqual([]); + expect(hasLiveTaskStorage(RUNNER)).toBe(false); + }); + it('creates nothing when the signal is already aborted', async () => { const groups: ProcessGroup[] = []; await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), From bd956a6fad6f35c12c98ba8aba1bf0eb6718cd9b Mon Sep 17 00:00:00 2001 From: mchwang Date: Mon, 28 Sep 2026 16:31:38 -0700 Subject: [PATCH 12/12] Reject deadlines longer than a Node timer; make the between-calls abort tests deterministic setTimeout treats delays above 2^31-1 ms as 1 ms, so a larger timeoutMs made the async allocation time out at once while the sync one worked. Both now refuse it up front, and runInProcessGroup reports it as not started. The between-calls abort tests aborted one turn after a child's exit, which in about a quarter of runs landed during the next call instead. They now wrap runInProcessGroup and abort after one call settles and before the next is requested, and assert that exactly one call was refused before it started. Found by a second independent review agent. Co-Authored-By: Claude Opus 5.5 --- agents/container/storage.ts | 4 +++- agents/process-group.ts | 6 +++-- test/agent-clone-async.test.ts | 31 +++++++++++++----------- test/agent-process-group.test.ts | 2 ++ test/agent-storage-async.test.ts | 41 ++++++++++++++++++++++---------- 5 files changed, 54 insertions(+), 30 deletions(-) diff --git a/agents/container/storage.ts b/agents/container/storage.ts index acdd9b2f..7f61bdc9 100644 --- a/agents/container/storage.ts +++ b/agents/container/storage.ts @@ -7,7 +7,7 @@ import { assertTaskClone } from '../../git/clone.ts'; import { assertBuiltAgentImage } from './image.ts'; import { createOutcomeUnknown, DOCKER_ID } from '../client-outcome.ts'; import { DockerError, pause, runDocker, type DockerOutcome } from '../docker.ts'; -import { runInProcessGroup, type ProcessGroup } from '../process-group.ts'; +import { MAXIMUM_TIMER_MS, runInProcessGroup, type ProcessGroup } from '../process-group.ts'; import { ALLOCATION_IN_USE, allocationListCommands, assertResourceOwner, claimAllocationId, hasOwnerLabels, ownerLabelArgs, releaseAllocationId, type ResourceOwner, UUID_V4 } from '../labels.ts'; @@ -49,6 +49,8 @@ const validLimit = (value: number, name: string) => { }; const createDeadline = (timeoutMs: number) => { validLimit(timeoutMs, 'timeoutMs'); + // Both variants share these steps, and the asynchronous one arms Node timers, which cannot wait longer than this. + if (timeoutMs > MAXIMUM_TIMER_MS) throw new Error(`timeoutMs must be at most ${MAXIMUM_TIMER_MS}.`); const deadline = performance.now() + timeoutMs; return () => { const value = Math.ceil(deadline - performance.now()); diff --git a/agents/process-group.ts b/agents/process-group.ts index 836677a5..db7bad60 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -23,6 +23,8 @@ export interface ProcessGroupOptions { } const DEFAULT_GRACE_MS = 5_000; +/** The longest delay a Node timer honours; `setTimeout` treats anything longer as 1 ms. */ +export const MAXIMUM_TIMER_MS = 2 ** 31 - 1; const DEFAULT_MAX_BUFFER = 16 * 1024 * 1024; // After the leader exits, how long to keep killing and polling other members of its group before giving up. const DRAIN_LIMIT_MS = 10_000; @@ -62,8 +64,8 @@ const drainGroup = async (pgid: number) => { */ export function runInProcessGroup(file: string, args: readonly string[], options: ProcessGroupOptions): Promise { - if (!Number.isSafeInteger(options.timeoutMs) || options.timeoutMs < 1) - return Promise.resolve(notStarted('Process deadline must be a positive integer.')); + if (!Number.isSafeInteger(options.timeoutMs) || options.timeoutMs < 1 || options.timeoutMs > MAXIMUM_TIMER_MS) + return Promise.resolve(notStarted(`Process deadline must be a positive integer of at most ${MAXIMUM_TIMER_MS} ms.`)); if (options.signal?.aborted) return Promise.resolve(notStarted(`${file} was cancelled before it started.`)); const graceMs = options.graceMs ?? DEFAULT_GRACE_MS, maxBuffer = options.maxBuffer ?? DEFAULT_MAX_BUFFER; return new Promise(resolve => { diff --git a/test/agent-clone-async.test.ts b/test/agent-clone-async.test.ts index a2d2d181..2ca95141 100644 --- a/test/agent-clone-async.test.ts +++ b/test/agent-clone-async.test.ts @@ -4,15 +4,16 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, describe, expect, it, vi } from 'vitest'; -// spawn passes through; a test can act one turn after a chosen Git call exits, before the next one starts. -const spawnHook = vi.hoisted(() => ({ afterExit: undefined as undefined | ((args: readonly string[]) => void) })); -vi.mock('node:child_process', async importOriginal => { - const actual = await importOriginal(); - return { ...actual, spawn: ((file: string, args: readonly string[], options: object) => { - const child = actual.spawn(file, args, options), afterExit = spawnHook.afterExit; - if (afterExit) child.once('exit', () => setImmediate(() => afterExit(args))); - return child; - }) as typeof actual.spawn }; +// Every Git call passes through; a test can act after one call settles and before the next is requested. +const calls = vi.hoisted(() => ({ made: [] as string[][], afterCall: undefined as undefined | ((args: readonly string[]) => void) })); +vi.mock('../agents/process-group.ts', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, runInProcessGroup: async (...call: Parameters) => { + calls.made.push([...call[1]]); + const outcome = await actual.runInProcessGroup(...call); + calls.afterCall?.(call[1]); + return outcome; + } }; }); const { assertTaskClone, createTaskCloneAsync } = await import('../git/clone.ts'); import type { ProcessGroup } from '../agents/process-group.ts'; @@ -47,7 +48,7 @@ const withStubbornClone = async (root: string, started: string, run: () => Pr try { return await run(); } finally { process.env.PATH = path; } }; afterEach(() => { - spawnHook.afterExit = undefined; + calls.afterCall = undefined; calls.made = []; for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); @@ -87,11 +88,13 @@ describe('asynchronous task clone', () => { }, 30_000); it('reports an abort that lands between two Git calls as a cancel, and removes the partial clone', async () => { - const input = fixture(), controller = new AbortController(); - // Aborted right after `git clone` exits: the next Git call never starts. - spawnHook.afterExit = args => { if (args.includes('clone')) controller.abort(); }; - const error = await createTaskCloneAsync({ ...input, signal: controller.signal }).then(() => undefined, caught => caught); + const input = fixture(), controller = new AbortController(), groups: ProcessGroup[] = []; + // Aborted after `git clone` settles and before the next Git call is requested: that call never starts. + calls.afterCall = args => { if (args.includes('clone')) controller.abort(); }; + const error = await createTaskCloneAsync({ ...input, signal: controller.signal, + onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught); expect(error).toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }); + expect(calls.made.length - groups.length).toBe(1); expect(readdirSync(input.parent)).toEqual([]); }); diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts index f7bb2ea3..844a2004 100644 --- a/test/agent-process-group.test.ts +++ b/test/agent-process-group.test.ts @@ -130,6 +130,8 @@ describe('runInProcessGroup', () => { for (const outcome of [ await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 5_000, signal: AbortSignal.abort(), onProcessGroup }), await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 0, onProcessGroup }), + // Longer than a Node timer can wait: accepting it would time the call out after 1 ms. + await runInProcessGroup('sleep', ['1'], { env, timeoutMs: 2 ** 31, onProcessGroup }), ]) { expect((outcome.error as NodeJS.ErrnoException).code).toBe(NOT_STARTED); expect(createOutcomeUnknown({ status: outcome.status, cause: outcome.error })).toBe(false); diff --git a/test/agent-storage-async.test.ts b/test/agent-storage-async.test.ts index 82dd64e6..bc0cb0ed 100644 --- a/test/agent-storage-async.test.ts +++ b/test/agent-storage-async.test.ts @@ -9,17 +9,23 @@ import type { ProcessGroup } from '../agents/process-group.ts'; // The asynchronous storage allocation (#51 item 5), against a fake Docker CLI that keeps its objects as files, so the // abort path runs without a daemon. The real-Docker suite covers the same path against Docker itself. // spawn passes through unless a test makes it throw, as it does synchronously for errors such as ENOMEM. -const spawnFault = vi.hoisted(() => ({ on: undefined as undefined | ((args: readonly string[]) => boolean), - afterExit: undefined as undefined | ((args: readonly string[]) => void) })); +const spawnFault = vi.hoisted(() => ({ on: undefined as undefined | ((args: readonly string[]) => boolean) })); +// Every Docker call passes through; a test can act after one call settles and before the next is requested. +const calls = vi.hoisted(() => ({ made: [] as string[][], afterCall: undefined as undefined | ((args: readonly string[]) => void) })); +vi.mock('../agents/process-group.ts', async importOriginal => { + const actual = await importOriginal(); + return { ...actual, runInProcessGroup: async (...call: Parameters) => { + calls.made.push([...call[1]]); + const outcome = await actual.runInProcessGroup(...call); + calls.afterCall?.(call[1]); + return outcome; + } }; +}); vi.mock('node:child_process', async importOriginal => { const actual = await importOriginal(); return { ...actual, spawn: ((file: string, args: readonly string[], options: object) => { if (spawnFault.on?.(args)) throw Object.assign(new Error('spawn ENOMEM'), { code: 'ENOMEM', syscall: 'spawn' }); - const child = actual.spawn(file, args, options); - // One turn after the exit, once the helper has finished with this call and before the next one starts. - const afterExit = spawnFault.afterExit; - if (afterExit) child.once('exit', () => setImmediate(() => afterExit(args))); - return child; + return actual.spawn(file, args, options); }) as typeof actual.spawn }; }); vi.mock('../agents/container/image.ts', async importOriginal => ({ @@ -100,7 +106,7 @@ beforeEach(() => { process.env.PATH = `${bin}:${path}`; }); afterEach(() => { - spawnFault.on = undefined; spawnFault.afterExit = undefined; + spawnFault.on = undefined; calls.afterCall = undefined; calls.made = []; process.env.PATH = path; rmSync(root, { recursive: true, force: true }); }); @@ -158,16 +164,25 @@ describe('asynchronous task storage allocation', () => { }); it('reports an abort that lands between two Docker calls as a cancel, and removes what it created', async () => { - const controller = new AbortController(); - // Aborted right after the work volume's create exits: the next call (the reuse check) never starts. - spawnFault.afterExit = args => { if (args[0] === 'volume' && args[1] === 'create') controller.abort(); }; - const error = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), { signal: controller.signal }) - .then(() => undefined, caught => caught); + const controller = new AbortController(), groups: ProcessGroup[] = []; + // Aborted after the work volume's create settles and before the next call (the reuse check) is requested. + calls.afterCall = args => { if (args[0] === 'volume' && args[1] === 'create') controller.abort(); }; + const error = await prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), + { signal: controller.signal, onProcessGroup: group => { groups.push(group); } }).then(() => undefined, caught => caught); expect(error).toMatchObject({ name: 'AbortError', code: 'ABORT_ERR' }); + // Exactly one call was refused before it started (the reuse check); every other call ran and reported its group. + expect(calls.made.length - groups.length).toBe(1); expect(stored()).toEqual([]); expect(hasLiveTaskStorage(RUNNER)).toBe(false); }); + it('refuses a deadline longer than a Node timer can wait, before any Docker call', async () => { + await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(), { timeoutMs: 2 ** 31 })) + .rejects.toThrow('at most'); + expect(calls.made).toEqual([]); + expect(stored()).toEqual([]); + }); + it('creates nothing when the signal is already aborted', async () => { const groups: ProcessGroup[] = []; await expect(prepareTaskFilesystemsAsync(clone(), LIMITS, IMAGE, owner(),