468 lines
13 KiB
TypeScript
468 lines
13 KiB
TypeScript
import { mkdir } from 'node:fs/promises';
|
|
import path from 'node:path';
|
|
import type { Readable } from 'node:stream';
|
|
import { execa } from 'execa';
|
|
|
|
import { env } from '../env';
|
|
|
|
type CommandResult = {
|
|
exitCode: number;
|
|
output: string;
|
|
};
|
|
|
|
const environmentArgs = (environment: Record<string, string>) =>
|
|
Object.entries(environment).flatMap(([name, value]) => [
|
|
'-e',
|
|
`${name}=${value}`,
|
|
]);
|
|
|
|
const networkArgs = () => (env.network ? ['--network', env.network] : []);
|
|
|
|
const containerRuntime = () => env.containerRuntime;
|
|
|
|
// `docker run` reuses a stale local `:latest` forever, so without an explicit
|
|
// pull the job image never updates in production. Pull once per worker process
|
|
// (i.e. once per deploy/restart) so a fresh worker always runs a fresh job
|
|
// image. Best-effort: if the registry is unreachable we fall back to whatever
|
|
// image is present locally rather than failing the job.
|
|
let jobImagePullPromise: Promise<void> | undefined;
|
|
export const ensureJobImagePulled = () => {
|
|
jobImagePullPromise ??= (async () => {
|
|
try {
|
|
await execa(containerRuntime(), ['pull', env.jobImage], {
|
|
reject: false,
|
|
stdin: 'ignore',
|
|
});
|
|
} catch {
|
|
// Ignore: keep running with the locally cached image.
|
|
}
|
|
})();
|
|
return jobImagePullPromise;
|
|
};
|
|
|
|
// execa with `reject: false` resolves (does not throw) even when the runtime
|
|
// binary is missing (ENOENT) — `exitCode` is then `undefined`. Coercing that to
|
|
// 0 makes a failed spawn look like a successful empty run, which is exactly how
|
|
// a worker image without a `docker` CLI silently produced empty agent
|
|
// responses. Normalize so any spawn failure is a non-zero exit carrying the
|
|
// real reason.
|
|
export const normalizeRunResult = (
|
|
// Declared nullable on purpose: execa's types claim these are always present,
|
|
// but on a spawn failure (e.g. missing `docker` binary) `exitCode`/`all` are
|
|
// actually undefined at runtime.
|
|
result: { exitCode?: number; shortMessage?: string },
|
|
output: string | undefined,
|
|
redact: (value: string) => string,
|
|
): CommandResult => {
|
|
const text = output ?? '';
|
|
if (result.exitCode == null) {
|
|
const reason = result.shortMessage ?? 'container runtime failed to start';
|
|
return {
|
|
exitCode: 1,
|
|
output: redact(`${text}${text ? '\n' : ''}${reason}`),
|
|
};
|
|
}
|
|
return { exitCode: result.exitCode, output: redact(text) };
|
|
};
|
|
|
|
const hostWorkspacePath = (workdir: string) => {
|
|
if (!env.hostWorkdir) return workdir;
|
|
const workerRoot = path.resolve(env.workdir);
|
|
const resolvedWorkdir = path.resolve(workdir);
|
|
const relative = path.relative(workerRoot, resolvedWorkdir);
|
|
if (relative.startsWith('..') || path.isAbsolute(relative)) {
|
|
return workdir;
|
|
}
|
|
return path.join(env.hostWorkdir, relative);
|
|
};
|
|
|
|
export const containerVolumeSuffix = () =>
|
|
env.containerVolumeOptions ??
|
|
(containerRuntime().endsWith('podman') ? 'Z' : undefined);
|
|
|
|
export { hostWorkspacePath };
|
|
|
|
export const jobWorkspaceVolumeSpec = (
|
|
workdir: string,
|
|
containerHome = '/workspace',
|
|
) => {
|
|
const volumeOptions =
|
|
env.containerVolumeOptions ??
|
|
(containerRuntime().endsWith('podman') ? 'Z' : undefined);
|
|
const source = hostWorkspacePath(workdir);
|
|
return volumeOptions
|
|
? `${source}:${containerHome}:${volumeOptions}`
|
|
: `${source}:${containerHome}`;
|
|
};
|
|
|
|
export const runInJobContainer = async (args: {
|
|
workdir: string;
|
|
containerHome?: string;
|
|
containerCwd?: string;
|
|
command: string[];
|
|
environment: Record<string, string>;
|
|
redact: (value: string) => string;
|
|
timeoutMs: number;
|
|
}): Promise<CommandResult> => {
|
|
await ensureJobImagePulled();
|
|
const result = await execa(
|
|
containerRuntime(),
|
|
[
|
|
'run',
|
|
'--rm',
|
|
'--memory',
|
|
'4g',
|
|
'--cpus',
|
|
'2',
|
|
...networkArgs(),
|
|
...environmentArgs(args.environment),
|
|
'-v',
|
|
jobWorkspaceVolumeSpec(args.workdir, args.containerHome),
|
|
'-w',
|
|
args.containerCwd ?? '/workspace/repo',
|
|
env.jobImage,
|
|
...args.command,
|
|
],
|
|
{
|
|
all: true,
|
|
reject: false,
|
|
stdin: 'ignore',
|
|
timeout: args.timeoutMs,
|
|
},
|
|
);
|
|
return normalizeRunResult(result, result.all, args.redact);
|
|
};
|
|
|
|
export const startWorkspaceContainer = async (args: {
|
|
workdir: string;
|
|
containerHome?: string;
|
|
containerCwd?: string;
|
|
containerName: string;
|
|
environment: Record<string, string>;
|
|
command?: string[];
|
|
publishTcpPort?: number;
|
|
}) => {
|
|
await ensureJobImagePulled();
|
|
await execa(containerRuntime(), ['rm', '-f', args.containerName], {
|
|
reject: false,
|
|
});
|
|
const result = await execa(
|
|
containerRuntime(),
|
|
[
|
|
'run',
|
|
'-d',
|
|
'--name',
|
|
args.containerName,
|
|
'--memory',
|
|
'4g',
|
|
'--cpus',
|
|
'2',
|
|
...networkArgs(),
|
|
...(args.publishTcpPort
|
|
? ['-p', `127.0.0.1::${args.publishTcpPort}`]
|
|
: []),
|
|
...environmentArgs(args.environment),
|
|
'-v',
|
|
jobWorkspaceVolumeSpec(args.workdir, args.containerHome),
|
|
'-w',
|
|
args.containerCwd ?? '/workspace/repo',
|
|
env.jobImage,
|
|
...(args.command ?? ['sleep', 'infinity']),
|
|
],
|
|
{ all: true, stdin: 'ignore' },
|
|
);
|
|
return {
|
|
containerId: result.stdout.trim(),
|
|
containerName: args.containerName,
|
|
hostPort: args.publishTcpPort
|
|
? await getPublishedPort(args.containerName, args.publishTcpPort)
|
|
: undefined,
|
|
};
|
|
};
|
|
|
|
const getPublishedPort = async (
|
|
containerName: string,
|
|
containerPort: number,
|
|
) => {
|
|
const result = await execa(
|
|
containerRuntime(),
|
|
['port', containerName, `${containerPort}/tcp`],
|
|
{ all: true, reject: false, stdin: 'ignore' },
|
|
);
|
|
const output = result.all.trim();
|
|
const match = /:(\d+)\s*$/.exec(output);
|
|
if (!match?.[1]) {
|
|
throw new Error(
|
|
`Could not determine published port for ${containerName}:${containerPort}.`,
|
|
);
|
|
}
|
|
return match[1];
|
|
};
|
|
|
|
export const execInWorkspaceContainer = async (args: {
|
|
containerName: string;
|
|
command: string[];
|
|
environment?: Record<string, string>;
|
|
redact: (value: string) => string;
|
|
timeoutMs: number;
|
|
}): Promise<CommandResult> => {
|
|
const result = await execa(
|
|
containerRuntime(),
|
|
[
|
|
'exec',
|
|
...(args.environment ? environmentArgs(args.environment) : []),
|
|
args.containerName,
|
|
...args.command,
|
|
],
|
|
{
|
|
all: true,
|
|
reject: false,
|
|
stdin: 'ignore',
|
|
timeout: args.timeoutMs,
|
|
},
|
|
);
|
|
return {
|
|
exitCode: result.exitCode ?? 0,
|
|
output: args.redact(result.all),
|
|
};
|
|
};
|
|
|
|
// Shared line-streaming + result normalization for a started subprocess
|
|
// (used by both `docker run` and `docker exec` paths).
|
|
type StreamingSubprocess = {
|
|
stdout: Readable | null;
|
|
stderr: Readable | null;
|
|
} & Promise<{ exitCode?: number; shortMessage?: string; all?: string }>;
|
|
|
|
const streamSubprocess = async (
|
|
subprocess: StreamingSubprocess,
|
|
redact: (value: string) => string,
|
|
onStdoutLine?: (line: string) => Promise<void>,
|
|
onStderrLine?: (line: string) => Promise<void>,
|
|
): Promise<CommandResult> => {
|
|
let stdoutBuffer = '';
|
|
let stderrBuffer = '';
|
|
const output: string[] = [];
|
|
let lineHandlers = Promise.resolve();
|
|
const consume = async (
|
|
chunk: Buffer,
|
|
source: 'stdout' | 'stderr',
|
|
handler?: (line: string) => Promise<void>,
|
|
) => {
|
|
output.push(chunk.toString('utf8'));
|
|
const next = `${source === 'stdout' ? stdoutBuffer : stderrBuffer}${chunk.toString('utf8')}`;
|
|
const lines = next.split(/\r?\n/);
|
|
const remainder = lines.pop() ?? '';
|
|
if (source === 'stdout') stdoutBuffer = remainder;
|
|
else stderrBuffer = remainder;
|
|
for (const line of lines) {
|
|
if (handler) await handler(redact(line));
|
|
}
|
|
};
|
|
subprocess.stdout?.on('data', (chunk: Buffer) => {
|
|
lineHandlers = lineHandlers.then(() =>
|
|
consume(chunk, 'stdout', onStdoutLine),
|
|
);
|
|
});
|
|
subprocess.stderr?.on('data', (chunk: Buffer) => {
|
|
lineHandlers = lineHandlers.then(() =>
|
|
consume(chunk, 'stderr', onStderrLine),
|
|
);
|
|
});
|
|
let result: Awaited<StreamingSubprocess>;
|
|
try {
|
|
result = await subprocess;
|
|
} catch (error) {
|
|
await lineHandlers;
|
|
const outputText = output.join('');
|
|
const message =
|
|
error instanceof Error ? error.message : 'Container command failed.';
|
|
return {
|
|
exitCode: 1,
|
|
output: redact(`${outputText}${outputText ? '\n' : ''}${message}`),
|
|
};
|
|
}
|
|
await lineHandlers;
|
|
if (stdoutBuffer && onStdoutLine) await onStdoutLine(redact(stdoutBuffer));
|
|
if (stderrBuffer && onStderrLine) await onStderrLine(redact(stderrBuffer));
|
|
return normalizeRunResult(result, output.join(''), redact);
|
|
};
|
|
|
|
export const streamInJobContainer = async (args: {
|
|
workdir: string;
|
|
containerHome?: string;
|
|
containerCwd?: string;
|
|
command: string[];
|
|
environment: Record<string, string>;
|
|
redact: (value: string) => string;
|
|
timeoutMs: number;
|
|
onStdoutLine?: (line: string) => Promise<void>;
|
|
onStderrLine?: (line: string) => Promise<void>;
|
|
}): Promise<CommandResult> => {
|
|
await ensureJobImagePulled();
|
|
const subprocess = execa(
|
|
containerRuntime(),
|
|
[
|
|
'run',
|
|
'--rm',
|
|
'--memory',
|
|
'4g',
|
|
'--cpus',
|
|
'2',
|
|
...networkArgs(),
|
|
...environmentArgs(args.environment),
|
|
'-v',
|
|
jobWorkspaceVolumeSpec(args.workdir, args.containerHome),
|
|
'-w',
|
|
args.containerCwd ?? '/workspace/repo',
|
|
env.jobImage,
|
|
...args.command,
|
|
],
|
|
{
|
|
all: true,
|
|
reject: false,
|
|
stdin: 'ignore',
|
|
timeout: args.timeoutMs,
|
|
},
|
|
);
|
|
return streamSubprocess(
|
|
subprocess,
|
|
args.redact,
|
|
args.onStdoutLine,
|
|
args.onStderrLine,
|
|
);
|
|
};
|
|
|
|
// Per-user persistent "box" container that all of a user's threads exec into
|
|
// (Phase 2). Started once, reused; the home volume persists state across stops.
|
|
export const userContainerName = (username: string) =>
|
|
`spoon-box-${username.replace(/[^a-zA-Z0-9_.-]/g, '-')}`;
|
|
|
|
export const ensureUserContainer = async (args: {
|
|
username: string;
|
|
workdir: string;
|
|
containerHome: string;
|
|
}): Promise<string> => {
|
|
await ensureJobImagePulled();
|
|
const name = userContainerName(args.username);
|
|
const inspect = await execa(
|
|
containerRuntime(),
|
|
['inspect', '-f', '{{.State.Running}}', name],
|
|
{ reject: false, stdin: 'ignore' },
|
|
);
|
|
if (inspect.exitCode === 0 && inspect.stdout.trim() === 'true') return name;
|
|
// The box mounts the per-user home, but it's created before the thread's clone
|
|
// populates it — ensure it exists first, since podman (unlike docker) refuses to
|
|
// bind-mount a missing source directory (statfs: no such file or directory).
|
|
await mkdir(args.workdir, { recursive: true });
|
|
// Not running: remove any stale container, then start fresh.
|
|
await execa(containerRuntime(), ['rm', '-f', name], { reject: false });
|
|
await execa(
|
|
containerRuntime(),
|
|
[
|
|
'run',
|
|
'-d',
|
|
'--init',
|
|
'--name',
|
|
name,
|
|
'--memory',
|
|
'4g',
|
|
'--cpus',
|
|
'2',
|
|
...networkArgs(),
|
|
'-v',
|
|
jobWorkspaceVolumeSpec(args.workdir, args.containerHome),
|
|
'-w',
|
|
args.containerHome,
|
|
env.jobImage,
|
|
'sleep',
|
|
'infinity',
|
|
],
|
|
{ stdin: 'ignore' },
|
|
);
|
|
return name;
|
|
};
|
|
|
|
export const streamExecInContainer = async (args: {
|
|
containerName: string;
|
|
command: string[];
|
|
environment: Record<string, string>;
|
|
containerCwd: string;
|
|
redact: (value: string) => string;
|
|
timeoutMs: number;
|
|
onStdoutLine?: (line: string) => Promise<void>;
|
|
onStderrLine?: (line: string) => Promise<void>;
|
|
}): Promise<CommandResult> => {
|
|
const subprocess = execa(
|
|
containerRuntime(),
|
|
[
|
|
'exec',
|
|
...environmentArgs(args.environment),
|
|
'-w',
|
|
args.containerCwd,
|
|
args.containerName,
|
|
...args.command,
|
|
],
|
|
{ all: true, reject: false, stdin: 'ignore', timeout: args.timeoutMs },
|
|
);
|
|
return streamSubprocess(
|
|
subprocess,
|
|
args.redact,
|
|
args.onStdoutLine,
|
|
args.onStderrLine,
|
|
);
|
|
};
|
|
|
|
export const runExecInContainer = async (args: {
|
|
containerName: string;
|
|
command: string[];
|
|
environment: Record<string, string>;
|
|
containerCwd: string;
|
|
redact: (value: string) => string;
|
|
timeoutMs: number;
|
|
}): Promise<CommandResult> => {
|
|
const result = await execa(
|
|
containerRuntime(),
|
|
[
|
|
'exec',
|
|
...environmentArgs(args.environment),
|
|
'-w',
|
|
args.containerCwd,
|
|
args.containerName,
|
|
...args.command,
|
|
],
|
|
{ all: true, reject: false, stdin: 'ignore', timeout: args.timeoutMs },
|
|
);
|
|
return normalizeRunResult(result, result.all, args.redact);
|
|
};
|
|
|
|
export const stopWorkspaceContainer = async (containerName: string) => {
|
|
await execa(containerRuntime(), ['rm', '-f', containerName], {
|
|
reject: false,
|
|
});
|
|
};
|
|
|
|
export const inspectWorkspaceContainer = async (containerName: string) => {
|
|
const result = await execa(containerRuntime(), ['inspect', containerName], {
|
|
all: true,
|
|
reject: false,
|
|
});
|
|
return {
|
|
exists: result.exitCode === 0,
|
|
output: result.all,
|
|
};
|
|
};
|
|
|
|
export const listWorkspaceContainerNames = async (prefix: string) => {
|
|
const result = await execa(
|
|
containerRuntime(),
|
|
['ps', '-a', '--format', '{{.Names}}'],
|
|
{ all: true, reject: false },
|
|
);
|
|
if (result.exitCode !== 0) return [];
|
|
return result.all
|
|
.split('\n')
|
|
.map((line) => line.trim())
|
|
.filter((line) => line.startsWith(prefix));
|
|
};
|