From 1b4de0b65c308da7f0314d1a7d21bc556c3d01df Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 18 Aug 2026 17:40:54 -0500 Subject: [PATCH] fix(workspaces): recover degraded runtime databases (#11651) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Managed workspaces run local web services and embedded PostgreSQL databases > - A listening process could return an unhealthy response and still be reused > - Embedded PostgreSQL failures had no bounded restart owner > - Cleanup inferred ownership from a branch slug instead of exact persisted instance data > - This pull request validates runtime health, supervises database recovery, and uses exact cleanup ownership > - The benefit is reliable replacement of degraded services without deleting active instances ## Linked Issues or Issue Description **What happened?** Workspace reconciliation could reuse a degraded Paperclip process after any successful HTTP response. Embedded PostgreSQL could stop without bounded recovery. Cleanup could infer database ownership from a branch slug and select the wrong instance. **Expected behavior** Paperclip must require a semantic healthy response from the assigned loopback listener. It must replace degraded processes. It must supervise embedded PostgreSQL with bounded restarts. Cleanup must use exact persisted worktree and instance-root ownership. **Steps to reproduce** 1. Start a managed workspace runtime. 2. Make its health endpoint return HTTP 200 with an unhealthy status, or stop its embedded PostgreSQL process. 3. Reconcile the workspace or run instance cleanup. 4. Observe that the old implementation can reuse the degraded runtime or infer ownership from its branch slug. **Paperclip version or commit** Current `master` before this pull request. **Deployment mode** Local dev with managed workspace services. ## What Changed - Require `{ "status": "ok" }` from the assigned loopback health endpoint before runtime reuse or adoption. - Refresh persisted runtime health and replace degraded managed processes. - Add bounded embedded PostgreSQL restart supervision with coordinated shutdown and hot-restart support. - Stop the unhealthy web process when PostgreSQL recovery is exhausted so reconciliation can replace it. - Require exact persisted instance-root ownership before cleanup can reclaim an embedded database. - Add focused regression tests for degraded HTTP responses, ownership mismatches, bounded recovery, active instance preservation, and confirmed orphan reclamation. ## Verification - `pnpm --filter @paperclipai/server typecheck` - `pnpm --filter @paperclipai/server test -- src/embedded-postgres-supervisor.test.ts src/services/workspace-instance-cleanup.test.ts src/services/workspace-runtime.test.ts src/services/execution-workspaces-service.test.ts` - `pnpm -r typecheck` - `pnpm build` - `git diff --check` - The full local stable test runner also found host-owned listeners on ports 42000 and 52000. Those listeners conflict with the exposure test fixture. The focused changed suites pass, and CI runs on a clean host. ## Risks - A custom process that returns HTTP 2xx without the Paperclip health contract is now degraded by design. - Restart exhaustion terminates the managed web process. The runtime reconciler then starts a clean process. - Cleanup now fails closed when persisted ownership is missing. This can retain an ambiguous orphan for manual review. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI GPT-5 Codex. The exact serving revision and context-window size are not exposed. The model used agentic reasoning, repository tools, code execution, and test execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [ ] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../embedded-postgres-supervisor.test.ts | 61 ++++ .../workspace-instance-cleanup.test.ts | 7 +- .../src/__tests__/workspace-runtime.test.ts | 82 ++++- server/src/embedded-postgres-supervisor.ts | 92 ++++++ server/src/index.ts | 50 ++- server/src/services/execution-workspaces.ts | 7 + .../services/workspace-instance-cleanup.ts | 9 +- server/src/services/workspace-runtime.ts | 297 ++++++++++++++++-- 8 files changed, 573 insertions(+), 32 deletions(-) create mode 100644 server/src/__tests__/embedded-postgres-supervisor.test.ts create mode 100644 server/src/embedded-postgres-supervisor.ts diff --git a/server/src/__tests__/embedded-postgres-supervisor.test.ts b/server/src/__tests__/embedded-postgres-supervisor.test.ts new file mode 100644 index 0000000000..6f1fcee809 --- /dev/null +++ b/server/src/__tests__/embedded-postgres-supervisor.test.ts @@ -0,0 +1,61 @@ +import { EventEmitter } from "node:events"; +import { describe, expect, it, vi } from "vitest"; +import { createEmbeddedPostgresSupervisor, type SupervisedEmbeddedPostgres } from "../embedded-postgres-supervisor.js"; + +function createInstance(startError?: Error) { + const process = new EventEmitter(); + const instance: SupervisedEmbeddedPostgres = { + process, + start: vi.fn(async () => { if (startError) throw startError; }), + stop: vi.fn(async () => undefined), + }; + return { instance, process }; +} + +describe("embedded PostgreSQL supervisor", () => { + it("restarts PostgreSQL after its managed child exits unexpectedly", async () => { + const initial = createInstance(); + const replacement = createInstance(); + const onRestarted = vi.fn(); + const supervisor = createEmbeddedPostgresSupervisor({ + initialInstance: initial.instance, + createInstance: () => replacement.instance, + restartDelaysMs: [0], + onRestarted, + }); + initial.process.emit("exit", 137, "SIGKILL"); + await supervisor.waitForRecovery(); + expect(replacement.instance.start).toHaveBeenCalledOnce(); + expect(supervisor.current()).toBe(replacement.instance); + expect(onRestarted).toHaveBeenCalledWith(1); + }); + + it("does not restart PostgreSQL during orderly shutdown", async () => { + const initial = createInstance(); + const createReplacement = vi.fn(() => createInstance().instance); + const supervisor = createEmbeddedPostgresSupervisor({ + initialInstance: initial.instance, + createInstance: createReplacement, + restartDelaysMs: [0], + }); + await supervisor.shutdown(); + expect(initial.instance.stop).toHaveBeenCalledOnce(); + expect(createReplacement).not.toHaveBeenCalled(); + }); + + it("bounds recovery attempts and reports the final failure", async () => { + const initial = createInstance(); + const failures = [new Error("first"), new Error("second"), new Error("third")]; + const onRecoveryExhausted = vi.fn(); + const supervisor = createEmbeddedPostgresSupervisor({ + initialInstance: initial.instance, + createInstance: () => createInstance(failures.shift()).instance, + restartDelaysMs: [0, 0, 0], + onRecoveryExhausted, + }); + initial.process.emit("exit", 1, null); + await supervisor.waitForRecovery(); + expect(onRecoveryExhausted).toHaveBeenCalledOnce(); + expect(onRecoveryExhausted).toHaveBeenCalledWith(expect.objectContaining({ message: "third" })); + }); +}); diff --git a/server/src/__tests__/workspace-instance-cleanup.test.ts b/server/src/__tests__/workspace-instance-cleanup.test.ts index c46b3bc272..353fceec72 100644 --- a/server/src/__tests__/workspace-instance-cleanup.test.ts +++ b/server/src/__tests__/workspace-instance-cleanup.test.ts @@ -114,7 +114,7 @@ describe("worktree instance cleanup", () => { await expect(fs.stat(instanceRoot)).rejects.toMatchObject({ code: "ENOENT" }); }); - it("falls back to deterministic instance ownership when persisted root metadata is absent", async () => { + it("preserves a collision-resistant active instance when persisted ownership is absent", async () => { const worktreesDir = await makeTempRoot("paperclip-managed-worktrees-"); const workspacePath = await makeTempRoot("paperclip-cleanup-workspace-"); const instanceId = deriveWorktreeInstanceId(workspacePath); @@ -133,8 +133,9 @@ describe("worktree instance cleanup", () => { worktreesDir, }); - expect(result).toMatchObject({ status: "removed", instanceRoot }); - await expect(fs.stat(instanceRoot)).rejects.toMatchObject({ code: "ENOENT" }); + expect(result).toMatchObject({ status: "refused", instanceRoot }); + expect((result as { warning: string }).warning).toContain("no persisted instance root"); + await expect(fs.readFile(path.join(instanceRoot, "marker"), "utf8")).resolves.toBe("remove me"); }); it("refuses and logs an instance pointer outside the managed worktree root", async () => { diff --git a/server/src/__tests__/workspace-runtime.test.ts b/server/src/__tests__/workspace-runtime.test.ts index 6d33393bb0..b96d455a05 100644 --- a/server/src/__tests__/workspace-runtime.test.ts +++ b/server/src/__tests__/workspace-runtime.test.ts @@ -3919,6 +3919,86 @@ describe("ensureRuntimeServicesForRun", () => { } }); + it("replaces a reused Paperclip dev runtime whose 2xx health payload is unhealthy", async () => { + const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-misreported-health-")); + const workspace = buildWorkspace(workspaceRoot); + const serviceCommand = + "node -e \"let healthy=true;const http=require('node:http');http.createServer((req,res)=>{if(req.url==='/misreport'){healthy=false;res.end('failed');return;}if(req.url==='/api/health'){res.setHeader('content-type','application/json');res.end(JSON.stringify(healthy?{status:'ok'}:{status:'unhealthy',error:'database_unreachable'}));return;}res.end('ok')}).listen(Number(process.env.PORT),'127.0.0.1')\""; + const input = { + actor: { id: "agent-1", name: "Codex Coder", companyId: "company-1" }, + issue: null, + workspace, + executionWorkspaceId: "execution-workspace-health", + config: { workspaceRuntime: { services: [{ + name: "paperclip-dev", + command: serviceCommand, + cwd: ".", + port: { type: "auto" as const }, + readiness: { type: "http" as const, urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 3, intervalMs: 100 }, + expose: { type: "url" as const, urlTemplate: "http://127.0.0.1:{{port}}" }, + lifecycle: "shared" as const, + stopPolicy: { type: "manual" as const }, + }] } }, + adapterEnv: {}, + }; + try { + const [first] = await startRuntimeServicesForWorkspaceControl(input); + await expect(fetch(`${first!.url}/misreport`)).resolves.toMatchObject({ ok: true }); + await expect(fetch(`${first!.url}/api/health`)).resolves.toMatchObject({ ok: true }); + const [[replacement], [concurrentReuse]] = await Promise.all([ + startRuntimeServicesForWorkspaceControl(input), + startRuntimeServicesForWorkspaceControl(input), + ]); + expect(replacement?.id).not.toBe(first?.id); + expect(replacement?.reused).toBe(false); + expect(concurrentReuse?.id).toBe(replacement?.id); + expect(concurrentReuse?.reused).toBe(true); + } finally { + await stopRuntimeServicesForExecutionWorkspace({ + executionWorkspaceId: "execution-workspace-health", + workspaceCwd: workspaceRoot, + }); + await fs.rm(workspaceRoot, { recursive: true, force: true }); + } + }); + + it("reuses a shared Paperclip dev runtime after one transient unhealthy response", async () => { + const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-transient-health-")); + const workspace = buildWorkspace(workspaceRoot); + const serviceCommand = + "node -e \"let failNext=false;const http=require('node:http');http.createServer((req,res)=>{if(req.url==='/fail-next'){failNext=true;res.end('armed');return;}if(req.url==='/api/health'){res.setHeader('content-type','application/json');const healthy=!failNext;failNext=false;res.end(JSON.stringify({status:healthy?'ok':'unhealthy'}));return;}res.end('ok')}).listen(Number(process.env.PORT),'127.0.0.1')\""; + const input = { + actor: { id: "agent-1", name: "Codex Coder", companyId: "company-1" }, + issue: null, + workspace, + executionWorkspaceId: "execution-workspace-transient-health", + config: { workspaceRuntime: { services: [{ + name: "paperclip-dev", + command: serviceCommand, + cwd: ".", + port: { type: "auto" as const }, + readiness: { type: "http" as const, urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 3, intervalMs: 100 }, + expose: { type: "url" as const, urlTemplate: "http://127.0.0.1:{{port}}" }, + lifecycle: "shared" as const, + stopPolicy: { type: "manual" as const }, + }] } }, + adapterEnv: {}, + }; + try { + const [first] = await startRuntimeServicesForWorkspaceControl(input); + await expect(fetch(`${first!.url}/fail-next`)).resolves.toMatchObject({ ok: true }); + const [reused] = await startRuntimeServicesForWorkspaceControl(input); + expect(reused?.id).toBe(first?.id); + expect(reused?.reused).toBe(true); + } finally { + await stopRuntimeServicesForExecutionWorkspace({ + executionWorkspaceId: "execution-workspace-transient-health", + workspaceCwd: workspaceRoot, + }); + await fs.rm(workspaceRoot, { recursive: true, force: true }); + } + }); + it("uses explicit readiness URL when exposed URL is not the local probe address", async () => { const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-explicit-readiness-")); const workspace = buildWorkspace(workspaceRoot); @@ -6803,7 +6883,7 @@ describeEmbeddedPostgres("workspace runtime startup reconciliation", () => { const projectWorkspaceId = randomUUID(); // Binds the app port and its HMR companion, both loopback-only. const command = - "node -e \"const http=require('node:http');const p=Number(process.env.PORT);for(const q of [p,p+10000])http.createServer((req,res)=>res.end('ok')).listen(q,'127.0.0.1');setInterval(()=>{},1000)\""; + "node -e \"const http=require('node:http');const p=Number(process.env.PORT);for(const q of [p,p+10000])http.createServer((req,res)=>{if(req.url==='/api/health'){res.setHeader('content-type','application/json');res.end(JSON.stringify({status:'ok'}));return;}res.end('ok')}).listen(q,'127.0.0.1');setInterval(()=>{},1000)\""; const workspaceRuntime = { services: [ { diff --git a/server/src/embedded-postgres-supervisor.ts b/server/src/embedded-postgres-supervisor.ts new file mode 100644 index 0000000000..399a248b0b --- /dev/null +++ b/server/src/embedded-postgres-supervisor.ts @@ -0,0 +1,92 @@ +export type EmbeddedPostgresExitListener = (code: number | null, signal: NodeJS.Signals | null) => void; + +export interface SupervisedEmbeddedPostgres { + start(): Promise; + stop(): Promise; + process?: { once(event: "exit", listener: EmbeddedPostgresExitListener): unknown }; +} + +export interface EmbeddedPostgresSupervisor { + current(): SupervisedEmbeddedPostgres; + shutdown(): Promise; + waitForRecovery(): Promise; +} + +type Options = { + initialInstance: SupervisedEmbeddedPostgres; + createInstance: () => SupervisedEmbeddedPostgres; + beforeRestart?: (attempt: number) => Promise | void; + restartDelaysMs?: number[]; + delay?: (milliseconds: number) => Promise; + onUnexpectedExit?: EmbeddedPostgresExitListener; + onRestartAttemptFailed?: (error: unknown, attempt: number) => void; + onRestarted?: (attempt: number) => void; + onRecoveryExhausted?: (error: unknown) => void; +}; + +const defaultDelay = (milliseconds: number) => new Promise((resolve) => setTimeout(resolve, milliseconds)); + +export function createEmbeddedPostgresSupervisor(options: Options): EmbeddedPostgresSupervisor { + const restartDelaysMs = options.restartDelaysMs ?? [0, 250, 1_000]; + const wait = options.delay ?? defaultDelay; + let activeInstance = options.initialInstance; + let activeInstanceExited = false; + let shuttingDown = false; + let recoveryPromise: Promise | null = null; + + const recover = async () => { + let lastError: unknown = new Error("Embedded PostgreSQL exited unexpectedly"); + for (let index = 0; index < restartDelaysMs.length; index += 1) { + if (shuttingDown) return; + const attempt = index + 1; + const delayMs = restartDelaysMs[index] ?? 0; + if (delayMs > 0) await wait(delayMs); + if (shuttingDown) return; + try { + await options.beforeRestart?.(attempt); + const replacement = options.createInstance(); + await replacement.start(); + if (shuttingDown) { + await replacement.stop(); + return; + } + activeInstance = replacement; + activeInstanceExited = false; + monitor(replacement); + options.onRestarted?.(attempt); + return; + } catch (error) { + lastError = error; + options.onRestartAttemptFailed?.(error, attempt); + } + } + if (!shuttingDown) options.onRecoveryExhausted?.(lastError); + }; + + const monitor = (instance: SupervisedEmbeddedPostgres) => { + const child = instance.process; + if (!child) { + options.onRecoveryExhausted?.(new Error("Embedded PostgreSQL started without a child process to monitor")); + return; + } + child.once("exit", (code, signal) => { + if (activeInstance !== instance) return; + activeInstanceExited = true; + if (shuttingDown) return; + options.onUnexpectedExit?.(code, signal); + recoveryPromise = recover().finally(() => { recoveryPromise = null; }); + }); + }; + + monitor(activeInstance); + return { + current: () => activeInstance, + waitForRecovery: async () => { await recoveryPromise; }, + shutdown: async () => { + if (shuttingDown) return; + shuttingDown = true; + await recoveryPromise; + if (!activeInstanceExited) await activeInstance.stop(); + }, + }; +} diff --git a/server/src/index.ts b/server/src/index.ts index ffbf98ec3d..eca86caf51 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -96,6 +96,11 @@ import { } from "./shutdown.js"; import { systemdNotify } from "./services/systemd-notify.js"; import { flushInFlightRunLogMirrors } from "./services/run-log-store.js"; +import { + createEmbeddedPostgresSupervisor, + type EmbeddedPostgresSupervisor, + type SupervisedEmbeddedPostgres, +} from "./embedded-postgres-supervisor.js"; import type { InstanceDatabaseBackupRunResult, InstanceDatabaseBackupTrigger, @@ -112,10 +117,8 @@ type BetterAuthSessionResult = { user: BetterAuthSessionUser | null; }; -type EmbeddedPostgresInstance = { +type EmbeddedPostgresInstance = SupervisedEmbeddedPostgres & { initialise(): Promise; - start(): Promise; - stop(): Promise; }; type EmbeddedPostgresCtor = new (opts: { @@ -326,6 +329,7 @@ export async function startServer(): Promise { let db; let pluginMigrationDb; let embeddedPostgres: EmbeddedPostgresInstance | null = null; + let embeddedPostgresSupervisor: EmbeddedPostgresSupervisor | null = null; let embeddedPostgresStartedByThisProcess = false; let migrationSummary: MigrationSummary = "skipped"; let activeDatabaseConnectionString: string; @@ -450,7 +454,7 @@ export async function startServer(): Promise { } port = detectedPort; logger.info(`Using embedded PostgreSQL because no DATABASE_URL set (dataDir=${dataDir}, port=${port})`); - embeddedPostgres = new EmbeddedPostgres({ + const createEmbeddedPostgres = () => new EmbeddedPostgres({ databaseDir: dataDir, user: "paperclip", password: "paperclip", @@ -460,6 +464,7 @@ export async function startServer(): Promise { onLog: appendEmbeddedPostgresLog, onError: appendEmbeddedPostgresLog, }); + embeddedPostgres = createEmbeddedPostgres(); if (!clusterAlreadyInitialized) { try { @@ -489,6 +494,36 @@ export async function startServer(): Promise { }); } embeddedPostgresStartedByThisProcess = true; + embeddedPostgresSupervisor = createEmbeddedPostgresSupervisor({ + initialInstance: embeddedPostgres, + createInstance: createEmbeddedPostgres, + beforeRestart: () => { + const runningPostgresPid = getRunningPid(); + if (runningPostgresPid) { + throw new Error(`Refusing embedded PostgreSQL recovery because the data directory reports a live process (pid=${runningPostgresPid})`); + } + if (existsSync(postmasterPidFile)) rmSync(postmasterPidFile, { force: true }); + }, + onUnexpectedExit: (code, signal) => logger.error( + { code, signal, recentLogs: logBuffer.getRecentLogs() }, + "Embedded PostgreSQL exited unexpectedly; attempting recovery", + ), + onRestartAttemptFailed: (err, attempt) => logger.error( + { err, attempt, recentLogs: logBuffer.getRecentLogs() }, + "Embedded PostgreSQL recovery attempt failed", + ), + onRestarted: (attempt) => logger.info( + { attempt, port }, + "Embedded PostgreSQL recovered after unexpected exit", + ), + onRecoveryExhausted: (err) => { + logger.fatal( + { err, recentLogs: logBuffer.getRecentLogs() }, + "Embedded PostgreSQL recovery exhausted; stopping the unhealthy server", + ); + process.kill(process.pid, "SIGTERM"); + }, + }); } } @@ -1635,8 +1670,9 @@ export async function startServer(): Promise { const appShutdown = (app as { locals?: { paperclipShutdown?: () => Promise } }).locals ?.paperclipShutdown; - const embeddedPostgresToStop = - embeddedPostgres && embeddedPostgresStartedByThisProcess ? embeddedPostgres : null; + const stopEmbeddedPostgres = embeddedPostgres && embeddedPostgresStartedByThisProcess + ? () => embeddedPostgresSupervisor?.shutdown() ?? embeddedPostgres!.stop() + : null; // Await the ordered application teardown before the process exits. A live // setup-token login session must stop and release its sandbox lease before @@ -1645,7 +1681,7 @@ export async function startServer(): Promise { await finalizeServerShutdown({ signal, shutdownAppServices: appShutdown, - stopEmbeddedPostgres: embeddedPostgresToStop ? () => embeddedPostgresToStop.stop() : null, + stopEmbeddedPostgres, shutdownInstrumentation, log: logger, }); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 360a5042f1..54b0a562f1 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -2218,6 +2218,13 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic .where(eq(executionWorkspaces.id, id)) .then((rows) => rows[0] ?? null); if (!row) return null; + const { refreshPersistedRuntimeServiceHealth } = await import("./workspace-runtime.js"); + await refreshPersistedRuntimeServiceHealth({ + db, + companyId: row.companyId, + executionWorkspaceId: row.id, + projectWorkspaceId: row.projectWorkspaceId, + }); const runtimeServicesByWorkspaceId = await loadEffectiveRuntimeServicesByExecutionWorkspace(db, row.companyId, [row]); return hydrateWorkspace( row, diff --git a/server/src/services/workspace-instance-cleanup.ts b/server/src/services/workspace-instance-cleanup.ts index bcfe06f5b0..9f5cd9d018 100644 --- a/server/src/services/workspace-instance-cleanup.ts +++ b/server/src/services/workspace-instance-cleanup.ts @@ -289,9 +289,12 @@ export async function cleanupWorktreeInstanceArtifacts(input: { return { status: "refused", instanceRoot: configuredInstanceRoot, warning }; } - const expectedInstanceRoot = input.expectedInstanceRoot - ? path.resolve(input.expectedInstanceRoot) - : null; + const expectedInstanceRoot = input.expectedInstanceRoot ? path.resolve(input.expectedInstanceRoot) : null; + if (!expectedInstanceRoot) { + warning = `Refusing to remove instance directory "${configuredInstanceRoot}" because execution workspace ${input.workspaceId} has no persisted instance root.`; + await recordRefusal({ refusalReason: "persisted_instance_root_missing" }); + return { status: "refused", instanceRoot: configuredInstanceRoot, warning }; + } if (expectedInstanceRoot && configuredInstanceRoot !== expectedInstanceRoot) { warning = `Refusing to remove instance directory "${configuredInstanceRoot}" because it does not match execution workspace ${input.workspaceId}'s persisted instance root "${expectedInstanceRoot}".`; await recordRefusal({ diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 0b04d2a4e5..3753fe60eb 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -232,6 +232,8 @@ const runtimeServicesById = new Map(); const runtimeServicesByReuseKey = new Map(); const runtimeServiceLeasesByRun = new Map(); const runtimeProvisionByWorkspace = new Map>(); +const runtimeControlStartByOwner = new Map>(); +const runtimeReplacementClaimsByReuseKey = new Map(); const quarantinedRuntimeExposurePorts = new Set(); /** * Pair-atomic in-process claims for exposure allocations that have not bound a @@ -477,6 +479,8 @@ export async function resetRuntimeServicesForTests( runtimeServicesByReuseKey.clear(); runtimeServiceLeasesByRun.clear(); runtimeProvisionByWorkspace.clear(); + runtimeControlStartByOwner.clear(); + runtimeReplacementClaimsByReuseKey.clear(); quarantinedRuntimeExposurePorts.clear(); exposurePortPairClaims.clear(); workspaceRuntimeExposureDeps = defaultWorkspaceRuntimeExposureDeps(); @@ -4726,14 +4730,26 @@ function resolveRuntimeServiceHealthUrl( async function isRuntimeServiceUrlHealthy( url: string | null, - input?: { serviceName?: string | null; command?: string | null }, + input?: { + serviceName?: string | null; + command?: string | null; + provider?: string | null; + port?: number | null; + }, ) { - if (!url) return true; - const healthUrl = resolveRuntimeServiceHealthUrl(url, input); + const localProbeUrl = input?.provider === "local_process" && input.port && isPaperclipDevRuntimeService(input) + ? `http://127.0.0.1:${input.port}` + : null; + const probeUrl = localProbeUrl ?? url; + if (!probeUrl) return true; + const healthUrl = resolveRuntimeServiceHealthUrl(probeUrl, input); if (!healthUrl) return false; try { const response = await fetch(healthUrl, { signal: AbortSignal.timeout(2_000) }); - return response.ok; + if (!response.ok) return false; + if (!isPaperclipDevRuntimeService(input ?? {})) return true; + const payload = await response.json().catch(() => null) as { status?: unknown } | null; + return payload?.status === "ok"; } catch { return false; } @@ -5931,6 +5947,37 @@ async function stopRuntimeService(serviceId: string) { await persistRuntimeServiceRecord(record.db, record); } +async function findHealthyRunningRuntimeService(reuseKey: string | null) { + const existingId = reuseKey ? runtimeServicesByReuseKey.get(reuseKey) : null; + const existing = existingId ? runtimeServicesById.get(existingId) : null; + if (!existing || existing.status !== "running") return null; + const healthInput = { + serviceName: existing.serviceName, + command: existing.command, + provider: existing.provider, + port: existing.port, + }; + let healthy = await isRuntimeServiceUrlHealthy(existing.url, healthInput); + if (!healthy) { + // A single timeout or connection reset is not enough evidence to destroy a + // shared runtime that active runs may still use. Confirm the failure after + // a short bounded delay before entering the destructive replacement path. + await delay(250); + healthy = await isRuntimeServiceUrlHealthy(existing.url, healthInput); + } + if (healthy) return existing; + if (existing.leaseRunIds.size > 0) { + existing.healthStatus = "unhealthy"; + if (reuseKey && runtimeServicesByReuseKey.get(reuseKey) === existing.id) { + runtimeServicesByReuseKey.delete(reuseKey); + } + await persistRuntimeServiceRecord(existing.db, existing); + return null; + } + await stopRuntimeService(existing.id); + return null; +} + async function markPersistedRuntimeServicesStoppedForExecutionWorkspace(input: { db: Db; executionWorkspaceId: string; @@ -6254,7 +6301,7 @@ async function isPersistedIsolatedExecutionWorkspace(input: { return row?.mode === "isolated_workspace"; } -export async function ensureRuntimeServicesForRun(input: { +type EnsureRuntimeServicesForRunInput = { db?: Db; runId: string; agent: ExecutionWorkspaceAgentRef; @@ -6265,7 +6312,11 @@ export async function ensureRuntimeServicesForRun(input: { adapterEnv: Record; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; recorder?: WorkspaceOperationRecorder | null; -}): Promise { +}; + +async function ensureRuntimeServicesForRunInvocation( + input: EnsureRuntimeServicesForRunInput, +): Promise { const rawServices = selectRuntimeServiceEntries({ config: input.config, respectDesiredStates: true, @@ -6304,9 +6355,8 @@ export async function ensureRuntimeServicesForRun(input: { }).reuseKey; if (reuseKey) { - const existingId = runtimeServicesByReuseKey.get(reuseKey); - const existing = existingId ? runtimeServicesById.get(existingId) : null; - if (existing && existing.status === "running") { + const existing = await findHealthyRunningRuntimeService(reuseKey); + if (existing) { existing.leaseRunIds.add(input.runId); existing.lastUsedAt = new Date().toISOString(); existing.stoppedAt = null; @@ -6354,6 +6404,124 @@ export async function ensureRuntimeServicesForRun(input: { return refs; } +async function withRuntimeStartMutex(ownerKey: string, start: () => Promise): Promise { + const previous = runtimeControlStartByOwner.get(ownerKey) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise((resolve) => { + release = resolve; + }); + const queued = previous.then(() => current); + runtimeControlStartByOwner.set(ownerKey, queued); + await previous; + try { + return await start(); + } finally { + release(); + if (runtimeControlStartByOwner.get(ownerKey) === queued) runtimeControlStartByOwner.delete(ownerKey); + } +} + +function resolveRuntimeStartMutexPlan(input: { + services: Array>; + workspace: RealizedExecutionWorkspace; + executionWorkspaceId?: string | null; + issue: ExecutionWorkspaceIssueRef | null; + runId: string; + agent: ExecutionWorkspaceAgentRef; + adapterEnv: Record; +}) { + const fallbackOwnerId = input.executionWorkspaceId + ?? input.workspace.workspaceId + ?? path.resolve(input.workspace.cwd); + const replacementReuseKeys: string[] = []; + const keys = input.services.map((service) => { + const { scopeType, scopeId } = resolveServiceScopeId({ + service, + workspace: input.workspace, + executionWorkspaceId: input.executionWorkspaceId, + issue: input.issue, + runId: input.runId, + agent: input.agent, + }); + const reuseKey = resolveRuntimeServiceReuseIdentity({ + service, + workspace: input.workspace, + agent: input.agent, + issue: input.issue, + adapterEnv: input.adapterEnv, + scopeType, + scopeId, + }).reuseKey; + // Converge all callers that can replace an existing shared runtime on its + // reuse identity. For an initial start, retain owner-level concurrency so + // the exposure allocator's in-flight pair claims remain authoritative. + if ( + reuseKey + && (runtimeServicesByReuseKey.has(reuseKey) || runtimeReplacementClaimsByReuseKey.has(reuseKey)) + ) { + runtimeReplacementClaimsByReuseKey.set( + reuseKey, + (runtimeReplacementClaimsByReuseKey.get(reuseKey) ?? 0) + 1, + ); + replacementReuseKeys.push(reuseKey); + return `reuse:${reuseKey}`; + } + return `${input.agent.companyId}:owner:${fallbackOwnerId}`; + }); + return { + ownerKeys: [...new Set(keys)].sort(), + replacementReuseKeys, + }; +} + +function releaseRuntimeReplacementClaims(reuseKeys: string[]) { + for (const reuseKey of reuseKeys) { + const next = (runtimeReplacementClaimsByReuseKey.get(reuseKey) ?? 1) - 1; + if (next <= 0) runtimeReplacementClaimsByReuseKey.delete(reuseKey); + else runtimeReplacementClaimsByReuseKey.set(reuseKey, next); + } +} + +async function withRuntimeStartMutexes( + ownerKeys: string[], + start: () => Promise, +): Promise { + const acquire = async (index: number): Promise => { + const ownerKey = ownerKeys[index]; + if (!ownerKey) return await start(); + return await withRuntimeStartMutex(ownerKey, () => acquire(index + 1)); + }; + return await acquire(0); +} + +export async function ensureRuntimeServicesForRun( + input: EnsureRuntimeServicesForRunInput, +): Promise { + const services = selectRuntimeServiceEntries({ + config: input.config, + respectDesiredStates: true, + defaultDesiredState: readDesiredRuntimeState(input.config.desiredState) ?? "running", + serviceStates: readConfiguredServiceStates(input.config), + }); + const mutexPlan = resolveRuntimeStartMutexPlan({ + services, + workspace: input.workspace, + executionWorkspaceId: input.executionWorkspaceId, + issue: input.issue, + runId: input.runId, + agent: input.agent, + adapterEnv: input.adapterEnv, + }); + try { + return await withRuntimeStartMutexes( + mutexPlan.ownerKeys, + () => ensureRuntimeServicesForRunInvocation(input), + ); + } finally { + releaseRuntimeReplacementClaims(mutexPlan.replacementReuseKeys); + } +} + type StartRuntimeServicesForWorkspaceControlInput = { db?: Db; invocationId?: string; @@ -6417,9 +6585,8 @@ async function startRuntimeServicesForWorkspaceControlUnlocked( }).reuseKey; if (reuseKey) { - const existingId = runtimeServicesByReuseKey.get(reuseKey); - const existing = existingId ? runtimeServicesById.get(existingId) : null; - if (existing && existing.status === "running") { + const existing = await findHealthyRunningRuntimeService(reuseKey); + if (existing) { const prepared = options?.preparedProvisioning; if (prepared?.service === service && prepared.record.id !== existing.id && persistenceDb) { await persistenceDb @@ -6543,7 +6710,7 @@ async function discardFailedDeferredRuntimeStart(db: Db, record: RuntimeServiceR await persistRuntimeServiceRecord(db, record); } -export async function startRuntimeServicesForWorkspaceControl( +async function startRuntimeServicesForWorkspaceControlInvocation( input: StartRuntimeServicesForWorkspaceControlInput, ): Promise { const rawServices = selectRuntimeServiceEntries({ @@ -6608,9 +6775,8 @@ export async function startRuntimeServicesForWorkspaceControl( scopeType, scopeId, }).reuseKey; - const existingId = reuseKey ? runtimeServicesByReuseKey.get(reuseKey) : null; - const existing = existingId ? runtimeServicesById.get(existingId) : null; - if (existing?.status === "running") continue; + const existing = await findHealthyRunningRuntimeService(reuseKey); + if (existing) continue; const record = await prepareRuntimeProvisioning({ db: input.db, @@ -6762,6 +6928,36 @@ export async function startRuntimeServicesForWorkspaceControl( } } +export async function startRuntimeServicesForWorkspaceControl( + input: StartRuntimeServicesForWorkspaceControlInput, +): Promise { + const services = selectRuntimeServiceEntries({ + config: input.config, + serviceIndex: input.serviceIndex, + respectDesiredStates: input.respectDesiredStates, + defaultDesiredState: readDesiredRuntimeState(input.config.desiredState) ?? "stopped", + serviceStates: readConfiguredServiceStates(input.config), + }); + const invocationId = input.invocationId ?? "workspace_control"; + const mutexPlan = resolveRuntimeStartMutexPlan({ + services, + workspace: input.workspace, + executionWorkspaceId: input.executionWorkspaceId, + issue: input.issue, + runId: invocationId, + agent: input.actor, + adapterEnv: input.adapterEnv, + }); + try { + return await withRuntimeStartMutexes( + mutexPlan.ownerKeys, + () => startRuntimeServicesForWorkspaceControlInvocation(input), + ); + } finally { + releaseRuntimeReplacementClaims(mutexPlan.replacementReuseKeys); + } +} + export async function releaseRuntimeServicesForRun(runId: string) { const acquired = runtimeServiceLeasesByRun.get(runId) ?? []; runtimeServiceLeasesByRun.delete(runId); @@ -6773,7 +6969,14 @@ export async function releaseRuntimeServicesForRun(runId: string) { const stopType = asString(record.stopPolicy?.type, record.lifecycle === "ephemeral" ? "on_run_finish" : "manual"); await persistRuntimeServiceRecord(record.db, record); if (record.leaseRunIds.size === 0) { - if (record.lifecycle === "ephemeral" || stopType === "on_run_finish") { + const detachedUnhealthySharedRuntime = record.healthStatus === "unhealthy" + && Boolean(record.reuseKey) + && runtimeServicesByReuseKey.get(record.reuseKey!) !== record.id; + if ( + record.lifecycle === "ephemeral" + || stopType === "on_run_finish" + || detachedUnhealthySharedRuntime + ) { await stopRuntimeService(serviceId); continue; } @@ -7004,6 +7207,59 @@ async function buildPersistedRuntimeExposureIntentLookup(db: Db) { }; } +export async function refreshPersistedRuntimeServiceHealth(input: { + db: Db; + companyId: string; + executionWorkspaceId: string; + projectWorkspaceId?: string | null; +}) { + const ownershipCondition = input.projectWorkspaceId + ? or( + eq(workspaceRuntimeServices.executionWorkspaceId, input.executionWorkspaceId), + and( + eq(workspaceRuntimeServices.projectWorkspaceId, input.projectWorkspaceId), + eq(workspaceRuntimeServices.scopeType, "project_workspace"), + ), + ) + : eq(workspaceRuntimeServices.executionWorkspaceId, input.executionWorkspaceId); + const rows = await input.db + .select({ + id: workspaceRuntimeServices.id, + serviceName: workspaceRuntimeServices.serviceName, + command: workspaceRuntimeServices.command, + provider: workspaceRuntimeServices.provider, + port: workspaceRuntimeServices.port, + url: workspaceRuntimeServices.url, + healthStatus: workspaceRuntimeServices.healthStatus, + }) + .from(workspaceRuntimeServices) + .where(and( + eq(workspaceRuntimeServices.companyId, input.companyId), + eq(workspaceRuntimeServices.provider, "local_process"), + eq(workspaceRuntimeServices.status, "running"), + ownershipCondition, + )); + const results = await Promise.all(rows.map(async (row) => ({ + row, + healthStatus: await isRuntimeServiceUrlHealthy(row.url, row) ? "healthy" as const : "unhealthy" as const, + }))); + await Promise.all(results.map(async ({ row, healthStatus }) => { + const liveRecord = runtimeServicesById.get(row.id); + if (liveRecord) liveRecord.healthStatus = healthStatus; + if (row.healthStatus === healthStatus) return; + await input.db.update(workspaceRuntimeServices).set({ healthStatus, updatedAt: new Date() }).where(and( + eq(workspaceRuntimeServices.id, row.id), + eq(workspaceRuntimeServices.companyId, input.companyId), + eq(workspaceRuntimeServices.status, "running"), + )); + })); + return { + checked: results.length, + healthy: results.filter((result) => result.healthStatus === "healthy").length, + unhealthy: results.filter((result) => result.healthStatus === "unhealthy").length, + }; +} + export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) { const rows = await db .select() @@ -7185,7 +7441,12 @@ export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) { if ( backfillDecision.action === "reprovision" || !exposureHealthMatches - || !(await isRuntimeServiceUrlHealthy(adoptedUrl, { serviceName: row.serviceName, command: row.command })) + || !(await isRuntimeServiceUrlHealthy(adoptedUrl, { + serviceName: row.serviceName, + command: row.command, + provider: "local_process", + port: adoptedRecord.port ?? row.port, + })) ) { if (backfillDecision.action === "reprovision") backfilled += 1; await terminateLocalService(adoptedRecord);