From bf75361d868450bb675d23a6d56e80e86b7db304 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 8 Jun 2026 14:33:15 +0000 Subject: [PATCH] Harden workspace runtime service recovery Co-Authored-By: Paperclip --- .../src/__tests__/workspace-runtime.test.ts | 232 +++++++++++++++++- server/src/services/workspace-runtime.ts | 146 +++++++---- 2 files changed, 335 insertions(+), 43 deletions(-) diff --git a/server/src/__tests__/workspace-runtime.test.ts b/server/src/__tests__/workspace-runtime.test.ts index 35ca5601c50..c14c03bbe16 100644 --- a/server/src/__tests__/workspace-runtime.test.ts +++ b/server/src/__tests__/workspace-runtime.test.ts @@ -1,5 +1,5 @@ -import { execFile } from "node:child_process"; -import { randomUUID } from "node:crypto"; +import { execFile, spawn } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import net from "node:net"; import os from "node:os"; @@ -48,6 +48,17 @@ import { } from "./helpers/embedded-postgres.js"; const execFileAsync = promisify(execFile); + +function stableStringifyForTest(value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map((entry) => stableStringifyForTest(entry)).join(",")}]`; + } + if (value && typeof value === "object") { + const rec = value as Record; + return `{${Object.keys(rec).sort().map((key) => `${JSON.stringify(key)}:${stableStringifyForTest(rec[key])}`).join(",")}}`; + } + return JSON.stringify(value); +} const leasedRunIds = new Set(); const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -3053,6 +3064,7 @@ describeEmbeddedPostgres("workspace runtime startup reconciliation", () => { expect(service?.url).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/); await expect(fetch(service!.url!)).resolves.toMatchObject({ ok: true }); + await fs.rm(paperclipHome, { recursive: true, force: true }); await resetRuntimeServicesForTests(); const result = await reconcilePersistedRuntimeServicesOnStartup(db); @@ -3075,6 +3087,222 @@ describeEmbeddedPostgres("workspace runtime startup reconciliation", () => { await expect(fetch(service!.url!)).rejects.toThrow(); }); + it("replaces unhealthy adopted Paperclip dev services on a reused auto port", async () => { + const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-unhealthy-adopt-")); + const paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-home-")); + process.env.PAPERCLIP_HOME = paperclipHome; + process.env.PAPERCLIP_INSTANCE_ID = `runtime-unhealthy-adopt-${randomUUID()}`; + + const portProbe = net.createServer(); + await new Promise((resolve) => portProbe.listen(0, "127.0.0.1", resolve)); + const address = portProbe.address(); + const stalePort = typeof address === "object" && address ? address.port : null; + await new Promise((resolve, reject) => { + portProbe.close((err) => { + if (err) reject(err); + else resolve(); + }); + }); + expect(stalePort).toBeTypeOf("number"); + + const companyId = randomUUID(); + const agentId = randomUUID(); + const projectId = randomUUID(); + const runId = randomUUID(); + const executionWorkspaceId = randomUUID(); + const stoppedServiceId = randomUUID(); + const serviceCommand = + "node -e \"const http=require('node:http'); const stale=process.env.STALE_HEALTH==='1'; http.createServer((req,res)=>{ if (req.url==='/api/health' && stale) { res.statusCode=503; res.end('database_unreachable'); return; } res.end('ok'); }).listen(Number(process.env.PORT), '127.0.0.1')\""; + const scopeType = "agent"; + const scopeId = agentId; + const reuseKey = createHash("sha256") + .update( + stableStringifyForTest({ + scopeType, + scopeId, + serviceName: "paperclip-dev", + command: serviceCommand, + cwd: workspaceRoot, + port: null, + env: {}, + }), + ) + .digest("hex"); + + const staleProcess = spawn(resolveShell(), ["-lc", serviceCommand], { + cwd: workspaceRoot, + env: { + ...process.env, + PORT: String(stalePort), + STALE_HEALTH: "1", + }, + detached: process.platform !== "win32", + stdio: "ignore", + }); + staleProcess.unref(); + + try { + const rootUrl = `http://127.0.0.1:${stalePort}`; + const healthUrl = `${rootUrl}/api/health`; + for (let attempt = 0; attempt < 50; attempt += 1) { + try { + const response = await fetch(rootUrl); + if (response.ok) break; + } catch { + // Keep polling until the stale process has bound its port. + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + await expect(fetch(rootUrl)).resolves.toMatchObject({ ok: true }); + await expect(fetch(healthUrl)).resolves.toMatchObject({ ok: false, status: 503 }); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Codex Coder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "manual", + status: "running", + startedAt: new Date(), + updatedAt: new Date(), + }); + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Runtime unhealthy adoption test", + status: "in_progress", + }); + await db.insert(executionWorkspaces).values({ + id: executionWorkspaceId, + companyId, + projectId, + projectWorkspaceId: null, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "Runtime unhealthy adoption", + status: "active", + cwd: workspaceRoot, + providerType: "git_worktree", + providerRef: workspaceRoot, + }); + await db.insert(workspaceRuntimeServices).values({ + id: stoppedServiceId, + companyId, + projectId, + projectWorkspaceId: null, + executionWorkspaceId, + issueId: null, + scopeType, + scopeId, + serviceName: "paperclip-dev", + status: "stopped", + lifecycle: "shared", + reuseKey, + command: serviceCommand, + cwd: workspaceRoot, + port: stalePort, + url: rootUrl, + provider: "local_process", + providerRef: String(staleProcess.pid ?? ""), + ownerAgentId: null, + startedByRunId: null, + lastUsedAt: new Date(), + startedAt: new Date(), + stoppedAt: new Date(), + stopPolicy: { type: "manual" }, + healthStatus: "unknown", + }); + + leasedRunIds.add(runId); + const services = await ensureRuntimeServicesForRun({ + db, + runId, + agent: { + id: agentId, + name: "Codex Coder", + companyId, + }, + issue: null, + workspace: { + ...buildWorkspace(workspaceRoot), + projectId, + workspaceId: null, + }, + executionWorkspaceId, + config: { + workspaceRuntime: { + services: [ + { + name: "paperclip-dev", + command: serviceCommand, + cwd: ".", + port: { type: "auto" }, + readiness: { + type: "http", + urlTemplate: "http://127.0.0.1:{{port}}", + timeoutSec: 10, + intervalMs: 100, + }, + expose: { + type: "url", + urlTemplate: "http://127.0.0.1:{{port}}", + }, + lifecycle: "shared", + reuseScope: "agent", + stopPolicy: { + type: "manual", + }, + }, + ], + }, + }, + adapterEnv: {}, + }); + + expect(services).toHaveLength(1); + expect(services[0]?.reused).toBe(false); + expect(services[0]?.id).toBe(stoppedServiceId); + expect(services[0]?.port).toBe(stalePort); + await expect(fetch(healthUrl)).resolves.toMatchObject({ ok: true }); + expect(await readLocalServicePortOwner(stalePort!)).not.toBe(staleProcess.pid); + } finally { + leasedRunIds.delete(runId); + await releaseRuntimeServicesForRun(runId); + await stopRuntimeServicesForExecutionWorkspace({ + db, + executionWorkspaceId, + workspaceCwd: workspaceRoot, + }); + if (staleProcess.pid) { + try { + process.kill(-staleProcess.pid, "SIGKILL"); + } catch { + try { + process.kill(staleProcess.pid, "SIGKILL"); + } catch { + // Ignore cleanup races. + } + } + } + } + }, 20_000); + it("marks persisted local services stopped when the registry pid is stale", async () => { const companyId = randomUUID(); const runtimeServiceId = randomUUID(); diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 9681b574250..2f89eba17db 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -1901,10 +1901,43 @@ async function waitForReadiness(input: { throw new Error(`Readiness check failed for ${input.url}: ${lastError}`); } -async function isRuntimeServiceUrlHealthy(url: string | null) { - if (!url) return true; +function isPaperclipDevRuntimeService(input: { serviceName?: string | null; command?: string | null }) { + const serviceName = (input.serviceName ?? "").trim().toLowerCase(); + const command = (input.command ?? "").trim().toLowerCase(); + return ( + serviceName === "paperclip-dev" + || serviceName === "paperclip-dev-once" + || (command.includes("dev:once") && command.includes("tailscale-auth")) + ); +} + +function resolveRuntimeServiceHealthUrl( + url: string | null, + input?: { serviceName?: string | null; command?: string | null }, +) { + if (!url || !isPaperclipDevRuntimeService(input ?? {})) return url; try { - const response = await fetch(url, { signal: AbortSignal.timeout(2_000) }); + const parsed = new URL(url); + if (parsed.pathname === "/" || parsed.pathname === "") { + parsed.pathname = "/api/health"; + parsed.search = ""; + parsed.hash = ""; + return parsed.toString(); + } + } catch { + return url; + } + return url; +} + +async function isRuntimeServiceUrlHealthy( + url: string | null, + input?: { serviceName?: string | null; command?: string | null }, +) { + if (!url) return true; + const healthUrl = resolveRuntimeServiceHealthUrl(url, input); + try { + const response = await fetch(healthUrl, { signal: AbortSignal.timeout(2_000) }); return response.ok; } catch { return false; @@ -2188,42 +2221,48 @@ async function startLocalRuntimeService(input: { url, }); if (adoptedRecord) { - 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, - }; + const adoptedUrl = adoptedRecord.url ?? url; + if (!(await isRuntimeServiceUrlHealthy(adoptedUrl, { serviceName, command }))) { + await terminateLocalService(adoptedRecord); + 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, + }; + } } if (identityPort) { const ownerPid = await readLocalServicePortOwner(identityPort); @@ -2841,13 +2880,38 @@ export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) { let adopted = 0; let stopped = 0; for (const row of rows) { - const adoptedRecord = await findLocalServiceRegistryRecordByRuntimeServiceId({ + let adoptedRecord = await findLocalServiceRegistryRecordByRuntimeServiceId({ runtimeServiceId: row.id, profileKind: "workspace-runtime", }); + if (!adoptedRecord && row.command && row.cwd) { + adoptedRecord = await findAdoptableLocalService({ + serviceKey: createLocalServiceKey({ + profileKind: "workspace-runtime", + serviceName: row.serviceName, + cwd: row.cwd, + command: row.command, + envFingerprint: row.reuseKey ?? "", + port: null, + scope: { + scopeType: row.scopeType as RuntimeServiceRecord["scopeType"], + scopeId: row.scopeId ?? null, + executionWorkspaceId: row.executionWorkspaceId ?? null, + reuseKey: row.reuseKey ?? null, + }, + }), + profileKind: "workspace-runtime", + serviceName: row.serviceName, + command: row.command, + cwd: row.cwd, + envFingerprint: row.reuseKey ?? "", + port: row.port ?? null, + url: row.url ?? null, + }); + } if (adoptedRecord) { const adoptedUrl = adoptedRecord.url ?? row.url ?? null; - if (!(await isRuntimeServiceUrlHealthy(adoptedUrl))) { + if (!(await isRuntimeServiceUrlHealthy(adoptedUrl, { serviceName: row.serviceName, command: row.command }))) { await removeLocalServiceRegistryRecord(adoptedRecord.serviceKey); } else { const record: RuntimeServiceRecord = {