diff --git a/server/src/__tests__/execution-workspaces-service.test.ts b/server/src/__tests__/execution-workspaces-service.test.ts index e7fc8db03f..858ed344f2 100644 --- a/server/src/__tests__/execution-workspaces-service.test.ts +++ b/server/src/__tests__/execution-workspaces-service.test.ts @@ -1901,7 +1901,7 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { ), ]); - const reconcilePromise = svc.reconcileExecutionWorkspaceBranch(executionWorkspaceId, { + const reconcileErrorPromise = svc.reconcileExecutionWorkspaceBranch(executionWorkspaceId, { mode: "override", reason: "operator override still requires stopped services", actor: { @@ -1910,10 +1910,15 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { agentId: null, runId: null, }, - }); + }).then( + () => { + throw new Error("Branch reconciliation unexpectedly succeeded while a runtime service was starting"); + }, + (error) => error, + ); startedServices = await startPromise; - await expect(reconcilePromise).rejects.toMatchObject({ + await expect(reconcileErrorPromise).resolves.toMatchObject({ status: 422, message: "Execution workspace branch reconciliation requires all runtime services to be stopped", details: { @@ -1926,7 +1931,7 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { expect.objectContaining({ id: startedServices[0]?.id, serviceName: "web", - status: "running", + status: "starting", }), ], }, diff --git a/server/src/__tests__/workspace-runtime.test.ts b/server/src/__tests__/workspace-runtime.test.ts index 0bfc8dd562..aa20eadf3d 100644 --- a/server/src/__tests__/workspace-runtime.test.ts +++ b/server/src/__tests__/workspace-runtime.test.ts @@ -42,7 +42,13 @@ import { stopRuntimeServicesForExecutionWorkspace, type RealizedExecutionWorkspace, } from "../services/workspace-runtime.ts"; -import { readLocalServicePortOwner, writeLocalServiceRegistryRecord } from "../services/local-service-supervisor.ts"; +import { + findAdoptableLocalService, + isLocalServiceRegistryCwdCompatible, + isLocalServiceProcessInWorkspace, + readLocalServicePortOwner, + writeLocalServiceRegistryRecord, +} from "../services/local-service-supervisor.ts"; import { resolvePaperclipConfigPath } from "../paths.ts"; import type { WorkspaceOperation } from "@paperclipai/shared"; import type { WorkspaceOperationRecorder } from "../services/workspace-operations.ts"; @@ -4069,6 +4075,12 @@ describe("resolveShell (shell fallback)", () => { }); describe("readLocalServicePortOwner", () => { + const originalPlatform = process.platform; + + afterEach(() => { + Object.defineProperty(process, "platform", { value: originalPlatform }); + }); + it("detects the owner of a listening TCP port", async () => { try { await execFileAsync("lsof", ["-v"]); @@ -4094,6 +4106,167 @@ describe("readLocalServicePortOwner", () => { }); } }); + + it("accepts service cwd nested within the requested workspace", async () => { + const workspace = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-workspace-")); + const serviceCwd = path.join(workspace, "server"); + await fs.mkdir(serviceCwd); + + await expect(isLocalServiceProcessInWorkspace(serviceCwd, workspace)).resolves.toBe(true); + }); + + it("keeps a live registry record adoptable when cwd inspection is unsupported", async () => { + try { + await execFileAsync("lsof", ["-v"]); + } catch { + return; + } + + const server = net.createServer(); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + const port = typeof address === "object" && address ? address.port : null; + const serviceKey = `unsupported-cwd-${randomUUID()}`; + const paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-home-")); + process.env.PAPERCLIP_HOME = paperclipHome; + process.env.PAPERCLIP_INSTANCE_ID = `unsupported-cwd-${randomUUID()}`; + expect(port).toBeTypeOf("number"); + + try { + await writeLocalServiceRegistryRecord({ + version: 1, + serviceKey, + profileKind: "workspace-runtime", + serviceName: "node", + command: "node", + cwd: process.cwd(), + envFingerprint: "", + port, + url: null, + pid: process.pid, + processGroupId: null, + provider: "local_process", + runtimeServiceId: null, + reuseKey: null, + startedAt: new Date().toISOString(), + lastSeenAt: new Date().toISOString(), + metadata: null, + }); + Object.defineProperty(process, "platform", { value: "darwin" }); + + await expect(findAdoptableLocalService({ + serviceKey, + cwd: process.cwd(), + port, + })).resolves.toMatchObject({ pid: expect.any(Number), port }); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()); + }); + await fs.rm(paperclipHome, { recursive: true, force: true }); + } + }); + + it("trusts unavailable cwd for registry records only off Linux", async () => { + Object.defineProperty(process, "platform", { value: "darwin" }); + await expect(isLocalServiceRegistryCwdCompatible(null, process.cwd())).resolves.toBe(true); + + Object.defineProperty(process, "platform", { value: "linux" }); + await expect(isLocalServiceRegistryCwdCompatible(null, process.cwd())).resolves.toBe(false); + }); + + it("refuses to adopt a listener whose real cwd belongs to another workspace", async () => { + if (process.platform !== "linux") return; + try { + await execFileAsync("lsof", ["-v"]); + } catch { + return; + } + + const targetWorkspace = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-target-")); + const ownerWorkspace = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-owner-")); + const paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-home-")); + process.env.PAPERCLIP_HOME = paperclipHome; + process.env.PAPERCLIP_INSTANCE_ID = `cross-workspace-${randomUUID()}`; + const serviceKey = `cross-workspace-${randomUUID()}`; + const child = spawn( + process.execPath, + [ + "-e", + "const server=require('node:http').createServer((req,res)=>res.end('ok')); server.listen(0, '127.0.0.1', () => console.log(server.address().port));", + ], + { cwd: ownerWorkspace, stdio: ["ignore", "pipe", "inherit"] }, + ); + const port = await new Promise((resolve, reject) => { + let output = ""; + child.stdout?.on("data", (chunk) => { + output += String(chunk); + const value = Number.parseInt(output.trim(), 10); + if (Number.isInteger(value) && value > 0) resolve(value); + }); + child.once("error", reject); + child.once("exit", (code) => reject(new Error(`Port owner exited before listening: ${code ?? "unknown"}`))); + }); + + try { + await expect(findAdoptableLocalService({ + serviceKey, + serviceName: "node", + command: "node", + cwd: targetWorkspace, + port, + })).resolves.toBeNull(); + + await writeLocalServiceRegistryRecord({ + version: 1, + serviceKey, + profileKind: "workspace-runtime", + serviceName: "node", + command: "node", + cwd: targetWorkspace, + envFingerprint: "", + port, + url: null, + pid: child.pid!, + processGroupId: null, + provider: "local_process", + runtimeServiceId: null, + reuseKey: null, + startedAt: new Date().toISOString(), + lastSeenAt: new Date().toISOString(), + metadata: null, + }); + await expect(findAdoptableLocalService({ + serviceKey, + serviceName: "node", + command: "node", + cwd: targetWorkspace, + port, + })).resolves.toBeNull(); + + await expect(startRuntimeServicesForWorkspaceControl({ + actor: { id: "agent-1", name: "Codex Coder", companyId: "company-1" }, + issue: null, + workspace: buildWorkspace(targetWorkspace), + config: { + workspaceRuntime: { + services: [{ + name: "web", + command: "node", + cwd: ".", + port, + lifecycle: "shared", + }], + }, + }, + adapterEnv: {}, + })).rejects.toThrow(new RegExp(`cross-workspace port conflict.*pid ${child.pid}.*${ownerWorkspace.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}`, "i")); + } finally { + child.kill("SIGTERM"); + await new Promise((resolve) => child.once("exit", () => resolve())); + await fs.rm(paperclipHome, { recursive: true, force: true }); + } + }); }); describeEmbeddedPostgres("workspace dirty quarantine branch repair", () => { @@ -4738,6 +4911,214 @@ describeEmbeddedPostgres("workspace dirty quarantine branch repair", () => { }, 20_000); }); +describeEmbeddedPostgres("workspace runtime service control persistence", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-workspace-runtime-control-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + afterEach(async () => { + await resetRuntimeServicesForTests(); + await db.delete(workspaceRuntimeServices); + await db.delete(executionWorkspaces); + await db.delete(projectWorkspaces); + await db.delete(issues); + await db.delete(projects); + await db.delete(agents); + await db.delete(companies); + }); + + it("commits a starting service row before waiting for slow readiness", async () => { + const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-slow-control-")); + const paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-control-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-control-${randomUUID()}`; + + const companyId = randomUUID(); + const projectId = randomUUID(); + const projectWorkspaceId = randomUUID(); + const issueId = randomUUID(); + const executionWorkspaceId = randomUUID(); + const markerPath = path.join(workspaceRoot, "runtime-spawned.marker"); + const serverScript = [ + `require("node:fs").writeFileSync(${JSON.stringify(markerPath)}, "spawned");`, + "setTimeout(() => {", + " require(\"node:http\")", + " .createServer((_req, res) => { res.end(\"ok\"); })", + " .listen(Number(process.env.PORT), \"127.0.0.1\");", + "}, 700);", + "setInterval(() => {}, 1000);", + ].join(" "); + const command = `${JSON.stringify(process.execPath)} -e ${JSON.stringify(serverScript)}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Runtime control", + status: "in_progress", + }); + await db.insert(projectWorkspaces).values({ + id: projectWorkspaceId, + companyId, + projectId, + name: "Primary", + sourceType: "local_path", + cwd: workspaceRoot, + isPrimary: true, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + projectId, + title: "Source task", + status: "in_progress", + priority: "high", + }); + await db.insert(executionWorkspaces).values({ + id: executionWorkspaceId, + companyId, + projectId, + projectWorkspaceId, + sourceIssueId: issueId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "Runtime control workspace", + status: "active", + providerType: "git_worktree", + cwd: workspaceRoot, + providerRef: workspaceRoot, + branchName: "feature/runtime-control", + baseRef: "main", + }); + + const waitForMarker = async () => { + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + if (existsSync(markerPath)) return; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + throw new Error("Timed out waiting for runtime service process marker"); + }; + const waitForPersistedStatus = async (status: string) => { + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + const row = await db + .select() + .from(workspaceRuntimeServices) + .where(eq(workspaceRuntimeServices.executionWorkspaceId, executionWorkspaceId)) + .then((rows) => rows[0] ?? null); + if (row?.status === status) return row; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + throw new Error(`Timed out waiting for persisted runtime service status ${status}`); + }; + + const startPromise = startRuntimeServicesForWorkspaceControl({ + db, + invocationId: randomUUID(), + actor: { + id: null, + name: "Board", + companyId, + }, + issue: { + id: issueId, + identifier: null, + title: "Source task", + }, + workspace: { + baseCwd: workspaceRoot, + source: "task_session", + projectId, + workspaceId: projectWorkspaceId, + repoUrl: null, + repoRef: "main", + strategy: "git_worktree", + cwd: workspaceRoot, + branchName: "feature/runtime-control", + worktreePath: workspaceRoot, + warnings: [], + created: false, + }, + executionWorkspaceId, + config: { + workspaceRuntime: { + services: [ + { + name: "web", + command, + lifecycle: "shared", + reuseScope: "execution_workspace", + port: { type: "auto", envKey: "PORT" }, + expose: { urlTemplate: "http://127.0.0.1:{{port}}" }, + readiness: { type: "http", intervalMs: 50, timeoutSec: 10 }, + stopPolicy: { type: "manual" }, + }, + ], + }, + }, + adapterEnv: {}, + }); + startPromise.catch(() => undefined); + + try { + await waitForMarker(); + const startingRow = await waitForPersistedStatus("starting"); + expect(startingRow).toMatchObject({ + companyId, + projectId, + projectWorkspaceId, + executionWorkspaceId, + issueId, + serviceName: "web", + status: "starting", + healthStatus: "unknown", + }); + expect(startingRow.providerRef).toMatch(/^\d+$/); + expect(startingRow.port).toEqual(expect.any(Number)); + + const services = await startPromise; + expect(services).toHaveLength(1); + expect(services[0]).toMatchObject({ + id: startingRow.id, + status: "running", + healthStatus: "healthy", + }); + + const runningRow = await waitForPersistedStatus("running"); + expect(runningRow.id).toBe(startingRow.id); + await expect(fetch(services[0]!.url!)).resolves.toMatchObject({ ok: true }); + } finally { + await startPromise.catch(() => undefined); + await stopRuntimeServicesForExecutionWorkspace({ + db, + executionWorkspaceId, + workspaceCwd: workspaceRoot, + }); + await fs.rm(paperclipHome, { recursive: true, force: true }); + 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; + } + }, 15_000); +}); + describeEmbeddedPostgres("workspace runtime startup reconciliation", () => { let db!: ReturnType; let tempDb: Awaited> | null = null; diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index c1dc1426b4..d47d81801e 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -1667,7 +1667,7 @@ export function executionWorkspaceService(db: Db) { return db.transaction(async (tx) => { const txDb = tx as unknown as Db; // Runtime-service activation takes this same row lock before spawning - // local services and holds it until the running service row is persisted. + // local services and persists a `starting` row before releasing it. const lockedRow = await tx .select() .from(executionWorkspaces) diff --git a/server/src/services/local-service-supervisor.ts b/server/src/services/local-service-supervisor.ts index 86d270e37a..b459ad3cff 100644 --- a/server/src/services/local-service-supervisor.ts +++ b/server/src/services/local-service-supervisor.ts @@ -207,6 +207,10 @@ export async function findLocalServiceRegistryRecordByRuntimeServiceId(input: { await removeLocalServiceRegistryRecord(record.serviceKey); return null; } + if (!(await doesLocalServiceRecordMatchCwd(candidate))) { + await removeLocalServiceRegistryRecord(record.serviceKey); + return null; + } return candidate; } @@ -270,6 +274,10 @@ export async function findAdoptableLocalService(input: { await removeLocalServiceRegistryRecord(input.serviceKey); return null; } + if (!(await doesLocalServiceRecordMatchCwd(record))) { + await removeLocalServiceRegistryRecord(input.serviceKey); + return null; + } if (input.command && record.command !== input.command) return null; if (input.cwd && path.resolve(record.cwd) !== path.resolve(input.cwd)) return null; if (input.envFingerprint && record.envFingerprint !== input.envFingerprint) return null; @@ -302,6 +310,13 @@ async function adoptLocalServiceFromPortOwner(input: { const ownerPid = await readLocalServicePortOwner(input.port); if (!ownerPid) return null; + if (input.cwd) { + const ownerCwd = await readLocalServiceProcessCwd(ownerPid); + if (!ownerCwd || !(await isLocalServiceProcessInWorkspace(ownerCwd, input.cwd))) { + return null; + } + } + const processGroupId = await readProcessGroupId(ownerPid); const pid = processGroupId && isPidAlive(processGroupId) ? processGroupId : ownerPid; const now = new Date().toISOString(); @@ -402,3 +417,38 @@ export async function readLocalServicePortOwner(port: number) { return null; } } + +export async function readLocalServiceProcessCwd(pid: number) { + if (!Number.isInteger(pid) || pid <= 0 || process.platform !== "linux") return null; + try { + return await fs.readlink(`/proc/${pid}/cwd`); + } catch { + return null; + } +} + +export async function isLocalServiceProcessInWorkspace(processCwd: string, workspaceCwd: string) { + try { + const [resolvedProcessCwd, resolvedWorkspaceCwd] = await Promise.all([ + fs.realpath(processCwd), + fs.realpath(workspaceCwd), + ]); + const relativePath = path.relative(resolvedWorkspaceCwd, resolvedProcessCwd); + return relativePath === "" || (!relativePath.startsWith(`..${path.sep}`) && relativePath !== ".."); + } catch { + return false; + } +} + +export async function isLocalServiceRegistryCwdCompatible(processCwd: string | null, workspaceCwd: string) { + if (!processCwd) return process.platform !== "linux"; + return isLocalServiceProcessInWorkspace(processCwd, workspaceCwd); +} + +async function doesLocalServiceRecordMatchCwd(record: LocalServiceRegistryRecord) { + if (!record.port) return true; + const ownerPid = await readLocalServicePortOwner(record.port); + if (!ownerPid) return false; + const ownerCwd = await readLocalServiceProcessCwd(ownerPid); + return isLocalServiceRegistryCwdCompatible(ownerCwd, record.cwd); +} diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 3960116769..7e56da6bd5 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -24,6 +24,8 @@ import { createLocalServiceKey, findLocalServiceRegistryRecordByRuntimeServiceId, findAdoptableLocalService, + isLocalServiceProcessInWorkspace, + readLocalServiceProcessCwd, readLocalServicePortOwner, removeLocalServiceRegistryRecord, terminateLocalService, @@ -127,6 +129,11 @@ interface RuntimeServiceRecord extends RuntimeServiceRef { processGroupId: number | null; } +type LocalRuntimeServiceStart = { + record: RuntimeServiceRecord; + readiness: Promise; +}; + type StoppedRuntimeServiceReuseCandidate = { id: string; port: number | null; @@ -3785,7 +3792,7 @@ export function normalizeAdapterManagedRuntimeServices(input: { }); } -async function startLocalRuntimeService(input: { +type StartLocalRuntimeServiceInput = { db?: Db; runId: string; leaseRunId?: string | null; @@ -3800,7 +3807,9 @@ async function startLocalRuntimeService(input: { reuseKey: string | null; scopeType: "project_workspace" | "execution_workspace" | "run" | "agent"; scopeId: string | null; -}): Promise { +}; + +async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): Promise { const leaseRunId = input.leaseRunId === undefined ? input.runId : input.leaseRunId; const startedByRunId = input.startedByRunId === undefined ? input.runId : input.startedByRunId; const identity = resolveRuntimeServiceReuseIdentity({ @@ -3899,48 +3908,61 @@ async function startLocalRuntimeService(input: { await removeLocalServiceRegistryRecord(adoptedRecord.serviceKey); } else { return { - id: adoptedRecord.runtimeServiceId ?? randomUUID(), - companyId: input.agent.companyId, - projectId: input.workspace.projectId, - projectWorkspaceId: input.workspace.workspaceId, - executionWorkspaceId: input.executionWorkspaceId ?? null, - issueId: input.issue?.id ?? null, - serviceName, - status: "running", - lifecycle, - scopeType: input.scopeType, - scopeId: input.scopeId, - reuseKey: input.reuseKey, - command, - cwd: serviceCwd, - port: adoptedRecord.port ?? port, - url: adoptedRecord.url ?? url, - provider: "local_process", - providerRef: String(adoptedRecord.pid), - ownerAgentId: input.agent.id ?? null, - startedByRunId, - lastUsedAt: new Date().toISOString(), - startedAt: adoptedRecord.startedAt, - stoppedAt: null, - stopPolicy, - healthStatus: "healthy", - reused: true, - db: input.db, - child: null, - leaseRunIds: leaseRunId ? new Set([leaseRunId]) : new Set(), - idleTimer: null, - envFingerprint, - serviceKey, - profileKind: "workspace-runtime", - processGroupId: adoptedRecord.processGroupId ?? null, + record: { + id: adoptedRecord.runtimeServiceId ?? randomUUID(), + companyId: input.agent.companyId, + projectId: input.workspace.projectId, + projectWorkspaceId: input.workspace.workspaceId, + executionWorkspaceId: input.executionWorkspaceId ?? null, + issueId: input.issue?.id ?? null, + serviceName, + status: "running", + lifecycle, + scopeType: input.scopeType, + scopeId: input.scopeId, + reuseKey: input.reuseKey, + command, + cwd: serviceCwd, + port: adoptedRecord.port ?? port, + url: adoptedRecord.url ?? url, + provider: "local_process", + providerRef: String(adoptedRecord.pid), + ownerAgentId: input.agent.id ?? null, + startedByRunId, + lastUsedAt: new Date().toISOString(), + startedAt: adoptedRecord.startedAt, + stoppedAt: null, + stopPolicy, + healthStatus: "healthy", + reused: true, + db: input.db, + child: null, + leaseRunIds: leaseRunId ? new Set([leaseRunId]) : new Set(), + idleTimer: null, + envFingerprint, + serviceKey, + profileKind: "workspace-runtime", + processGroupId: adoptedRecord.processGroupId ?? null, + }, + readiness: Promise.resolve(), }; } } if (identityPort) { - const ownerPid = await readLocalServicePortOwner(identityPort); + const ownerPid = await readLocalServicePortOwner(identityPort); if (ownerPid) { + const ownerCwd = await readLocalServiceProcessCwd(ownerPid); + const ownerIsInWorkspace = ownerCwd + ? await isLocalServiceProcessInWorkspace(ownerCwd, serviceCwd) + : null; + 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.`, + ); + } throw new Error( - `Runtime service "${serviceName}" could not start because port ${identityPort} is already in use by pid ${ownerPid}`, + `Runtime service "${serviceName}" could not start because port ${identityPort} is already in use by ${ownerDescription}`, ); } } @@ -3974,18 +3996,7 @@ async function startLocalRuntimeService(input: { if (input.onLog) await input.onLog("stderr", `[service:${serviceName}] ${text}`); }); - try { - await Promise.race([ - waitForReadiness({ service: input.service, serviceName, command, url }), - spawnErrorPromise, - ]); - } catch (err) { - terminateChildProcess(child); - throw new Error( - `Failed to start runtime service "${serviceName}": ${err instanceof Error ? err.message : String(err)}${stderrExcerpt ? ` | stderr: ${stderrExcerpt.trim()}` : ""}`, - ); - } - + const nowIso = new Date().toISOString(); const record: RuntimeServiceRecord = { id: stoppedReuseCandidate?.id ?? randomUUID(), companyId: input.agent.companyId, @@ -3994,7 +4005,7 @@ async function startLocalRuntimeService(input: { executionWorkspaceId: input.executionWorkspaceId ?? null, issueId: input.issue?.id ?? null, serviceName, - status: "running", + status: "starting", lifecycle, scopeType: input.scopeType, scopeId: input.scopeId, @@ -4007,11 +4018,11 @@ async function startLocalRuntimeService(input: { providerRef: child.pid ? String(child.pid) : null, ownerAgentId: input.agent.id ?? null, startedByRunId, - lastUsedAt: new Date().toISOString(), - startedAt: new Date().toISOString(), + lastUsedAt: nowIso, + startedAt: nowIso, stoppedAt: null, stopPolicy, - healthStatus: "healthy", + healthStatus: "unknown", reused: false, db: input.db, child, @@ -4052,7 +4063,37 @@ async function startLocalRuntimeService(input: { }); } - return record; + const readinessPromise = Promise.race([ + waitForReadiness({ service: input.service, serviceName, command, url }), + spawnErrorPromise, + ]).then(async () => { + record.status = "running"; + record.healthStatus = "healthy"; + record.lastUsedAt = new Date().toISOString(); + record.stoppedAt = null; + await touchLocalServiceRegistryRecord(record.serviceKey, { + runtimeServiceId: record.id, + lastSeenAt: record.lastUsedAt, + }); + }).catch(async (err) => { + terminateChildProcess(child); + record.status = "stopped"; + record.healthStatus = "unhealthy"; + record.lastUsedAt = new Date().toISOString(); + record.stoppedAt = new Date().toISOString(); + await removeLocalServiceRegistryRecord(record.serviceKey).catch(() => undefined); + throw new Error( + `Failed to start runtime service "${serviceName}": ${err instanceof Error ? err.message : String(err)}${stderrExcerpt ? ` | stderr: ${stderrExcerpt.trim()}` : ""}`, + ); + }); + + return { record, readiness: readinessPromise }; +} + +async function startLocalRuntimeService(input: StartLocalRuntimeServiceInput): Promise { + const started = await spawnLocalRuntimeService(input); + await started.readiness; + return started.record; } function scheduleIdleStop(record: RuntimeServiceRecord) { @@ -4332,14 +4373,23 @@ type StartRuntimeServicesForWorkspaceControlInput = { respectDesiredStates?: boolean; }; +type WorkspaceControlStartBatch = { + refs: RuntimeServiceRef[]; + pendingReadiness: LocalRuntimeServiceStart[]; + startedServiceIds: string[]; +}; + async function startRuntimeServicesForWorkspaceControlUnlocked( input: StartRuntimeServicesForWorkspaceControlInput, rawServices: Record[], invocationId: string, persistenceDb = input.db, registryDb = input.db, -): Promise { + options?: { deferReadiness?: boolean }, +): Promise { const refs: RuntimeServiceRef[] = []; + const pendingReadiness: LocalRuntimeServiceStart[] = []; + const startedServiceIds: string[] = []; for (const service of rawServices) { const { scopeType, scopeId } = resolveServiceScopeId({ @@ -4377,9 +4427,7 @@ async function startRuntimeServicesForWorkspaceControlUnlocked( } } - // Manually controlled services are not tied to a heartbeat run lifecycle, so they do not - // retain a run lease and never persist a startedByRunId foreign key. - const record = await startLocalRuntimeService({ + const startInput: StartLocalRuntimeServiceInput = { db: persistenceDb, runId: invocationId, leaseRunId: null, @@ -4394,13 +4442,30 @@ async function startRuntimeServicesForWorkspaceControlUnlocked( reuseKey, scopeType, scopeId, - }); - registerRuntimeService(registryDb, record); - await persistRuntimeServiceRecord(persistenceDb, record); - refs.push(toRuntimeServiceRef(record)); + }; + + // Manually controlled services are not tied to a heartbeat run lifecycle, so they do not + // retain a run lease and never persist a startedByRunId foreign key. + const started = options?.deferReadiness + ? await spawnLocalRuntimeService(startInput) + : { + record: await startLocalRuntimeService(startInput), + readiness: Promise.resolve(), + }; + registerRuntimeService(registryDb, started.record); + await persistRuntimeServiceRecord(persistenceDb, started.record); + refs.push(toRuntimeServiceRef(started.record)); + + if (options?.deferReadiness && !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); + pendingReadiness.push(started); + startedServiceIds.push(started.record.id); + } } - return refs; + return { refs, pendingReadiness, startedServiceIds }; } export async function startRuntimeServicesForWorkspaceControl( @@ -4416,12 +4481,17 @@ export async function startRuntimeServicesForWorkspaceControl( const invocationId = input.invocationId ?? randomUUID(); if (rawServices.length === 0 || !input.db || (!input.executionWorkspaceId && !input.workspace.workspaceId)) { - return startRuntimeServicesForWorkspaceControlUnlocked(input, rawServices, invocationId); + const batch = await startRuntimeServicesForWorkspaceControlUnlocked(input, rawServices, invocationId); + return batch.refs; } - let startedRefs: RuntimeServiceRef[] = []; + let startBatch: WorkspaceControlStartBatch = { + refs: [], + pendingReadiness: [], + startedServiceIds: [], + }; try { - return await input.db.transaction(async (tx) => { + await input.db.transaction(async (tx) => { const txDb = tx as unknown as Db; if (input.executionWorkspaceId) { @@ -4453,20 +4523,35 @@ export async function startRuntimeServicesForWorkspaceControl( } // Branch reconciliation takes these same parent row locks before mutating - // a recorded branch. Holding them until the running service row is - // persisted closes the process-start window before the FK insert. - startedRefs = await startRuntimeServicesForWorkspaceControlUnlocked( + // a recorded branch. Persisting a `starting` service row before commit closes + // the process-start window without holding the DB transaction for readiness. + startBatch = await startRuntimeServicesForWorkspaceControlUnlocked( { ...input, db: txDb }, rawServices, invocationId, txDb, input.db, + { deferReadiness: true }, ); - return startedRefs; + }); + + for (const pending of startBatch.pendingReadiness) { + try { + await pending.readiness; + await persistRuntimeServiceRecord(input.db, pending.record); + } catch (error) { + await persistRuntimeServiceRecord(input.db, pending.record).catch(() => undefined); + throw error; + } + } + + return startBatch.refs.map((ref) => { + const record = runtimeServicesById.get(ref.id); + return record ? toRuntimeServiceRef(record, { reused: ref.reused }) : ref; }); } catch (error) { - for (const ref of startedRefs) { - await stopRuntimeService(ref.id).catch(() => undefined); + for (const serviceId of startBatch.startedServiceIds) { + await stopRuntimeService(serviceId).catch(() => undefined); } throw error; } diff --git a/ui/src/pages/ExecutionWorkspaceDetail.service-ports.test.ts b/ui/src/pages/ExecutionWorkspaceDetail.service-ports.test.ts new file mode 100644 index 0000000000..81ddb08726 --- /dev/null +++ b/ui/src/pages/ExecutionWorkspaceDetail.service-ports.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; +import { + getConfiguredRuntimeServicePortWarnings, + readConfiguredRuntimeServicePorts, + updateConfiguredRuntimeServicePort, +} from "./ExecutionWorkspaceDetail"; + +describe("execution workspace service port configuration", () => { + it("reads commands and legacy services, then saves a fixed port without mutating the source config", () => { + const runtimeConfig = { + commands: [ + { id: "web", name: "Web app", kind: "service", command: "pnpm dev", port: { type: "auto" } }, + { id: "migrate", name: "Migrate", kind: "job", command: "pnpm db:migrate" }, + ], + services: [{ name: "Legacy", command: "pnpm legacy", port: 3100 }], + }; + + const services = readConfiguredRuntimeServicePorts(runtimeConfig); + expect(services).toEqual([ + { collection: "commands", index: 0, name: "Web app", port: null, invalidPort: false }, + { collection: "services", index: 0, name: "Legacy", port: 3100, invalidPort: false }, + ]); + + expect(updateConfiguredRuntimeServicePort({ + runtimeConfig, + service: services[0]!, + port: "4200", + })).toEqual({ + commands: [ + { id: "web", name: "Web app", kind: "service", command: "pnpm dev", port: { type: "fixed", value: 4200 } }, + { id: "migrate", name: "Migrate", kind: "job", command: "pnpm db:migrate" }, + ], + services: [{ name: "Legacy", command: "pnpm legacy", port: 3100 }], + }); + expect(runtimeConfig.commands[0]?.port).toEqual({ type: "auto" }); + }); + + it("warns when fixed ports collide in the same workspace configuration", () => { + expect(getConfiguredRuntimeServicePortWarnings([ + { collection: "commands", index: 0, name: "Web", port: 3100, invalidPort: false }, + { collection: "commands", index: 1, name: "Admin", port: 3100, invalidPort: false }, + { collection: "services", index: 0, name: "Worker", port: 3200, invalidPort: false }, + ])).toEqual(["Port 3100 is assigned to multiple services: Web, Admin."]); + }); + + it("preserves auto-port metadata when switching between automatic and fixed ports", () => { + const runtimeConfig = { + commands: [{ name: "Web", kind: "service", command: "pnpm dev", port: { type: "auto", envKey: "APP_PORT" } }], + }; + const [service] = readConfiguredRuntimeServicePorts(runtimeConfig); + + const fixed = updateConfiguredRuntimeServicePort({ runtimeConfig, service: service!, port: "4200" }); + expect(fixed.commands).toEqual([ + { name: "Web", kind: "service", command: "pnpm dev", port: { type: "fixed", envKey: "APP_PORT", value: 4200 } }, + ]); + + expect(updateConfiguredRuntimeServicePort({ runtimeConfig: fixed, service: service!, port: "" }).commands).toEqual([ + { name: "Web", kind: "service", command: "pnpm dev", port: { type: "auto", envKey: "APP_PORT" } }, + ]); + }); + + it("marks malformed and out-of-range configured ports as invalid", () => { + expect(readConfiguredRuntimeServicePorts({ + commands: [ + { name: "Too high", kind: "service", port: { type: "fixed", value: 70000 } }, + { name: "Fractional", kind: "service", port: 3100.5 }, + { name: "String", kind: "service", port: { type: "fixed", value: "3100" } }, + ], + })).toEqual([ + { collection: "commands", index: 0, name: "Too high", port: 70000, invalidPort: true }, + { collection: "commands", index: 1, name: "Fractional", port: 3100.5, invalidPort: true }, + { collection: "commands", index: 2, name: "String", port: null, invalidPort: true }, + ]); + }); +}); diff --git a/ui/src/pages/ExecutionWorkspaceDetail.tsx b/ui/src/pages/ExecutionWorkspaceDetail.tsx index 7c01d66526..e130cf1b21 100644 --- a/ui/src/pages/ExecutionWorkspaceDetail.tsx +++ b/ui/src/pages/ExecutionWorkspaceDetail.tsx @@ -58,6 +58,14 @@ type WorkspaceFormState = { workspaceRuntime: string; }; +type ConfiguredRuntimeServicePort = { + collection: "commands" | "services"; + index: number; + name: string; + port: number | null; + invalidPort: boolean; +}; + type ExecutionWorkspaceBaseTab = "services" | "configuration" | "runtime_logs" | "issues" | "routines"; type ExecutionWorkspacePluginTab = `plugin:${string}`; type ExecutionWorkspaceTab = ExecutionWorkspaceBaseTab | ExecutionWorkspacePluginTab; @@ -164,6 +172,93 @@ function parseWorkspaceRuntimeJson(value: string) { } } +export function readConfiguredRuntimeServicePorts(runtimeConfig: Record | null) { + if (!runtimeConfig) return [] as ConfiguredRuntimeServicePort[]; + + const entries: ConfiguredRuntimeServicePort[] = []; + const addServices = (collection: ConfiguredRuntimeServicePort["collection"], services: unknown, commandsRequireServiceKind: boolean) => { + if (!Array.isArray(services)) return; + services.forEach((service, index) => { + if (!service || typeof service !== "object" || Array.isArray(service)) return; + const config = service as Record; + if (commandsRequireServiceKind && config.kind !== "service") return; + const portConfig = config.port; + const hasObjectPortValue = Boolean( + portConfig + && typeof portConfig === "object" + && !Array.isArray(portConfig) + && Object.hasOwn(portConfig, "value"), + ); + const portValue = + typeof portConfig === "number" + ? portConfig + : hasObjectPortValue + ? (portConfig as Record).value + : null; + entries.push({ + collection, + index, + name: typeof config.name === "string" && config.name.trim() ? config.name : `Service ${index + 1}`, + port: typeof portValue === "number" ? portValue : null, + invalidPort: (typeof portConfig === "number" || hasObjectPortValue) + && (typeof portValue !== "number" || !Number.isInteger(portValue) || portValue < 1 || portValue > 65535), + }); + }); + }; + + addServices("commands", runtimeConfig.commands, true); + addServices("services", runtimeConfig.services, false); + return entries; +} + +export function updateConfiguredRuntimeServicePort(input: { + runtimeConfig: Record; + service: ConfiguredRuntimeServicePort; + port: string; +}) { + const runtimeConfig = structuredClone(input.runtimeConfig); + const entries = runtimeConfig[input.service.collection]; + if (!Array.isArray(entries)) return runtimeConfig; + const entry = entries[input.service.index]; + if (!entry || typeof entry !== "object" || Array.isArray(entry)) return runtimeConfig; + const config = entry as Record; + const existingPort = config.port && typeof config.port === "object" && !Array.isArray(config.port) + ? config.port as Record + : null; + + const trimmedPort = input.port.trim(); + if (!trimmedPort) { + if (existingPort) { + const autoPort: Record = { ...existingPort, type: "auto" }; + delete autoPort.value; + config.port = autoPort; + } else { + delete config.port; + } + return runtimeConfig; + } + const port = Number(trimmedPort); + if (!Number.isInteger(port) || port < 1 || port > 65535) return runtimeConfig; + config.port = { ...existingPort, type: "fixed", value: port }; + return runtimeConfig; +} + +export function getConfiguredRuntimeServicePortWarnings(services: ConfiguredRuntimeServicePort[]) { + const servicesByPort = new Map(); + for (const service of services) { + if (service.invalidPort || !service.port) continue; + const servicesForPort = servicesByPort.get(service.port) ?? []; + servicesForPort.push(service); + servicesByPort.set(service.port, servicesForPort); + } + + return Array.from(servicesByPort.entries()) + .filter(([, servicesForPort]) => servicesForPort.length > 1) + .map(([port, servicesForPort]) => + `Port ${port} is assigned to multiple services: ${servicesForPort.map((service) => service.name).join(", ")}.`, + ); +} + function formStateFromWorkspace(workspace: ExecutionWorkspace): WorkspaceFormState { return { name: workspace.name, @@ -235,6 +330,8 @@ function validateForm(form: WorkspaceFormState) { if (!runtimeJson.ok) { return runtimeJson.error; } + const invalidPort = readConfiguredRuntimeServicePorts(runtimeJson.value).find((service) => service.invalidPort); + if (invalidPort) return `${invalidPort.name} has an invalid fixed port.`; } return null; @@ -678,6 +775,20 @@ export function ExecutionWorkspaceDetail() { ? "project_workspace" : "none"; + const configuredRuntimeConfig = useMemo(() => { + if (!form || form.inheritRuntime) return inheritedRuntimeConfig; + const parsed = parseWorkspaceRuntimeJson(form.workspaceRuntime); + return parsed.ok ? parsed.value : null; + }, [form, inheritedRuntimeConfig]); + const configuredRuntimeServicePorts = useMemo( + () => readConfiguredRuntimeServicePorts(configuredRuntimeConfig), + [configuredRuntimeConfig], + ); + const configuredRuntimeServicePortWarnings = useMemo( + () => getConfiguredRuntimeServicePortWarnings(configuredRuntimeServicePorts), + [configuredRuntimeServicePorts], + ); + const initialState = useMemo(() => (workspace ? formStateFromWorkspace(workspace) : null), [workspace]); const isDirty = Boolean(form && initialState && JSON.stringify(form) !== JSON.stringify(initialState)); const projectRef = project ? projectRouteRef(project) : workspace?.projectId ?? ""; @@ -1064,6 +1175,56 @@ export function ExecutionWorkspaceDetail() { + + {configuredRuntimeServicePorts.length > 0 ? ( +
+
+
Service ports
+

+ Set a fixed port for a service or leave it blank to use its configured automatic behavior. Editing an inherited service creates an execution-workspace runtime override. +

+
+
+ {configuredRuntimeServicePorts.map((service) => ( + + { + setForm((current) => { + if (!current) return current; + const parsed = current.inheritRuntime + ? { ok: true as const, value: inheritedRuntimeConfig } + : parseWorkspaceRuntimeJson(current.workspaceRuntime); + if (!parsed.ok || !parsed.value) return current; + return { + ...current, + inheritRuntime: false, + workspaceRuntime: formatJson(updateConfiguredRuntimeServicePort({ + runtimeConfig: parsed.value, + service, + port: event.target.value, + })), + }; + }); + }} + /> + + ))} +
+ {configuredRuntimeServicePortWarnings.length > 0 ? ( +
+ {configuredRuntimeServicePortWarnings.map((warning) =>

{warning}

)} +
+ ) : null} +

+ Paperclip checks fixed ports again when a service starts and rejects cross-workspace conflicts. +

+
+ ) : null}