fix(runtime): harden sibling port allocation

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-14 07:17:00 +00:00
1 parent aac6ce82e1
commit 54b4e0c2fb
2 files changed
+874 -49

No files matched your search

@@ -43,6 +43,7 @@ import {
sanitizeRuntimeServiceBaseEnv,
startRuntimeServicesForWorkspaceControl,
stopRuntimeServicesForExecutionWorkspace,
WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS,
type RealizedExecutionWorkspace,
} from "../services/workspace-runtime.ts";
import {
@@ -282,6 +283,48 @@ function buildWorkspace(cwd: string): RealizedExecutionWorkspace {
};
}
async function closeNetServer(server: net.Server) {
if (!server.listening) return;
await new Promise<void>((resolve, reject) => {
server.close((error) => error ? reject(error) : resolve());
});
}
async function listenOnPort(port: number) {
const server = net.createServer();
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(port, "127.0.0.1", resolve);
});
return server;
}
async function findFreePort() {
const server = await listenOnPort(0);
const address = server.address();
const port = typeof address === "object" && address ? address.port : null;
await closeNetServer(server);
if (!port) throw new Error("Failed to find a free test port");
return port;
}
async function reserveContiguousPorts(count: number) {
for (let attempt = 0; attempt < 100; attempt += 1) {
const basePort = await findFreePort();
if (basePort + count - 1 > 65_535) continue;
const servers: net.Server[] = [];
try {
for (let offset = 0; offset < count; offset += 1) {
servers.push(await listenOnPort(basePort + offset));
}
return { basePort, servers };
} catch {
await Promise.all(servers.map((server) => closeNetServer(server).catch(() => undefined)));
}
}
throw new Error(`Failed to reserve ${count} contiguous test ports`);
}
function createWorkspaceOperationRecorderDouble() {
const operations: Array<{
phase: string;
@@ -5620,6 +5663,472 @@ describeEmbeddedPostgres("workspace runtime service control persistence", () =>
else process.env.PAPERCLIP_INSTANCE_ID = previousPaperclipInstanceId;
}
}, 15_000);
async function createRuntimeFixture(input?: {
companyId?: string;
workspaceModes?: Array<"isolated_workspace" | "shared_workspace">;
}) {
const companyId = input?.companyId ?? randomUUID();
const agentId = randomUUID();
const projectId = randomUUID();
const projectWorkspaceId = randomUUID();
const baseRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-port-fixture-"));
const workspaceModes = input?.workspaceModes ?? ["isolated_workspace"];
const workspaceRows = await Promise.all(workspaceModes.map(async (mode, index) => {
const cwd = await fs.mkdtemp(path.join(baseRoot, `workspace-${index}-`));
return {
id: randomUUID(),
cwd,
mode,
};
}));
await db.insert(companies).values({
id: companyId,
name: `Runtime ports ${companyId.slice(0, 8)}`,
issuePrefix: `R${companyId.replace(/-/g, "").slice(0, 7).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "Runtime Port Agent",
role: "engineer",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(projects).values({
id: projectId,
companyId,
name: "Runtime port allocation",
status: "in_progress",
});
await db.insert(projectWorkspaces).values({
id: projectWorkspaceId,
companyId,
projectId,
name: "Primary",
sourceType: "local_path",
cwd: baseRoot,
isPrimary: true,
});
await db.insert(executionWorkspaces).values(workspaceRows.map((workspace, index) => ({
id: workspace.id,
companyId,
projectId,
projectWorkspaceId,
mode: workspace.mode,
strategyType: workspace.mode === "isolated_workspace" ? "git_worktree" : "project_primary",
name: `Runtime workspace ${index + 1}`,
status: "active",
providerType: "local_fs",
cwd: workspace.cwd,
providerRef: workspace.cwd,
})));
const realizedWorkspace = (workspace: typeof workspaceRows[number]): RealizedExecutionWorkspace => ({
baseCwd: baseRoot,
source: "task_session",
projectId,
workspaceId: projectWorkspaceId,
repoUrl: null,
repoRef: "HEAD",
strategy: workspace.mode === "isolated_workspace" ? "git_worktree" : "project_primary",
cwd: workspace.cwd,
branchName: null,
worktreePath: workspace.mode === "isolated_workspace" ? workspace.cwd : null,
warnings: [],
created: false,
});
return {
companyId,
agentId,
projectId,
projectWorkspaceId,
workspaces: workspaceRows,
actor: { id: agentId, name: "Runtime Port Agent", companyId },
realizedWorkspace,
cleanup: async () => await fs.rm(baseRoot, { recursive: true, force: true }),
};
}
function fixedPortRuntimeConfig(port: number, command?: string) {
const serverCommand = command ?? [
JSON.stringify(process.execPath),
"-e",
JSON.stringify(
"require('node:http').createServer((_req,res)=>res.end('ok')).listen(Number(process.env.PORT),'127.0.0.1')",
),
].join(" ");
return {
workspaceRuntime: {
services: [{
name: "web",
command: serverCommand,
port: { type: "fixed", value: port, envKey: "PORT" },
readiness: {
type: "http",
urlTemplate: "http://127.0.0.1:{{port}}",
timeoutSec: 5,
intervalMs: 25,
},
expose: { type: "url", urlTemplate: "http://127.0.0.1:{{port}}" },
lifecycle: "shared",
reuseScope: "execution_workspace",
stopPolicy: { type: "manual" },
}],
},
};
}
async function createRuntimeHome() {
const paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-port-home-"));
const previousPaperclipHome = process.env.PAPERCLIP_HOME;
const previousPaperclipInstanceId = process.env.PAPERCLIP_INSTANCE_ID;
process.env.PAPERCLIP_HOME = paperclipHome;
process.env.PAPERCLIP_INSTANCE_ID = `runtime-ports-${randomUUID()}`;
return async () => {
if (previousPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME;
else process.env.PAPERCLIP_HOME = previousPaperclipHome;
if (previousPaperclipInstanceId === undefined) delete process.env.PAPERCLIP_INSTANCE_ID;
else process.env.PAPERCLIP_INSTANCE_ID = previousPaperclipInstanceId;
await fs.rm(paperclipHome, { recursive: true, force: true });
};
}
it("allocates distinct persisted ports for concurrent isolated siblings without stopping either service", async () => {
const fixture = await createRuntimeFixture({
workspaceModes: ["isolated_workspace", "isolated_workspace"],
});
const cleanupRuntimeHome = await createRuntimeHome();
const basePort = await findFreePort();
const config = fixedPortRuntimeConfig(basePort);
const startedWorkspaceIds: string[] = [];
try {
const [first, second] = await Promise.all(fixture.workspaces.map(async (workspace) => {
const result = await startRuntimeServicesForWorkspaceControl({
db,
actor: fixture.actor,
issue: null,
workspace: fixture.realizedWorkspace(workspace),
executionWorkspaceId: workspace.id,
config,
adapterEnv: {},
});
startedWorkspaceIds.push(workspace.id);
return result[0]!;
}));
expect(first.port).toBe(basePort);
expect(second.port).not.toBe(first.port);
expect(second.port).toBeGreaterThan(basePort);
expect(first.url).toBe(`http://127.0.0.1:${first.port}`);
expect(second.url).toBe(`http://127.0.0.1:${second.port}`);
await expect(fetch(first.url!)).resolves.toMatchObject({ ok: true });
await expect(fetch(second.url!)).resolves.toMatchObject({ ok: true });
const persisted = await db
.select()
.from(workspaceRuntimeServices)
.where(eq(workspaceRuntimeServices.companyId, fixture.companyId));
expect(persisted).toEqual(expect.arrayContaining([
expect.objectContaining({
executionWorkspaceId: fixture.workspaces[0]!.id,
port: first.port,
url: first.url,
status: "running",
}),
expect.objectContaining({
executionWorkspaceId: fixture.workspaces[1]!.id,
port: second.port,
url: second.url,
status: "running",
}),
]));
} finally {
await Promise.all(startedWorkspaceIds.map(async (executionWorkspaceId) => {
const workspace = fixture.workspaces.find((entry) => entry.id === executionWorkspaceId)!;
await stopRuntimeServicesForExecutionWorkspace({
db,
executionWorkspaceId,
workspaceCwd: workspace.cwd,
}).catch(() => undefined);
}));
await cleanupRuntimeHome();
await fixture.cleanup();
}
}, 20_000);
it("reuses a stopped isolated workspace's actual port when it is free", async () => {
const fixture = await createRuntimeFixture();
const cleanupRuntimeHome = await createRuntimeHome();
const workspace = fixture.workspaces[0]!;
const config = fixedPortRuntimeConfig(await findFreePort());
try {
const first = (await startRuntimeServicesForWorkspaceControl({
db,
actor: fixture.actor,
issue: null,
workspace: fixture.realizedWorkspace(workspace),
executionWorkspaceId: workspace.id,
config,
adapterEnv: {},
}))[0]!;
await stopRuntimeServicesForExecutionWorkspace({
db,
executionWorkspaceId: workspace.id,
workspaceCwd: workspace.cwd,
});
const restarted = (await startRuntimeServicesForWorkspaceControl({
db,
actor: fixture.actor,
issue: null,
workspace: fixture.realizedWorkspace(workspace),
executionWorkspaceId: workspace.id,
config,
adapterEnv: {},
}))[0]!;
expect(restarted.id).toBe(first.id);
expect(restarted.port).toBe(first.port);
expect(restarted.url).toBe(first.url);
await expect(fetch(restarted.url!)).resolves.toMatchObject({ ok: true });
} finally {
await stopRuntimeServicesForExecutionWorkspace({
db,
executionWorkspaceId: workspace.id,
workspaceCwd: workspace.cwd,
}).catch(() => undefined);
await cleanupRuntimeHome();
await fixture.cleanup();
}
}, 20_000);
it("retries a bind race on the next bounded port and persists the selected URL", async () => {
const fixture = await createRuntimeFixture();
const cleanupRuntimeHome = await createRuntimeHome();
const workspace = fixture.workspaces[0]!;
const basePort = await findFreePort();
const markerPath = path.join(workspace.cwd, "bind-race.marker");
const blockerPidPath = path.join(workspace.cwd, "bind-race.pid");
const blockerScript = [
"const net=require('node:net');",
"const port=Number(process.argv[1]);",
"net.createServer((socket)=>socket.destroy()).listen(port,'127.0.0.1');",
].join("");
const raceScript = [
"const fs=require('node:fs');",
"const http=require('node:http');",
"const {spawn}=require('node:child_process');",
`const marker=${JSON.stringify(markerPath)};`,
`const pidFile=${JSON.stringify(blockerPidPath)};`,
`const blockerScript=${JSON.stringify(blockerScript)};`,
"const port=Number(process.env.PORT);",
"const start=()=>http.createServer((_req,res)=>res.end('ok')).listen(port,'127.0.0.1');",
"if(!fs.existsSync(marker)){",
"fs.writeFileSync(marker,String(port));",
"const blocker=spawn(process.execPath,['-e',blockerScript,String(port)],{detached:true,stdio:'ignore'});",
"fs.writeFileSync(pidFile,String(blocker.pid));blocker.unref();setTimeout(start,200);",
"}else{start();}",
"setInterval(()=>{},1000);",
].join("");
const command = `${JSON.stringify(process.execPath)} -e ${JSON.stringify(raceScript)}`;
try {
const started = (await startRuntimeServicesForWorkspaceControl({
db,
actor: fixture.actor,
issue: null,
workspace: fixture.realizedWorkspace(workspace),
executionWorkspaceId: workspace.id,
config: fixedPortRuntimeConfig(basePort, command),
adapterEnv: {},
}))[0]!;
expect(await fs.readFile(markerPath, "utf8")).toBe(String(basePort));
expect(started.port).toBeGreaterThan(basePort);
expect(started.url).toBe(`http://127.0.0.1:${started.port}`);
await expect(fetch(started.url!)).resolves.toMatchObject({ ok: true });
const persisted = await db
.select()
.from(workspaceRuntimeServices)
.where(eq(workspaceRuntimeServices.executionWorkspaceId, workspace.id))
.then((rows) => rows[0] ?? null);
expect(persisted).toMatchObject({ port: started.port, url: started.url, status: "running" });
} finally {
await stopRuntimeServicesForExecutionWorkspace({
db,
executionWorkspaceId: workspace.id,
workspaceCwd: workspace.cwd,
}).catch(() => undefined);
const blockerPid = Number.parseInt(await fs.readFile(blockerPidPath, "utf8").catch(() => ""), 10);
if (Number.isInteger(blockerPid) && blockerPid > 0) {
try {
process.kill(blockerPid, "SIGTERM");
} catch {
// The bind-race process already exited.
}
}
await cleanupRuntimeHome();
await fixture.cleanup();
}
}, 20_000);
it("returns a bounded structured conflict and exposes only same-company workspace references", async () => {
const fixture = await createRuntimeFixture({
workspaceModes: ["isolated_workspace", "isolated_workspace"],
});
const otherCompanyFixture = await createRuntimeFixture();
const cleanupRuntimeHome = await createRuntimeHome();
const reservation = await reserveContiguousPorts(WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS);
const [baseReservation, ...otherReservations] = reservation.servers;
await closeNetServer(baseReservation!);
const firstWorkspace = fixture.workspaces[0]!;
const secondWorkspace = fixture.workspaces[1]!;
const otherCompanyWorkspace = otherCompanyFixture.workspaces[0]!;
const config = fixedPortRuntimeConfig(reservation.basePort);
try {
const first = (await startRuntimeServicesForWorkspaceControl({
db,
actor: fixture.actor,
issue: null,
workspace: fixture.realizedWorkspace(firstWorkspace),
executionWorkspaceId: firstWorkspace.id,
config,
adapterEnv: {},
}))[0]!;
expect(first.port).toBe(reservation.basePort);
let sameCompanyError: unknown;
try {
await startRuntimeServicesForWorkspaceControl({
db,
actor: fixture.actor,
issue: null,
workspace: fixture.realizedWorkspace(secondWorkspace),
executionWorkspaceId: secondWorkspace.id,
config,
adapterEnv: {},
});
} catch (error) {
sameCompanyError = error;
}
expect(sameCompanyError).toMatchObject({
status: 409,
details: {
code: "workspace_runtime_port_allocation_exhausted",
port: reservation.basePort,
attemptedPortCount: WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS,
conflictingExecutionWorkspaceId: firstWorkspace.id,
remediation: expect.any(String),
},
});
expect(JSON.stringify(sameCompanyError)).not.toMatch(/(?:pid|cwd)/i);
let otherCompanyError: unknown;
try {
await startRuntimeServicesForWorkspaceControl({
db,
actor: otherCompanyFixture.actor,
issue: null,
workspace: otherCompanyFixture.realizedWorkspace(otherCompanyWorkspace),
executionWorkspaceId: otherCompanyWorkspace.id,
config,
adapterEnv: {},
});
} catch (error) {
otherCompanyError = error;
}
expect(otherCompanyError).toMatchObject({
status: 409,
details: {
code: "workspace_runtime_port_allocation_exhausted",
port: reservation.basePort,
attemptedPortCount: WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS,
},
});
expect(otherCompanyError).not.toMatchObject({
details: {
conflictingExecutionWorkspaceId: firstWorkspace.id,
},
});
expect(JSON.stringify(otherCompanyError)).not.toContain(firstWorkspace.id);
} finally {
await stopRuntimeServicesForExecutionWorkspace({
db,
executionWorkspaceId: firstWorkspace.id,
workspaceCwd: firstWorkspace.cwd,
}).catch(() => undefined);
await Promise.all(otherReservations.map((server) => closeNetServer(server).catch(() => undefined)));
await cleanupRuntimeHome();
await fixture.cleanup();
await otherCompanyFixture.cleanup();
}
}, 30_000);
it("keeps fixed ports strict for shared persisted workspaces and identity-less starts", async () => {
const fixture = await createRuntimeFixture({ workspaceModes: ["shared_workspace"] });
const cleanupRuntimeHome = await createRuntimeHome();
const workspace = fixture.workspaces[0]!;
const occupiedPort = await findFreePort();
const occupant = await listenOnPort(occupiedPort);
const config = fixedPortRuntimeConfig(occupiedPort);
try {
let sharedWorkspaceError: unknown;
try {
await startRuntimeServicesForWorkspaceControl({
db,
actor: fixture.actor,
issue: null,
workspace: fixture.realizedWorkspace(workspace),
executionWorkspaceId: workspace.id,
config,
adapterEnv: {},
});
} catch (error) {
sharedWorkspaceError = error;
}
expect(sharedWorkspaceError).toBeInstanceOf(Error);
expect(sharedWorkspaceError).not.toMatchObject({ status: 409 });
const identitylessCwd = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-identityless-"));
try {
await expect(startRuntimeServicesForWorkspaceControl({
actor: { id: null, name: "Board", companyId: fixture.companyId },
issue: null,
workspace: buildWorkspace(identitylessCwd),
config,
adapterEnv: {},
})).rejects.toBeInstanceOf(Error);
} finally {
await fs.rm(identitylessCwd, { recursive: true, force: true });
}
const rows = await db
.select()
.from(workspaceRuntimeServices)
.where(eq(workspaceRuntimeServices.executionWorkspaceId, workspace.id));
expect(rows.every((row) => row.port === occupiedPort)).toBe(true);
} finally {
await closeNetServer(occupant);
await stopRuntimeServicesForExecutionWorkspace({
db,
executionWorkspaceId: workspace.id,
workspaceCwd: workspace.cwd,
}).catch(() => undefined);
await cleanupRuntimeHome();
await fixture.cleanup();
}
}, 20_000);
});
describeEmbeddedPostgres("workspace runtime startup reconciliation", () => {
+365 -49
View File
@@ -20,8 +20,9 @@ import {
type WorkspaceRuntimeDesiredState,
type WorkspaceRuntimeServiceStateMap,
} from "@paperclipai/shared";
import { and, desc, eq, inArray, isNull, ne } from "drizzle-orm";
import { and, desc, eq, gte, inArray, isNull, lte, ne, or } from "drizzle-orm";
import { asNumber, asString, parseObject, renderTemplate } from "../adapters/utils.js";
import { conflict } from "../errors.js";
import { resolveHomeAwarePath } from "../home-paths.js";
import {
createLocalServiceKey,
@@ -183,6 +184,18 @@ const runtimeServicesByReuseKey = new Map<string, string>();
const runtimeServiceLeasesByRun = new Map<string, string[]>();
const runtimeProvisionByWorkspace = new Map<string, Promise<void>>();
const DEFAULT_EXECUTE_PROCESS_OUTPUT_BYTES = 256 * 1024;
export const WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS = 32;
const ACTIVE_RUNTIME_PORT_RESERVATION_STATUSES = ["provisioning", "starting", "running"] as const;
class RuntimeServicePortBindCollision extends Error {
readonly port: number;
constructor(port: number) {
super(`Runtime service could not bind allocated port ${port}`);
this.name = "RuntimeServicePortBindCollision";
this.port = port;
}
}
type ProcessOutputCapture = {
text: string;
@@ -2396,21 +2409,6 @@ export function formatManagedGitWorktreeBranchInspection(input: ManagedGitWorktr
};
}
function terminateChildProcess(child: ChildProcess) {
if (!child.pid) return;
if (process.platform !== "win32") {
try {
process.kill(-child.pid, "SIGTERM");
return;
} catch {
// Fall through to the direct child kill.
}
}
if (!child.killed) {
child.kill("SIGTERM");
}
}
function buildWorkspaceCommandEnv(input: {
base: ExecutionWorkspaceInput;
repoRoot: string;
@@ -3560,6 +3558,160 @@ async function allocatePort(): Promise<number> {
});
}
async function canBindRuntimePort(port: number): Promise<boolean> {
return await new Promise<boolean>((resolve, reject) => {
const server = net.createServer();
server.once("error", (error: NodeJS.ErrnoException) => {
if (error.code === "EADDRINUSE" || error.code === "EACCES") {
resolve(false);
return;
}
reject(error);
});
server.listen(port, "127.0.0.1", () => {
server.close((error) => {
if (error) reject(error);
else resolve(true);
});
});
});
}
async function readReservedRuntimePorts(input: {
db?: Db;
ports: number[];
executionWorkspaceId: string | null;
}) {
if (!input.db || input.ports.length === 0) return new Set<number>();
const lowerBound = Math.min(...input.ports);
const upperBound = Math.max(...input.ports);
const otherWorkspaceCondition = input.executionWorkspaceId
? or(
isNull(workspaceRuntimeServices.executionWorkspaceId),
ne(workspaceRuntimeServices.executionWorkspaceId, input.executionWorkspaceId),
)
: undefined;
const rows = await input.db
.select({ port: workspaceRuntimeServices.port })
.from(workspaceRuntimeServices)
.where(
and(
inArray(workspaceRuntimeServices.status, [...ACTIVE_RUNTIME_PORT_RESERVATION_STATUSES]),
gte(workspaceRuntimeServices.port, lowerBound),
lte(workspaceRuntimeServices.port, upperBound),
otherWorkspaceCondition,
),
);
return new Set(rows.flatMap((row) => row.port === null ? [] : [row.port]));
}
async function buildRuntimePortAllocationConflict(input: {
db?: Db;
companyId: string;
executionWorkspaceId: string | null;
preferredPort: number;
attemptedPorts: number[];
}) {
const conflictRows = input.db && input.attemptedPorts.length > 0
? await input.db
.select({
port: workspaceRuntimeServices.port,
executionWorkspaceId: workspaceRuntimeServices.executionWorkspaceId,
projectWorkspaceId: workspaceRuntimeServices.projectWorkspaceId,
})
.from(workspaceRuntimeServices)
.where(
and(
eq(workspaceRuntimeServices.companyId, input.companyId),
inArray(workspaceRuntimeServices.status, [...ACTIVE_RUNTIME_PORT_RESERVATION_STATUSES]),
inArray(workspaceRuntimeServices.port, input.attemptedPorts),
input.executionWorkspaceId
? or(
isNull(workspaceRuntimeServices.executionWorkspaceId),
ne(workspaceRuntimeServices.executionWorkspaceId, input.executionWorkspaceId),
)
: undefined,
),
)
: [];
const attemptedRank = new Map(input.attemptedPorts.map((port, index) => [port, index]));
const authorizedConflict = conflictRows
.sort((left, right) =>
(attemptedRank.get(left.port ?? -1) ?? Number.MAX_SAFE_INTEGER)
- (attemptedRank.get(right.port ?? -1) ?? Number.MAX_SAFE_INTEGER))
.find((row) => row.executionWorkspaceId || row.projectWorkspaceId) ?? null;
const remediation =
"Stop the conflicting managed service or configure a different preferred port, then retry the start.";
return conflict(
`No safe runtime service port is available in the bounded allocation range starting at ${input.preferredPort}.`,
{
code: "workspace_runtime_port_allocation_exhausted",
port: input.preferredPort,
attemptedPortCount: input.attemptedPorts.length,
...(authorizedConflict?.executionWorkspaceId
? { conflictingExecutionWorkspaceId: authorizedConflict.executionWorkspaceId }
: {}),
...(authorizedConflict?.projectWorkspaceId
? { conflictingProjectWorkspaceId: authorizedConflict.projectWorkspaceId }
: {}),
remediation,
},
);
}
async function allocateIsolatedWorkspacePort(input: {
db?: Db;
companyId: string;
executionWorkspaceId: string;
preferredPort: number;
stoppedPort: number | null;
excludedPorts: ReadonlySet<number>;
}) {
const lastPort = Math.min(
65_535,
input.preferredPort + WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS - 1,
);
const candidates: number[] = [];
const addCandidate = (port: number | null) => {
if (
port === null
|| port < input.preferredPort
|| port > lastPort
|| input.excludedPorts.has(port)
|| candidates.includes(port)
) return;
candidates.push(port);
};
addCandidate(input.stoppedPort);
for (let port = input.preferredPort; port <= lastPort; port += 1) addCandidate(port);
const reservedPorts = await readReservedRuntimePorts({
db: input.db,
ports: candidates,
executionWorkspaceId: input.executionWorkspaceId,
});
for (const port of candidates) {
if (reservedPorts.has(port)) continue;
if (await canBindRuntimePort(port)) return port;
}
const attemptedPorts = [
...input.excludedPorts,
...candidates,
].filter((port, index, ports) =>
port >= input.preferredPort
&& port <= lastPort
&& ports.indexOf(port) === index);
throw await buildRuntimePortAllocationConflict({
db: input.db,
companyId: input.companyId,
executionWorkspaceId: input.executionWorkspaceId,
preferredPort: input.preferredPort,
attemptedPorts,
});
}
function buildTemplateData(input: {
workspace: RealizedExecutionWorkspace;
agent: ExecutionWorkspaceAgentRef;
@@ -3824,6 +3976,48 @@ async function waitForReadiness(input: {
throw new Error(`Readiness check failed for ${readinessUrl}: ${lastError}`);
}
async function waitForAllocatedPortBind(input: {
service: Record<string, unknown>;
port: number;
child: ChildProcess;
serviceCwd: string;
}) {
const deadline = Date.now() + resolveWorkspaceRuntimeReadinessTimeoutSec(input.service) * 1000;
while (Date.now() < deadline) {
if (input.child.exitCode !== null || input.child.signalCode !== null) {
throw new Error("service process exited before binding its allocated port");
}
const ownerPid = await readLocalServicePortOwner(input.port);
if (ownerPid) {
const ownerCwd = await readLocalServiceProcessCwd(ownerPid);
if (ownerCwd && !(await isLocalServiceProcessInWorkspace(ownerCwd, input.serviceCwd))) {
throw new RuntimeServicePortBindCollision(input.port);
}
// Give the launched process enough time to surface a losing EADDRINUSE race before
// accepting the listener. The readiness promise still verifies HTTP services separately.
await delay(250);
if (input.child.exitCode !== null || input.child.signalCode !== null) {
throw new Error("service process exited after losing its allocated port");
}
return;
}
// `lsof` is not guaranteed to exist on every supported host. A failed bind probe still
// confirms that a listener appeared; the stability delay above/below lets an unsuccessful
// child report EADDRINUSE before the allocation is accepted.
if (!(await canBindRuntimePort(input.port))) {
await delay(250);
if (input.child.exitCode !== null || input.child.signalCode !== null) {
throw new Error("service process exited after losing its allocated port");
}
return;
}
await delay(50);
}
throw new Error(`Runtime service did not bind allocated port ${input.port} before timeout`);
}
function isPaperclipDevRuntimeService(input: { serviceName?: string | null; command?: string | null }) {
const serviceName = (input.serviceName ?? "").trim().toLowerCase();
const command = (input.command ?? "").trim().toLowerCase();
@@ -4088,6 +4282,8 @@ type StartLocalRuntimeServiceInput = {
provisionCoordinator?: RuntimeProvisionCoordinator;
preparedProvisioningRecord?: RuntimeServiceRecord | null;
runtimeServiceId?: string;
allowFixedPortFallback?: boolean;
excludedPorts?: ReadonlySet<number>;
reuseKey: string | null;
scopeType: "project_workspace" | "execution_workspace" | "run" | "agent";
scopeId: string | null;
@@ -4266,6 +4462,13 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
const serviceIdentityFingerprint = input.reuseKey ?? envFingerprint;
const explicitPort = identity.explicitPort;
const identityPort = identity.identityPort;
const portType = asString(portConfig.type, "");
const canAllocateFixedPort = Boolean(
input.allowFixedPortFallback
&& input.db
&& input.executionWorkspaceId
&& explicitPort > 0,
);
const stoppedReuseCandidate = await findStoppedRuntimeServiceReuseCandidate({
db: input.db,
companyId: input.agent.companyId,
@@ -4277,16 +4480,37 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
scopeId: input.scopeId,
});
let reusableStoppedPort: number | null = null;
if (asString(portConfig.type, "") === "auto" && stoppedReuseCandidate?.port) {
const ownerPid = await readLocalServicePortOwner(stoppedReuseCandidate.port);
reusableStoppedPort = ownerPid ? null : stoppedReuseCandidate.port;
if (portType === "auto" && stoppedReuseCandidate?.port && !input.excludedPorts?.has(stoppedReuseCandidate.port)) {
reusableStoppedPort = await canBindRuntimePort(stoppedReuseCandidate.port)
? stoppedReuseCandidate.port
: null;
}
let port: number | null = null;
if (portType === "auto") {
port = reusableStoppedPort;
for (let attempt = 0; port === null && attempt < WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS; attempt += 1) {
const candidate = await allocatePort();
if (!input.excludedPorts?.has(candidate)) port = candidate;
}
if (port === null) {
throw conflict("No safe automatically allocated runtime service port is available.", {
code: "workspace_runtime_port_allocation_exhausted",
attemptedPortCount: WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS,
remediation: "Retry the start or configure a different runtime service port.",
});
}
} else if (canAllocateFixedPort) {
port = await allocateIsolatedWorkspacePort({
db: input.db,
companyId: input.agent.companyId,
executionWorkspaceId: input.executionWorkspaceId!,
preferredPort: explicitPort,
stoppedPort: stoppedReuseCandidate?.port ?? null,
excludedPorts: input.excludedPorts ?? new Set<number>(),
});
} else {
port = explicitPort > 0 ? explicitPort : null;
}
const port =
asString(portConfig.type, "") === "auto"
? (reusableStoppedPort ?? await allocatePort())
: explicitPort > 0
? explicitPort
: null;
const templateData = buildTemplateData({
workspace: input.workspace,
agent: input.agent,
@@ -4390,9 +4614,12 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
};
}
}
if (identityPort) {
const ownerPid = await readLocalServicePortOwner(identityPort);
if (port) {
const ownerPid = await readLocalServicePortOwner(port);
if (ownerPid) {
if (canAllocateFixedPort || portType === "auto") {
throw new RuntimeServicePortBindCollision(port);
}
const ownerCwd = await readLocalServiceProcessCwd(ownerPid);
const ownerIsInWorkspace = ownerCwd
? await isLocalServiceProcessInWorkspace(ownerCwd, serviceCwd)
@@ -4400,11 +4627,11 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
const ownerDescription = ownerCwd ? `pid ${ownerPid} (cwd: ${ownerCwd})` : `pid ${ownerPid} (cwd unavailable)`;
if (ownerIsInWorkspace === false) {
throw new Error(
`Runtime service "${serviceName}" could not start because port ${identityPort} has a cross-workspace port conflict with ${ownerDescription}; requested workspace: ${serviceCwd}. Stop the other service or configure a different port.`,
`Runtime service "${serviceName}" could not start because port ${port} has a cross-workspace port conflict with ${ownerDescription}; requested workspace: ${serviceCwd}. Stop the other service or configure a different port.`,
);
}
throw new Error(
`Runtime service "${serviceName}" could not start because port ${identityPort} is already in use by ${ownerDescription}`,
`Runtime service "${serviceName}" could not start because port ${port} is already in use by ${ownerDescription}`,
);
}
}
@@ -4425,6 +4652,13 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
reject(err);
});
});
const earlyExitPromise = new Promise<never>((_, reject) => {
child.once("exit", (code, signal) => {
reject(new Error(
`service process exited before readiness (code ${code ?? "unknown"}, signal ${signal ?? "none"})`,
));
});
});
let stderrExcerpt = "";
let stdoutExcerpt = "";
child.stdout?.on("data", async (chunk) => {
@@ -4506,8 +4740,14 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
}
const readinessPromise = Promise.race([
waitForReadiness({ service: input.service, serviceName, command, url, readinessUrl }),
Promise.all([
waitForReadiness({ service: input.service, serviceName, command, url, readinessUrl }),
canAllocateFixedPort && port
? waitForAllocatedPortBind({ service: input.service, port, child, serviceCwd })
: Promise.resolve(),
]).then(() => undefined),
spawnErrorPromise,
earlyExitPromise,
]).then(async () => {
record.status = "running";
record.healthStatus = "healthy";
@@ -4518,14 +4758,26 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
lastSeenAt: record.lastUsedAt,
});
}).catch(async (err) => {
terminateChildProcess(child);
const failureMessage = err instanceof Error ? err.message : String(err);
const bindCollision = err instanceof RuntimeServicePortBindCollision || Boolean(
port
&& (input.allowFixedPortFallback || portType === "auto")
&& /(?:EADDRINUSE|address already in use)/i.test(`${failureMessage}\n${stderrExcerpt}`),
);
if (child.pid) {
await terminateLocalService({
pid: child.pid,
processGroupId: child.pid,
});
}
record.status = "stopped";
record.healthStatus = "unhealthy";
record.lastUsedAt = new Date().toISOString();
record.stoppedAt = new Date().toISOString();
await removeLocalServiceRegistryRecord(record.serviceKey).catch(() => undefined);
if (bindCollision && port) throw new RuntimeServicePortBindCollision(port);
throw new Error(
`Failed to start runtime service "${serviceName}": ${err instanceof Error ? err.message : String(err)}${stderrExcerpt ? ` | stderr: ${stderrExcerpt.trim()}` : ""}`,
`Failed to start runtime service "${serviceName}": ${failureMessage}${stderrExcerpt ? ` | stderr: ${stderrExcerpt.trim()}` : ""}`,
);
});
@@ -4595,24 +4847,57 @@ async function startLocalRuntimeService(
? await prepareRuntimeProvisioning(input)
: input.preparedProvisioningRecord;
let started: LocalRuntimeServiceStart | null = null;
const excludedPorts = new Set(input.excludedPorts ?? []);
const portConfig = parseObject(input.service.port);
const portType = asString(portConfig.type, "");
const explicitPort = asNumber(portConfig.value, asNumber(input.service.port, 0));
const fixedPortFallbackEnabled = Boolean(
input.allowFixedPortFallback && portType !== "auto" && explicitPort > 0,
);
const retryBindCollisions = fixedPortFallbackEnabled || portType === "auto";
const deferReadiness = Boolean(options?.deferReadiness && !fixedPortFallbackEnabled);
try {
started = await spawnLocalRuntimeService({
...input,
runtimeServiceId: provisioningRecord?.id ?? input.runtimeServiceId,
for (let attempt = 0; attempt < WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS; attempt += 1) {
try {
started = await spawnLocalRuntimeService({
...input,
excludedPorts,
runtimeServiceId: provisioningRecord?.id ?? input.runtimeServiceId,
});
if (runtimeProvisionCommand) {
await persistRuntimeServiceRecord(input.db, started.record);
}
if (provisioningRecord && started.record.id !== provisioningRecord.id && input.db) {
await input.db
.delete(workspaceRuntimeServices)
.where(eq(workspaceRuntimeServices.id, provisioningRecord.id));
}
if (!deferReadiness) {
await started.readiness;
}
return started;
} catch (error) {
if (!(error instanceof RuntimeServicePortBindCollision) || !retryBindCollisions) throw error;
excludedPorts.add(error.port);
started = null;
}
}
if (fixedPortFallbackEnabled && input.executionWorkspaceId) {
throw await buildRuntimePortAllocationConflict({
db: input.db,
companyId: input.agent.companyId,
executionWorkspaceId: input.executionWorkspaceId,
preferredPort: explicitPort,
attemptedPorts: [...excludedPorts],
});
}
throw conflict("No safe automatically allocated runtime service port is available.", {
code: "workspace_runtime_port_allocation_exhausted",
attemptedPortCount: excludedPorts.size,
remediation: "Retry the start or configure a different runtime service port.",
});
if (runtimeProvisionCommand) {
await persistRuntimeServiceRecord(input.db, started.record);
}
if (provisioningRecord && started.record.id !== provisioningRecord.id && input.db) {
await input.db
.delete(workspaceRuntimeServices)
.where(eq(workspaceRuntimeServices.id, provisioningRecord.id));
}
if (!options?.deferReadiness) {
await started.readiness;
}
return started;
} catch (error) {
if (!started && provisioningRecord && provisioningRecord.status === "starting") {
const nowIso = new Date().toISOString();
@@ -4802,6 +5087,26 @@ function selectRuntimeServiceEntries(input: {
});
}
async function isPersistedIsolatedExecutionWorkspace(input: {
db?: Db;
companyId: string;
executionWorkspaceId?: string | null;
}) {
if (!input.db || !input.executionWorkspaceId) return false;
const row = await input.db
.select({ mode: executionWorkspaces.mode })
.from(executionWorkspaces)
.where(
and(
eq(executionWorkspaces.id, input.executionWorkspaceId),
eq(executionWorkspaces.companyId, input.companyId),
),
)
.limit(1)
.then((rows) => rows[0] ?? null);
return row?.mode === "isolated_workspace";
}
export async function ensureRuntimeServicesForRun(input: {
db?: Db;
runId: string;
@@ -4824,6 +5129,11 @@ export async function ensureRuntimeServicesForRun(input: {
const refs: RuntimeServiceRef[] = [];
const runtimeProvisionCommand = resolveRuntimeProvisionCommand(input);
const provisionCoordinator = createRuntimeProvisionCoordinator();
const allowFixedPortFallback = await isPersistedIsolatedExecutionWorkspace({
db: input.db,
companyId: input.agent.companyId,
executionWorkspaceId: input.executionWorkspaceId,
});
runtimeServiceLeasesByRun.set(input.runId, acquiredServiceIds);
try {
@@ -4878,6 +5188,7 @@ export async function ensureRuntimeServicesForRun(input: {
runtimeProvisionCommand,
recorder: input.recorder,
provisionCoordinator,
allowFixedPortFallback,
reuseKey,
scopeType,
scopeId,
@@ -4925,6 +5236,7 @@ async function startRuntimeServicesForWorkspaceControlUnlocked(
registryDb = input.db,
options?: {
deferReadiness?: boolean;
allowFixedPortFallback?: boolean;
runtimeProvisionCommand?: string;
provisionCoordinator?: RuntimeProvisionCoordinator;
preparedProvisioning?: {
@@ -4998,6 +5310,7 @@ async function startRuntimeServicesForWorkspaceControlUnlocked(
options?.preparedProvisioning?.service === service
? options.preparedProvisioning.record
: undefined,
allowFixedPortFallback: options?.allowFixedPortFallback,
reuseKey,
scopeType,
scopeId,
@@ -5012,7 +5325,7 @@ async function startRuntimeServicesForWorkspaceControlUnlocked(
await persistRuntimeServiceRecord(persistenceDb, started.record);
refs.push(toRuntimeServiceRef(started.record));
if (options?.deferReadiness && !started.record.reused) {
if (options?.deferReadiness && started.record.status === "starting" && !started.record.reused) {
// Attach a rejection handler immediately; the caller awaits the same promise after
// the DB transaction commits, but transaction failures may skip that wait path.
started.readiness.catch(() => undefined);
@@ -5059,6 +5372,7 @@ export async function startRuntimeServicesForWorkspaceControl(
service: Record<string, unknown>;
record: RuntimeServiceRecord;
} | null = null;
let allowFixedPortFallback = false;
try {
if (runtimeProvisionCommand) {
for (const service of rawServices) {
@@ -5112,7 +5426,7 @@ export async function startRuntimeServicesForWorkspaceControl(
if (input.executionWorkspaceId) {
const [lockedExecutionWorkspace] = await tx
.select({ id: executionWorkspaces.id })
.select({ id: executionWorkspaces.id, mode: executionWorkspaces.mode })
.from(executionWorkspaces)
.where(
and(
@@ -5122,6 +5436,7 @@ export async function startRuntimeServicesForWorkspaceControl(
)
.for("update");
if (!lockedExecutionWorkspace) throw new Error("Execution workspace not found before starting runtime services");
allowFixedPortFallback = lockedExecutionWorkspace.mode === "isolated_workspace";
}
if (input.workspace.workspaceId) {
@@ -5149,6 +5464,7 @@ export async function startRuntimeServicesForWorkspaceControl(
input.db,
{
deferReadiness: true,
allowFixedPortFallback,
runtimeProvisionCommand,
provisionCoordinator,
preparedProvisioning,