From 54b4e0c2fb6a76938b7d5fa9c396f93d6a67aab3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 12 Aug 2026 14:20:45 +0000 Subject: [PATCH] fix(runtime): harden sibling port allocation Co-Authored-By: Paperclip --- .../src/__tests__/workspace-runtime.test.ts | 509 ++++++++++++++++++ server/src/services/workspace-runtime.ts | 414 ++++++++++++-- 2 files changed, 874 insertions(+), 49 deletions(-) diff --git a/server/src/__tests__/workspace-runtime.test.ts b/server/src/__tests__/workspace-runtime.test.ts index cf052cb4372..ec050549ac0 100644 --- a/server/src/__tests__/workspace-runtime.test.ts +++ b/server/src/__tests__/workspace-runtime.test.ts @@ -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((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()); + }); +} + +async function listenOnPort(port: number) { + const server = net.createServer(); + await new Promise((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", () => { diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index cb96c5f9d4d..f98fd019d1b 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -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(); const runtimeServiceLeasesByRun = new Map(); const runtimeProvisionByWorkspace = new Map>(); 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 { }); } +async function canBindRuntimePort(port: number): Promise { + return await new Promise((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(); + 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; +}) { + 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; + 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; 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(), + }); + } 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((_, 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; 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,