diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index cc944cdf8a1..ec09bf74eb1 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1087,6 +1087,20 @@ that classification finishes. - A verified live runner re-registers its existing PRP authority and reconnects with the same operating-system PID. Paperclip does not spawn a competing runner. +- For a running sandbox session, recovery checks the original provider lease, + remote workspace, durable runner identity, and process marker inside that + sandbox. The process marker must include the Linux boot ID and start ticks; + recovery compares them with the live process before adoption and signaling. + Collection uses the required Node runtime and is optional for fresh launches. Older + markers or images without that proof remain blocked for recovery. Remote + PIDs are never interpreted as controller-local PIDs. The runner must + authenticate to its existing PRP authority; reconnection neither + launches another provider nor consumes a provider retry. A replacement + sandbox or mismatched identity blocks adoption without overwriting evidence. +- Shutdown waits up to 30 seconds for an in-progress native startup to reach + its detach acknowledgement. It reports a startup deadline failure instead + of claiming that an unfinished bootstrap detached safely. A queued turn + waits for the previous executor to finish releasing its task resources. - A verified dead runner starts a replacement from the same durable root and resumes the same provider checkpoint. Only the operating-system PID changes. - A runner that died before its first authenticated connection can restart on diff --git a/docker/daytona-runner/Dockerfile b/docker/daytona-runner/Dockerfile index 83492fb143e..88cf863aa62 100644 --- a/docker/daytona-runner/Dockerfile +++ b/docker/daytona-runner/Dockerfile @@ -28,7 +28,7 @@ COPY cli/package.json ./cli/package.json # The complete resolved lock (including transitive integrity hashes) is reviewed. # Reject registry-time drift BEFORE installing packages or running lifecycle code. # Refresh this digest together with source/provider dependency changes. -ARG PAPERCLIP_RUNNER_LOCK_SHA256=133cc415964e4ee3f348251c315d560cc8f8f51a8dddde2da84e065ba4ae3fa6 +ARG PAPERCLIP_RUNNER_LOCK_SHA256=d7d96cf0d98cf0946f6195e29ba173b03711a947a1c38f312b67cda56c254c22 RUN pnpm install --resolution-only --ignore-scripts --no-frozen-lockfile \ && printf '%s pnpm-lock.yaml\n' "${PAPERCLIP_RUNNER_LOCK_SHA256}" > /tmp/provider-lock.sha256 \ && sha256sum -c /tmp/provider-lock.sha256 \ diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 4c82b369384..6bb26879725 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -14335,7 +14335,18 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { .set({ nextAttemptAt: null }) .where(eq(chatDeliveries.endpointId, endpoint.id)); - const processing = service.processPendingDeliveries(); + // This suite shares its database. Drain only this fixture's conversation; + // a global recovery pass can renew unrelated retained deliveries with this + // test's failure hook and make its ownership assertion depend on test order. + const [firstDelivery] = await db + .select({ id: chatDeliveries.id }) + .from(chatDeliveries) + .where(and( + eq(chatDeliveries.endpointId, endpoint.id), + eq(chatDeliveries.providerEventId, `${thread.thread.id}:${first.id}`), + )); + if (!firstDelivery) throw new Error("Expected first delivery"); + const processing = service.processPendingDeliveries(25, firstDelivery.id); await firstDeliveryEntered; await renewalAttempted; releaseFirstDelivery(); diff --git a/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts b/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts new file mode 100644 index 00000000000..5660976284c --- /dev/null +++ b/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts @@ -0,0 +1,70 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.js"; + +const execute = vi.hoisted(() => vi.fn()); +vi.mock("../adapters/index.js", async () => { + const actual = await vi.importActual("../adapters/index.js"); + return { ...actual, getServerAdapter: vi.fn(() => ({ supportsLocalAgentJwt: false, execute })) }; +}); +const support = await getEmbeddedPostgresTestSupport(); +const suite = support.supported ? describe : describe.skip; + +suite("native terminal-run cleanup admission", () => { + let db: ReturnType; + let temporary: Awaited>; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-cleanup-admission-"); + db = createDb(temporary.connectionString); + }, 60_000); + afterAll(async () => { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + }); + + it("holds the same task until its native executor settles while allowing unrelated tasks", async () => { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), otherIssueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Cleanup admission", issuePrefix: "CLN", defaultResponsibleUserId: "owner" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", role: "engineer", status: "idle", adapterType: "process", + adapterConfig: {}, permissions: {}, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 20 } } }); + await db.insert(issues).values([issueId, otherIssueId].map(id => ({ id, companyId, title: "Recovery task", status: "todo", assigneeAgentId: agentId, responsibleUserId: "owner" }))); + let release!: () => void, entered!: () => void; + const cleanupGate = new Promise(resolve => { release = resolve; }); + const terminal = new Promise(resolve => { entered = resolve; }); + let first = true; + execute.mockImplementation(async (context: { runId: string; context: { issueId: string } }) => { + if (first) { + first = false; + // Native finalization commits success before workspace/lease cleanup + // finishes. Keep the real heartbeat executor alive across that boundary. + await db.update(heartbeatRuns).set({ status: "succeeded", runtimeMode: "native", finishedAt: new Date() }).where(eq(heartbeatRuns.id, context.runId)); + entered(); + await cleanupGate; + } else { + await db.update(issues).set({ status: "done" }).where(eq(issues.id, context.context.issueId)); + } + return { exitCode: 0, signal: null, timedOut: false, summary: "Completed" }; + }); + const heartbeat = heartbeatService(db); + const nextId = randomUUID(), otherId = randomUUID(); + try { + await heartbeat.invoke(agentId, "on_demand", { issueId }, "manual"); + await terminal; + await db.insert(heartbeatRuns).values([ + { id: nextId, companyId, agentId, status: "queued", invocationSource: "on_demand", responsibleUserId: "owner", contextSnapshot: { issueId } }, + { id: otherId, companyId, agentId, status: "queued", invocationSource: "on_demand", responsibleUserId: "owner", contextSnapshot: { issueId: otherIssueId } }, + ]); + await heartbeat.resumeQueuedRuns(); + expect((await heartbeat.getRun(nextId))?.status).toBe("queued"); + expect((await heartbeat.getRun(otherId))?.status).not.toBe("queued"); + } finally { + release(); + await heartbeat.drainActiveRunExecutions(); + } + expect((await heartbeat.getRun(nextId))?.status).toBe("succeeded"); + expect(execute).toHaveBeenCalledTimes(3); + }, 30_000); +}); diff --git a/server/src/__tests__/native-workspace-sync.test.ts b/server/src/__tests__/native-workspace-sync.test.ts index 508e1737598..3925b49425d 100644 --- a/server/src/__tests__/native-workspace-sync.test.ts +++ b/server/src/__tests__/native-workspace-sync.test.ts @@ -12,6 +12,7 @@ import { nativeWorkspaceSyncInternals, readNativeWorkspaceSyncReference, resumeNativeWorkspaceSync, + prepareNativeWorkspaceSync, } from "../services/native-runtime/native-workspace-sync.js"; const digest = "a".repeat(64); @@ -94,6 +95,20 @@ describe("native workspace sync durable metadata", () => { ).toThrow("native_workspace_sync_unexpected_existing_descriptor"); }); + it("refuses a different sandbox before preparing restart workspace sync", async () => { + const select = vi.fn(); + await expect(prepareNativeWorkspaceSync({ + db: { select } as never, runId: "run", companyId: "company", workspaceId: "workspace", + workspaceLocalDir: "/tmp/local-workspace", lease: { id: "lease", providerLeaseId: "replacement" } as never, + target: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", providerKey: "daytona", + sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "replacement" } } as never, + restartRecovery: { kind: "reattach_remote_runner", runId: "run", leaseOwner: "controller", controllerGeneration: 2, + providerAttempt: 1, restartKind: "graceful", recoveryRequestId: null, + remote: { providerLeaseId: "original", remoteCwd: "/workspace" } }, + })).rejects.toThrow("native_remote_recovery_lease_mismatch"); + expect(select).not.toHaveBeenCalled(); + }); + it("reads backward-compatible references and the resource disposition", () => { const base = { schema: "paperclip.native-workspace-sync/v1", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6fc77e2423e..64565ca229f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -14839,6 +14839,7 @@ export function heartbeatService( const claims = dispositions.filter( (disposition): disposition is NativeRestartRecoveryClaim => disposition.kind === "reattach_existing_runner" || + disposition.kind === "reattach_remote_runner" || disposition.kind === "resume_dead_runner" || disposition.kind === "bootstrap_incomplete", ); @@ -14847,6 +14848,7 @@ export function heartbeatService( if (run) { const isClaim = disposition.kind === "reattach_existing_runner" || + disposition.kind === "reattach_remote_runner" || disposition.kind === "resume_dead_runner" || disposition.kind === "bootstrap_incomplete"; await appendRunEvent(run, { @@ -14854,7 +14856,8 @@ export function heartbeatService( stream: "system", level: disposition.kind === "blocked" ? "warn" : "info", message: - disposition.kind === "reattach_existing_runner" + disposition.kind === "reattach_existing_runner" || + disposition.kind === "reattach_remote_runner" ? "Recovering the existing native runner process after server restart" : disposition.kind === "resume_dead_runner" ? "Resuming the durable native provider session after runner process loss" @@ -17129,6 +17132,23 @@ export function heartbeatService( } const issueId = readNonEmptyString(context.issueId); + if (issueId && activeRunExecutions.size > 0) { + // Native finalization publishes success before workspace synchronization, + // provider suspension, and lease release finish. A queued comment must + // not acquire a fresh sandbox while its predecessor still owns that work. + // The executor's finally block retries this agent after removing its owner. + const [settlingOwner] = await db.select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, run.companyId), + eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.runtimeMode, "native"), + inArray(heartbeatRuns.id, [...activeRunExecutions]), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + )) + .limit(1); + if (settlingOwner) return null; + } if (issueId) { const activePauseHold = await treeControlSvc.getActivePauseHoldGate( run.companyId, @@ -19903,7 +19923,8 @@ export function heartbeatService( if ( runOptions.nativeLeaseOwner && run.runtimeMode === "native" && - runOptions.nativeRestartRecovery?.kind !== "reattach_existing_runner" + runOptions.nativeRestartRecovery?.kind !== "reattach_existing_runner" && + runOptions.nativeRestartRecovery?.kind !== "reattach_remote_runner" ) { // A numeric PID or process-group ID is a liveness signal, never an // ownership capability: the OS may have recycled it after the service diff --git a/server/src/services/native-runtime/native-restart-recovery.test.ts b/server/src/services/native-runtime/native-restart-recovery.test.ts index 542d0585579..ee69e3f494a 100644 --- a/server/src/services/native-runtime/native-restart-recovery.test.ts +++ b/server/src/services/native-runtime/native-restart-recovery.test.ts @@ -9,16 +9,72 @@ import { describe("native restart recovery classification", () => { it("does not reopen a failed checkpoint or reset an exhausted provider budget", () => { - const evidence = { runnerPidAlive: false, runnerGroupAlive: false, processStartMatches: false, hasCheckpoint: true, hasProviderEvidence: true }; - expect(classifyNativeRunnerRecoveryEvidence({ ...evidence, checkpointFailed: true })).toMatchObject({ claimKind: null, reason: "provider_checkpoint_permanently_failed" }); - expect(classifyNativeRunnerRecoveryEvidence({ ...evidence, providerAttempt: 3 })).toMatchObject({ claimKind: null, reason: "execution_recovery_budget_exhausted" }); + const evidence = { + runnerPidAlive: false, + runnerGroupAlive: false, + processStartMatches: false, + hasCheckpoint: true, + hasProviderEvidence: true, + }; + expect( + classifyNativeRunnerRecoveryEvidence({ + ...evidence, + checkpointFailed: true, + }), + ).toMatchObject({ + claimKind: null, + reason: "provider_checkpoint_permanently_failed", + }); + expect( + classifyNativeRunnerRecoveryEvidence({ ...evidence, providerAttempt: 3 }), + ).toMatchObject({ + claimKind: null, + reason: "execution_recovery_budget_exhausted", + }); }); it("keeps controller-only recovery out of the provider retry budget", () => { expect(nextNativeProviderAttempt(2, "reattach_existing_runner")).toBe(2); + expect(nextNativeProviderAttempt(2, "reattach_remote_runner")).toBe(2); expect(nextNativeProviderAttempt(2, "bootstrap_incomplete")).toBe(3); expect(nextNativeProviderAttempt(2, "resume_dead_runner")).toBe(3); }); + it("verifies remote ownership in the sandbox instead of interpreting its PIDs locally", () => { + const evidence = { + remoteSandbox: true, + runnerPidAlive: true, + runnerGroupAlive: true, + processStartMatches: false, + knownProviderProcessAlive: true, + knownProviderProcessIdentityAmbiguous: true, + hasCheckpoint: true, + checkpointIdentityMatches: true, + hasProviderEvidence: true, + providerAttempt: 3, + }; + expect(classifyNativeRunnerRecoveryEvidence(evidence).claimKind).toBe( + "reattach_remote_runner", + ); + expect( + classifyNativeRunnerRecoveryEvidence({ + ...evidence, + checkpointIdentityMatches: false, + }).claimKind, + ).toBeNull(); + expect( + classifyNativeRunnerRecoveryEvidence({ + ...evidence, + checkpointFailed: true, + }).claimKind, + ).toBeNull(); + expect( + classifyNativeRunnerRecoveryEvidence({ + ...evidence, + hasCheckpoint: false, + }).claimKind, + ).toBeNull(); + }); + it.each([ { name: "reattaches an exact live runner", diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 5d5743e9b59..2fc54b97a25 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; import { recordNativeLocalProcessStop } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; import { and, desc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm"; @@ -39,6 +40,16 @@ export type NativeRestartRecoveryClaim = startedAt: string; }; } + | { + kind: "reattach_remote_runner"; + runId: string; + leaseOwner: string; + controllerGeneration: number; + providerAttempt: number; + restartKind: NativeRestartKind; + recoveryRequestId: string | null; + remote: { providerLeaseId: string; remoteCwd: string }; + } | { kind: "resume_dead_runner"; runId: string; @@ -70,7 +81,8 @@ export function nextNativeProviderAttempt( currentAttempt: number, recoveryKind?: NativeRestartRecoveryClaim["kind"], ): number { - return recoveryKind === "reattach_existing_runner" + return recoveryKind === "reattach_existing_runner" || + recoveryKind === "reattach_remote_runner" ? currentAttempt : currentAttempt + 1; } @@ -273,6 +285,7 @@ export async function evaluateNativeProviderProcesses(input: { export function classifyNativeRunnerRecoveryEvidence(input: { runnerPidAlive: boolean; runnerGroupAlive: boolean; + remoteSandbox?: boolean; processStartMatches: boolean; knownProviderProcessAlive?: boolean; knownProviderProcessIdentityAmbiguous?: boolean; @@ -290,6 +303,19 @@ export function classifyNativeRunnerRecoveryEvidence(input: { claimKind: null, reason: "provider_checkpoint_permanently_failed", }; + // Sandbox PIDs belong to another host. This claim only authorizes remote + // inspection; the executor must prove the original lease and authenticate + // the surviving runner before it can resume delivery. + if (input.remoteSandbox) { + return input.hasCheckpoint && + input.checkpointIdentityMatches === true && + input.hasProviderEvidence + ? { + claimKind: "reattach_remote_runner", + reason: "remote_runner_requires_sandbox_verification", + } + : { claimKind: null, reason: "ambiguous_remote_recovery_evidence" }; + } if (input.runnerPidAlive && input.processStartMatches) { return { claimKind: "reattach_existing_runner", @@ -369,7 +395,8 @@ function historyEntry(input: { disposition: input.disposition, reason: input.reason, stateRootAction: - input.disposition === "reattach_existing_runner" + input.disposition === "reattach_existing_runner" || + input.disposition === "reattach_remote_runner" ? "reopen_exact_root" : input.disposition === "resume_dead_runner" ? "reuse_exact_root" @@ -584,8 +611,15 @@ export async function claimNativeRestartRecoveries(input: { } as const; } - const runnerPidAlive = processIsAlive(row.run.processPid); - const runnerGroupAlive = processGroupIsAlive(row.run.processGroupId); + const profile = row.run.runnerProfileJson ?? {}; + const remoteWorkspace = readNativeWorkspaceSyncReference( + profile.nativeWorkspaceSync, + ); + const remoteSandbox = profile.nativeWorkspaceSync != null; + const runnerPidAlive = + !remoteSandbox && processIsAlive(row.run.processPid); + const runnerGroupAlive = + !remoteSandbox && processGroupIsAlive(row.run.processGroupId); const observedRunnerStart = row.run.processPid && runnerPidAlive ? await observedProcessStart(row.run.processPid) @@ -595,7 +629,6 @@ export async function claimNativeRestartRecoveries(input: { row.run.processPid !== null && sameProcessStart(row.run.processStartedAt, observedRunnerStart); - const profile = row.run.runnerProfileJson ?? {}; const checkpoint = profile.sessionCheckpoint; const checkpointRecord = checkpoint && @@ -709,7 +742,7 @@ export async function claimNativeRestartRecoveries(input: { } } const providerProcesses = await evaluateNativeProviderProcesses({ - identities: providerProcessIdentities.filter( + identities: (remoteSandbox ? [] : providerProcessIdentities).filter( (identity) => identity.pid !== row.run.processPid, ), }); @@ -717,6 +750,7 @@ export async function claimNativeRestartRecoveries(input: { hasCheckpointProviderIdentity || providerEvents.length > 0; const classification = classifyNativeRunnerRecoveryEvidence({ + remoteSandbox, runnerPidAlive, runnerGroupAlive, processStartMatches: exactRunnerIdentity, @@ -724,7 +758,9 @@ export async function claimNativeRestartRecoveries(input: { knownProviderProcessIdentityAmbiguous: providerProcesses.ambiguousLivePids.length > 0, hasCheckpoint, - checkpointIdentityMatches, + checkpointIdentityMatches: + checkpointIdentityMatches && + (!remoteSandbox || remoteWorkspace !== null), hasProviderEvidence, checkpointFailed: (checkpointRecord.terminal as Record | undefined) @@ -885,7 +921,8 @@ export async function claimNativeRestartRecoveries(input: { controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: generation, recoveryState: - claimKind === "reattach_existing_runner" + claimKind === "reattach_existing_runner" || + claimKind === "reattach_remote_runner" ? "awaiting_runner_reattach" : claimKind === "resume_dead_runner" ? "resuming_session" @@ -919,7 +956,10 @@ export async function claimNativeRestartRecoveries(input: { } as const; } - if (claimKind !== "reattach_existing_runner") { + if ( + claimKind !== "reattach_existing_runner" && + claimKind !== "reattach_remote_runner" + ) { await recordNativeLocalProcessStop(tx as unknown as Db, row.run); } @@ -932,7 +972,8 @@ export async function claimNativeRestartRecoveries(input: { errorCode: null, nativePhase: "observed", nativePhaseUpdatedAt: now, - ...(claimKind === "reattach_existing_runner" + ...(claimKind === "reattach_existing_runner" || + claimKind === "reattach_remote_runner" ? {} : { processPid: null, @@ -951,6 +992,16 @@ export async function claimNativeRestartRecoveries(input: { restartKind: input.restartKind, recoveryRequestId: input.recoveryRequestId ?? null, }; + if (claimKind === "reattach_remote_runner") { + return { + kind: claimKind, + ...common, + remote: { + providerLeaseId: remoteWorkspace!.providerLeaseId, + remoteCwd: remoteWorkspace!.remoteCwd, + }, + } satisfies NativeRestartRecoveryClaim; + } if (claimKind === "reattach_existing_runner") { return { kind: claimKind, @@ -967,7 +1018,8 @@ export async function claimNativeRestartRecoveries(input: { ...common, } satisfies NativeRestartRecoveryClaim; }); - if (terminalRunToReport) void reportRunFailure(input.db, terminalRunToReport); + if (terminalRunToReport) + void reportRunFailure(input.db, terminalRunToReport); dispositions.push(disposition); } return dispositions; diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 87f2e0c4f86..820661f8893 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -364,6 +364,37 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( .where(eq(nativeRunFinalizations.runId, input.fixture.runId)); } + it.each([null, process.pid])("claims sandbox recovery without interpreting remote PID %s locally", async (remotePid) => { + const fixture = await seedRun(`remote-${remotePid ?? "warm"}`); + await fixture.db.update(heartbeatRuns).set({ + processPid: remotePid, + processStartedAt: new Date("2020-01-01T00:00:00Z"), + runnerProfileJson: { + nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v1", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), + workspaceId: "workspace", leaseId: "lease", providerLeaseId: "sandbox", + remoteCwd: "/workspace", + }, + sessionCheckpoint: { + identity: { companyId, agentId, issueId: fixture.issueId, runId: fixture.runId, + sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "provider-session", + process: { providerPid: process.pid }, + }, + }, + }).where(eq(heartbeatRuns.id, fixture.runId)); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "graceful", runIds: [fixture.runId] }); + expect(claim).toMatchObject({ kind: "reattach_remote_runner", runId: fixture.runId, + providerAttempt: 0, remote: { providerLeaseId: "sandbox", remoteCwd: "/workspace" } }); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator?.recoveryState).toBe("awaiting_runner_reattach"); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run?.status).toBe("running"); + expect(run?.processPid).toBe(remotePid); + }); + realProcessIt("adopts one active runner across hot and hard controller restarts without duplicating steering", async () => { const fixture = await seedRun("LIVE"); const stateDirectory = resolve(runtimeRoot, fixture.runId); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index bbf29891101..26f73b34013 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -73,6 +73,7 @@ type BackendFactoryOptions = { }; type RunnerTransportOptions = { + adoptExistingRunner?: { pid: number; isAlive: () => Promise | boolean }; stateDirectory?: string; runnerBinary?: string; prpIdentity?: { @@ -269,6 +270,8 @@ import { nativeUsageCostUsd, normalizeNativeUsage, parseRemoteRunnerProcessIdentity, + REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT, + verifyRemoteRunnerReattachment, readRemoteProviderPackManifest, providerSessionIdentityFromDurableProviderState, providerSessionIdentityTransitionIsAllowed, @@ -295,6 +298,257 @@ beforeEach(() => { state.assertCurrentWakeCommentsRead.mockReset().mockResolvedValue(undefined); }); +describe("remote controller restart adoption", () => { + const identity = { + runId: "run", + normalizedSessionId: "session", + runnerInstanceId: "runner", + environmentLeaseId: "lease", + turnId: "turn", + itemId: "item", + }; + const claim = { + kind: "reattach_remote_runner" as const, + runId: "run", + leaseOwner: "controller", + controllerGeneration: 2, + providerAttempt: 1, + restartKind: "graceful" as const, + recoveryRequestId: null, + remote: { providerLeaseId: "sandbox", remoteCwd: "/workspace" }, + }; + function fixture(overrides: Record = {}) { + const execute = vi.fn(async (request: { args: string[] }) => { + const script = request.args[1]; + const stdout = script.includes("base64") + ? Buffer.from( + JSON.stringify({ ...identity, lifecycle: "running", ...overrides }), + ).toString("base64") + : script.includes("cat --") + ? "nonce\n123\n2026-09-19T10:00:00.000Z\nrunner\nlinux:abcd-1234:100\n" + : ""; + return { stdout, stderr: "", exitCode: 0, timedOut: false }; + }); + const target = { + kind: "remote" as const, + transport: "sandbox" as const, + remoteCwd: "/workspace", + providerKey: "daytona", + leaseId: "lease", + sandboxLeaseAcquisition: { + outcome: "resumed" as const, + providerLeaseId: "sandbox", + }, + runner: { execute }, + }; + return { execute, target }; + } + it("adopts the exact live sandbox process without spawning or copying state", async () => { + const { execute, target } = fixture(); + const adopted = await verifyRemoteRunnerReattachment({ + claim, + target: target as never, + identity, + runId: "run", + normalizedSessionId: "session", + }); + expect(adopted.pid).toBe(123); + expect(await adopted.isAlive()).toBe(true); + expect( + execute.mock.calls.every( + ([request]) => !request.args.join(" ").match(/nohup|mkdir|rm -|mv -/), + ), + ).toBe(true); + execute.mockResolvedValueOnce({ + stdout: "", + stderr: "", + exitCode: 4, + timedOut: false, + }); + expect(await adopted.isAlive()).toBe(false); + }); + it.each([ + "runId", + "normalizedSessionId", + "runnerInstanceId", + "environmentLeaseId", + "turnId", + "itemId", + ])("rejects a changed remote %s", async (field) => { + const { execute, target } = fixture({ [field]: "different" }); + await expect( + verifyRemoteRunnerReattachment({ + claim, + target: target as never, + identity, + runId: "run", + normalizedSessionId: "session", + }), + ).rejects.toThrow("runner_remote_recovery_identity_mismatch"); + expect(execute).toHaveBeenCalledTimes(1); + }); + it("wires remote adoption when only controller state survived on the host", async () => { + const stateBase = await mkdtemp( + join(tmpdir(), "paperclip-remote-reattach-"), + ); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase; + const current = { + ...execution, + binding: { ...execution.binding, runId: "run" }, + session: { ...execution.session, normalizedSessionId: "session" }, + }; + const { target } = fixture(); + try { + state.createBackend.mockClear(); + state.createTransport.mockClear(); + await createRunnerdBackend({ + db: leaseDb(current), + execution: current, + runnerInstanceId: "runner", + runnerExecutionTarget: target as never, + }); + state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + const root = state.createTransport.mock.calls.at(-1)![0].stateDirectory!; + await mkdir(join(root, "control-plane"), { recursive: true }); + await writeFile( + join(root, "control-plane", "control-plane-state.json"), + JSON.stringify(durableControlPlaneState(identity)), + ); + await createRunnerdBackend({ + db: leaseDb(current), + execution: current, + runnerInstanceId: "runner", + runnerExecutionTarget: target as never, + restartRecovery: claim, + }); + state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + expect( + state.createTransport.mock.calls.at(-1)![0].adoptExistingRunner?.pid, + ).toBe(123); + expect(await readdir(root)).toContain("control-plane"); + expect(await readdir(root)).not.toContain("runner"); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(stateBase, { recursive: true, force: true }); + } + }); + it("rejects a replaced sandbox before reading or mutating it", async () => { + const { execute, target } = fixture(); + target.sandboxLeaseAcquisition.providerLeaseId = "replacement"; + await expect( + verifyRemoteRunnerReattachment({ + claim, + target: target as never, + identity, + runId: "run", + normalizedSessionId: "session", + }), + ).rejects.toThrow("native_remote_recovery_lease_mismatch"); + expect(execute).not.toHaveBeenCalled(); + }); + it("rejects a marker that cannot prove its process generation", async () => { + const { target, execute } = fixture(); + const original = execute.getMockImplementation()!; + execute.mockImplementation(async request => request.args[2] === "paperclip-runner-recovery-identity" + ? { stdout: "nonce\n123\n2026-09-19T10:00:00.000Z\nrunner\n", stderr: "", exitCode: 0, timedOut: false } + : original(request)); + await expect(verifyRemoteRunnerReattachment({ claim, target: target as never, identity, runId: "run", normalizedSessionId: "session" })) + .rejects.toThrow("runner_remote_process_identity_unavailable"); + }); + it.each(["start ticks", "boot identity"])("rejects PID reuse after the live %s changes", async (change) => { + const root = await mkdtemp(join(tmpdir(), "remote-generation-")); + const proc = join(root, "proc"); + const markerPath = join(root, "runner-process.identity"); + const pid = process.pid; + const marker = `nonce\n${pid}\n2026-09-19T10:00:00.000Z\nrunner\nlinux:abcd-1234:100\n`; + const stat = (ticks: number) => `${pid} (runner (child)) ${["S", ...Array(18).fill("0"), ticks].join(" ")}\n`; + const { target, execute } = fixture(); + const original = execute.getMockImplementation()!; + try { + await mkdir(join(proc, String(pid)), { recursive: true }); + await mkdir(join(proc, "sys/kernel/random"), { recursive: true }); + await writeFile(join(proc, String(pid), "stat"), stat(100)); + await writeFile(join(proc, String(pid), "cmdline"), "runner\0--runner-id\0runner\0"); + await writeFile(join(proc, "sys/kernel/random/boot_id"), "abcd-1234\n"); + await writeFile(markerPath, marker); + execute.mockImplementation(async request => { + if (request.args[2] === "paperclip-runner-recovery-identity") { + return { stdout: marker, stderr: "", exitCode: 0, timedOut: false }; + } + if (request.args[2] !== "paperclip-runner-recovery-check") return original(request); + // Run the actual ownership shell against controlled Linux proc files. + // kill -0 still checks a real live PID; only its generation changes. + const args = [...request.args]; + args[1] = args[1].replaceAll("/proc/", `${proc}/`); + args[3] = markerPath; + let exitCode = 0; + try { execFileSync("sh", args, { stdio: "pipe" }); } catch { exitCode = 4; } + return { stdout: "", stderr: "", exitCode, timedOut: false }; + }); + const adopted = await verifyRemoteRunnerReattachment({ claim, target: target as never, identity, runId: "run", normalizedSessionId: "session" }); + expect(await adopted.isAlive()).toBe(true); + if (change === "start ticks") await writeFile(join(proc, String(pid), "stat"), stat(101)); + else await writeFile(join(proc, "sys/kernel/random/boot_id"), "abcd-5678\n"); + expect(await adopted.isAlive()).toBe(false); + await expect(verifyRemoteRunnerReattachment({ claim, target: target as never, identity, runId: "run", normalizedSessionId: "session" })) + .rejects.toThrow("runner_remote_process_identity_unavailable"); + } finally { await rm(root, { recursive: true, force: true }); } + }); + it("rejects a stale process marker", async () => { + const { execute, target } = fixture(); + execute + .mockResolvedValueOnce({ + stdout: Buffer.from(JSON.stringify(identity)).toString("base64"), + stderr: "", + exitCode: 0, + timedOut: false, + }) + .mockResolvedValueOnce({ + stdout: "nonce\n123\n2026-09-19T10:00:00Z\nother-runner\n", + stderr: "", + exitCode: 0, + timedOut: false, + }); + await expect( + verifyRemoteRunnerReattachment({ + claim, + target: target as never, + identity, + runId: "run", + normalizedSessionId: "session", + }), + ).rejects.toThrow("runner_remote_process_identity_unavailable"); + }); +}); + + +describe("remote runner launch fingerprint compatibility", () => { + it.each([true, false])("launches with the guaranteed Node runtime when boot identity is available: %s", async (hasBootIdentity) => { + const root = await mkdtemp(join(tmpdir(), "remote-launch-generation-")); + try { + // The mocked proc stat is independent of the launch shell's PID. + await mkdir(join(root, "sys/kernel/random"), { recursive: true }); + await writeFile(join(root, "stat"), `123 (runner) ${["S", ...Array(18).fill("0"), 1234].join(" ")}\n`); + if (hasBootIdentity) await writeFile(join(root, "sys/kernel/random/boot_id"), "abcd-1234\n"); + const script = REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT + .replaceAll('"/proc/"+process.argv[1]+"/stat"', JSON.stringify(`${root}/stat`)) + .replaceAll("/proc/sys/kernel/random/boot_id", `${root}/sys/kernel/random/boot_id`); + // Node is part of the remote runtime contract; awk is not required. + const bin = join(root, "bin"); + await mkdir(bin); + await symlink(process.execPath, join(bin, "node")); + for (const name of ["mkdir", "chmod", "date", "mv"]) await symlink(execFileSync("sh", ["-c", `command -v ${name}`], { encoding: "utf8" }).trim(), join(bin, name)); + const marker = join(root, "identity"); + const result = execFileSync("/bin/sh", ["-c", script, "launch", marker, "nonce", "runner", join(root, "diagnostics"), "/bin/sh", "-c", "printf launched"], { env: { ...process.env, PATH: bin }, encoding: "utf8" }); + expect(result).toBe("launched"); + const lines = (await readFile(marker, "utf8")).split("\n"); + expect(lines[4]).toBe(hasBootIdentity ? "linux:abcd-1234:1234" : ""); + } finally { await rm(root, { recursive: true, force: true }); } + }); +}); + describe("remote runner process supervision", () => { it.each(["delivered", "sandbox_missing", "logging_failed"] as const)( "detaches runnerd and contains asynchronous signal failures (%s)", async (signalOutcome) => { @@ -4412,6 +4666,89 @@ function cancellationDb(options?: { } describe("native startup restart detachment", () => { + it("waits for in-flight runner startup and its detach acknowledgement before shutdown returns", async () => { + const root = await mkdtemp(join(tmpdir(), "native-startup-detach-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = root; + const restarting = structuredClone(execution); + restarting.binding.runId = "restart-inflight-bootstrap"; + restarting.session.normalizedSessionId = "restart-inflight-session"; + let open!: () => void, started!: () => void, acknowledge!: () => void; + const opening = new Promise(resolve => { open = resolve; }); + const admitted = new Promise(resolve => { started = resolve; }); + const acknowledgement = new Promise(resolve => { acknowledge = resolve; }); + const detached = vi.fn(); + const detach = vi.fn(async () => { await acknowledgement; }); + state.execute.mockReset().mockImplementationOnce(async (options) => { + started(); + await opening; + await options.onSession({ detachControllerForRestart: detach }); + await options.onSession(null); + throw new Error("detachment closed the old event stream"); + }); + const running = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner", useRunnerd: true }); + const outcome = running.catch(error => error); + let detaching: Promise | undefined; + try { + await admitted; + detaching = detachNativeSessionsForRestart([restarting.binding.runId]).then(detached); + await new Promise(resolve => setImmediate(resolve)); + expect(detached).not.toHaveBeenCalled(); + open(); + await vi.waitFor(() => expect(detach).toHaveBeenCalledOnce()); + expect(detached).not.toHaveBeenCalled(); + acknowledge(); + await detaching; + expect(detached).toHaveBeenCalledWith(expect.objectContaining({ detachedRunIds: [restarting.binding.runId] })); + expect(await outcome).toBeInstanceOf(NativeControllerDetachedForRestartError); + } finally { + open(); acknowledge(); + await outcome; + await detaching; + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(root, { recursive: true, force: true }); + } + }); + + it.each([false, true])("settles failed startup without claiming detachment (deadline exceeded: %s)", async (exceedDeadline) => { + const root = await mkdtemp(join(tmpdir(), "native-startup-failure-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = root; + const restarting = structuredClone(execution); + restarting.binding.runId = `restart-failed-bootstrap-${exceedDeadline}`; + restarting.session.normalizedSessionId = `restart-failed-session-${exceedDeadline}`; + let release!: () => void, started!: () => void; + const opening = new Promise(resolve => { release = resolve; }); + const admitted = new Promise(resolve => { started = resolve; }); + state.execute.mockReset().mockImplementationOnce(async () => { + started(); + await opening; + throw new Error("bootstrap failed before session publication"); + }); + const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner", useRunnerd: true }).catch(error => error); + try { + await admitted; + if (exceedDeadline) vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const detached = detachNativeSessionsForRestart([restarting.binding.runId]).catch(error => error); + if (exceedDeadline) { + await vi.advanceTimersByTimeAsync(30_000); + expect(await detached).toMatchObject({ message: "native_restart_startup_not_ready" }); + vi.useRealTimers(); + } else { + release(); + expect(await detached).toMatchObject({ detachedRunIds: [], inactiveRunIds: [restarting.binding.runId] }); + } + } finally { + vi.useRealTimers(); + release(); + await outcome; + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(root, { recursive: true, force: true }); + } + }); + it("remembers shutdown while the session is still opening and detaches its late publication", async () => { const restarting = structuredClone(execution); restarting.binding.runId = "restart-during-session-open"; @@ -5273,6 +5610,86 @@ describe("native warm session supervision", () => { }); }); + it.each(["checkpoint first", "turn first"])("serializes restart checkpoint and turn admission: %s", async (order) => { + const stateBase = await mkdtemp(join(tmpdir(), "paperclip-close-race-")); + const previousPaperclipHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = stateBase; + let finishClose!: () => void; + const closing = new Promise((resolve) => { finishClose = resolve; }); + const close = vi.fn(() => closing); + try { + const first = { + ...execution, + binding: { + ...execution.binding, + runId: "run-close-race-first", + executionWorkspaceId: "workspace-close-race", + }, + session: { + ...execution.session, + normalizedSessionId: "session-close-race", + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, + }, + } as NativeExecutionInputV1; + const result = { + result: { summary: "completed" }, + terminal: { runTerminalState: "succeeded" }, + turnId: "turn-close-race", + normalizedSessionId: first.session.normalizedSessionId, + providerSessionId: "provider-close-race", + driverKind: "test", + driverVersion: "1", + nativeEventCount: 1, + highestContiguousSourceSeq: 1, + usage: null, + }; + state.execute.mockReset().mockImplementationOnce(async (options) => { + await options.onSession?.({ close }); + return result; + }); + await executePaperclipNativeSession({ + db: leaseDb(first), execution: first, + runnerInstanceId: "runner-close-race", useRunnerd: true, + }); + const second = { ...first, binding: { ...first.binding, runId: "run-close-race-second" } }; + const reachedAdmission = new Error("test reached post-checkpoint admission"); + state.stageNativeRunnerWakeAttachments.mockClear(); + state.stageNativeRunnerWakeAttachments.mockImplementationOnce(async () => { + if (order === "turn first") { + await expect(closeIdleWarmNativeSessionsForRestart()).resolves.toMatchObject({ closed: 0, busy: 1 }); + expect(close).not.toHaveBeenCalled(); + } + throw reachedAdmission; + }); + const shutdown = order === "checkpoint first" ? closeIdleWarmNativeSessionsForRestart() : undefined; + if (shutdown) await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + const continuation = executePaperclipNativeSession({ + db: leaseDb(second), execution: second, + runnerInstanceId: "runner-close-race", useRunnerd: true, + }).catch((error) => error); + try { + if (order === "checkpoint first") { + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(state.stageNativeRunnerWakeAttachments).not.toHaveBeenCalled(); + } else { + expect(await continuation).toBe(reachedAdmission); + } + expect(state.execute).toHaveBeenCalledOnce(); + } finally { + finishClose(); + await shutdown; + expect(await continuation).toBe(reachedAdmission); + } + expect(state.stageNativeRunnerWakeAttachments).toHaveBeenCalledOnce(); + } finally { + finishClose(); + await closeIdleWarmNativeSessionsForRestart(); + if (previousPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = previousPaperclipHome; + await rm(stateBase, { recursive: true, force: true }); + } + }); + it.each([false, true])("checkpoints a busy warm session on release after the restart sweep: checkpoint fails=%s", async (checkpointFails) => { const checkpointError = new Error("restart checkpoint failed"); let finishCheckpoint!: () => void; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 549ea9f4022..8372e92787e 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -173,6 +173,7 @@ import { type ActiveNativeSession = { session: NativeSession; cancelRequested: boolean; + restartDetach?: Promise; }; class NativeResultPendingFinalizationError extends Error { @@ -202,6 +203,36 @@ const activeNativeSessions = new Map(); // the late publication detaches before it can dispatch another turn. const nativeRunsDetachingForRestart = new Set(); +// A restart must not cut authority rotation between archiving the previous +// runner state and authenticating its replacement. Keep the startup owner +// alive until onSession can detach it, or until startup itself has settled. +type NativeSessionStartup = { + promise: Promise; + resolve: (session: ActiveNativeSession | null) => void; +}; +const nativeSessionStartups = new Map(); + +function detachActiveNativeSessionForRestart(active: ActiveNativeSession) { + active.restartDetach ??= Promise.resolve().then(() => active.session.detachControllerForRestart!()); + return active.restartDetach; +} + +async function waitForNativeStartupForRestart(runId: string) { + const startup = nativeSessionStartups.get(runId); + if (!startup) return null; + let timer: ReturnType | undefined; + try { + return await Promise.race([ + startup.promise, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error("native_restart_startup_not_ready")), 30_000); + }), + ]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + export async function detachNativeSessionsForRestart( runIds: readonly string[], ): Promise<{ @@ -214,7 +245,7 @@ export async function detachNativeSessionsForRestart( const unsupportedRunIds: string[] = []; for (const runId of new Set(runIds)) { nativeRunsDetachingForRestart.add(runId); - const active = activeNativeSessions.get(runId); + const active = activeNativeSessions.get(runId) ?? await waitForNativeStartupForRestart(runId); if (!active) { inactiveRunIds.push(runId); continue; @@ -223,7 +254,7 @@ export async function detachNativeSessionsForRestart( unsupportedRunIds.push(runId); continue; } - await active.session.detachControllerForRestart(); + await detachActiveNativeSessionForRestart(active); detachedRunIds.push(runId); } return { detachedRunIds, inactiveRunIds, unsupportedRunIds }; @@ -353,6 +384,9 @@ type WarmNativeSession = { }; const warmNativeSessions = new Map(); +// Closing a remote owner saves its checkpoint asynchronously. A new turn must +// not inspect or quarantine that owner's state until the save has finished. +const closingWarmNativeSessions = new Map>(); /** * Close idle native sessions before an operator destroys their remote @@ -391,7 +425,7 @@ async function closeIdleWarmNativeSessions(input: { if (input.environmentId !== undefined && entry.environmentId !== input.environmentId) { continue; } - if (entry.busy) { + if (entry.busy || executingRunnerdSessionScopes.has(sessionId)) { // A busy turn can complete while another idle session is checkpointing. // Fence that entry now so its eventual release cannot leave a new idle // owner behind after the shutdown sweep has already passed it. @@ -403,11 +437,19 @@ async function closeIdleWarmNativeSessions(input: { // Remove ownership before awaiting close so a racing continuation cannot // adopt a session whose transport is already shutting down. warmNativeSessions.delete(sessionId); + const closing = Promise.resolve().then(() => + entry.session.close({ reason: input.reason }), + ); + closingWarmNativeSessions.set(sessionId, closing); try { - await entry.session.close({ reason: input.reason }); + await closing; closed += 1; } catch { failed += 1; + } finally { + if (closingWarmNativeSessions.get(sessionId) === closing) { + closingWarmNativeSessions.delete(sessionId); + } } } return { closed, busy, failed }; @@ -4343,6 +4385,9 @@ async function migrateRunnerdStateRootForExecution(input: { ) { await recoverQuiescentRunnerdState({ ...input, scoped }); } + if (input.restartRecovery?.kind === "reattach_remote_runner" && !existsSync(scoped)) { + throw new Error("runner_state_identity_mismatch"); + } if (existsSync(scoped)) { if (!isSafeNativeStateDirectory(scoped)) { throw new Error("runner_state_directory_unsafe"); @@ -4357,13 +4402,15 @@ async function migrateRunnerdStateRootForExecution(input: { await assertCleanupActivationCommitted(input.db, scoped, input.execution); const identity = readRunnerdDurableIdentity(scoped); if (!identity) { - if (input.restartRecovery?.kind !== "reattach_existing_runner") { + if (input.restartRecovery?.kind !== "reattach_existing_runner" && + input.restartRecovery?.kind !== "reattach_remote_runner") { quarantineRunnerdStateRoot(scoped, "identity_indeterminate"); } throw new Error("runner_state_identity_mismatch"); } if (!durableIdentityMatchesSession(identity, input.execution)) { - if (input.restartRecovery?.kind !== "reattach_existing_runner") { + if (input.restartRecovery?.kind !== "reattach_existing_runner" && + input.restartRecovery?.kind !== "reattach_remote_runner") { quarantineRunnerdStateRoot(scoped, "identity_mismatch"); } throw new Error("runner_state_identity_mismatch"); @@ -4378,6 +4425,13 @@ async function migrateRunnerdStateRootForExecution(input: { // root can authorize a fresh bootstrap; anything else stays fail-closed. throw new Error("runner_state_identity_mismatch"); } + if (input.restartRecovery?.kind === "reattach_remote_runner") { + await verifyRemoteRunnerReattachment({ + claim: input.restartRecovery, target: input.runnerExecutionTarget, identity, + runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution), + }); + return; + } if (durableIdentityMatchesExecution(identity, input.execution)) { if ( runnerdAuthorityLifecycleWithVerifiedBackup({ @@ -6911,6 +6965,7 @@ export async function executePaperclipNativeSession(input: { let sessionScopeId: string | null = null; let ownsSessionScope = false; let executionFailure: unknown; + let startup: NativeSessionStartup | undefined; try { // The session scope is unaffected by appending server-staged attachment // descriptors. Claim it before any workspace scrub/write so a duplicate @@ -6924,6 +6979,15 @@ export async function executePaperclipNativeSession(input: { input.execution.binding.runId, ); ownsSessionScope = true; + let resolveStartup!: (session: ActiveNativeSession | null) => void; + const startupPromise = new Promise(resolve => { resolveStartup = resolve; }); + startup = { promise: startupPromise, resolve: resolveStartup }; + nativeSessionStartups.set(input.execution.binding.runId, startup); + + // The shutdown sweep can have removed an idle owner while its remote + // checkpoint is still being saved. The scope reservation also prevents + // a later sweep from closing an owner this turn is about to acquire. + await closingWarmNativeSessions.get(sessionScopeId); const targetKind = input.runnerExecutionTarget?.kind ?? "local"; const chatAttachmentReadScope = new NativeChatAttachmentReadScope({ @@ -6974,6 +7038,10 @@ export async function executePaperclipNativeSession(input: { executionFailure = error; throw error; } finally { + startup?.resolve(null); + if (startup && nativeSessionStartups.get(input.execution.binding.runId) === startup) { + nativeSessionStartups.delete(input.execution.binding.runId); + } if ( ownsSessionScope && sessionScopeId !== null && @@ -8081,14 +8149,13 @@ async function executePaperclipNativeSessionWithinScope( } } if (session) { - activeNativeSessions.set(input.execution.binding.runId, { - session, - cancelRequested: false, - }); + const active: ActiveNativeSession = { session, cancelRequested: false }; + activeNativeSessions.set(input.execution.binding.runId, active); if (session.resolveRuntimeRequest) await liveQuestions.attach(); if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { - await session.detachControllerForRestart?.(); + if (session.detachControllerForRestart) await detachActiveNativeSessionForRestart(active); } + nativeSessionStartups.get(input.execution.binding.runId)?.resolve(active); } else { liveQuestions.close(); activeNativeSessions.delete(input.execution.binding.runId); @@ -9487,7 +9554,9 @@ async function readRemoteRunnerState(input: { timeoutMs: 10_000, }); if (result.exitCode !== 0 || result.timedOut) { - throw new Error("runner_remote_state_unavailable"); + throw new Error( + `runner_remote_state_unavailable: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`, + ); } return record( JSON.parse( @@ -9525,11 +9594,18 @@ async function readRemoteRunnerProviderState(input: { const REMOTE_RUNNER_PROCESS_IDENTITY_WAIT_MS = 20_000; const REMOTE_RUNNER_PROCESS_POLL_MS = 1_000; +// Linux boot identity plus start ticks identifies a process generation across +// PID reuse. exec preserves both the launch shell's PID and its start ticks. +const REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT = + "process_fingerprint=$(node -e 'const fs=require(\"node:fs\");try{const stat=fs.readFileSync(\"/proc/\"+process.argv[1]+\"/stat\",\"utf8\");const end=stat.lastIndexOf(\") \");const ticks=stat.slice(end+2).trim().split(/\\s+/)[19];const boot=fs.readFileSync(\"/proc/sys/kernel/random/boot_id\",\"utf8\").trim();if(end<0||!/^\\d+$/.test(ticks||\"\")||!/^[-a-fA-F0-9]+$/.test(boot))process.exit(4);process.stdout.write(\"linux:\"+boot+\":\"+ticks)}catch{process.exit(4)}' \"$pid\")"; + const REMOTE_RUNNER_IDENTITY_CHECK_SCRIPT = 'set -eu; identity_path=$1; expected_nonce=$2; expected_runner_id=$3; expected_pid=$4; test -f "$identity_path" && test ! -L "$identity_path" || exit 3; { IFS= read -r nonce; IFS= read -r pid; IFS= read -r started_at; IFS= read -r runner_id; } < "$identity_path"; test "$nonce" = "$expected_nonce" && test "$runner_id" = "$expected_runner_id" && test "$pid" = "$expected_pid" && test -n "$started_at" || exit 4; kill -0 "$pid" 2>/dev/null || exit 3; if test -r "/proc/$pid/cmdline"; then command_line=$(tr "\\000" "\\n" < "/proc/$pid/cmdline"); printf "%s\\n" "$command_line" | grep -Fqx -- "--runner-id" || exit 4; printf "%s\\n" "$command_line" | grep -Fqx -- "$expected_runner_id" || exit 4; fi'; -const REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT = - 'set -eu; identity_path=$1; identity_nonce=$2; runner_instance_id=$3; diagnostics_directory=$4; shift 4; umask 077; test ! -L "$diagnostics_directory"; if test -e "$diagnostics_directory"; then test -d "$diagnostics_directory"; else mkdir -p -- "$diagnostics_directory"; fi; chmod 0700 "$diagnostics_directory"; started_at=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ"); identity_tmp="${identity_path}.tmp.$$"; printf "%s\\n%s\\n%s\\n%s\\n" "$identity_nonce" "$$" "$started_at" "$runner_instance_id" > "$identity_tmp"; chmod 0600 "$identity_tmp"; mv -f -- "$identity_tmp" "$identity_path"; exec "$@"'; +export const REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT = + 'set -eu; identity_path=$1; identity_nonce=$2; runner_instance_id=$3; diagnostics_directory=$4; shift 4; umask 077; test ! -L "$diagnostics_directory"; if test -e "$diagnostics_directory"; then test -d "$diagnostics_directory"; else mkdir -p -- "$diagnostics_directory"; fi; chmod 0700 "$diagnostics_directory"; started_at=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ"); pid=$$; process_fingerprint=""; if ' + + REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT + + '; then :; else process_fingerprint=""; fi; identity_tmp="${identity_path}.tmp.$$"; printf "%s\\n%s\\n%s\\n%s\\n%s\\n" "$identity_nonce" "$$" "$started_at" "$runner_instance_id" "$process_fingerprint" > "$identity_tmp"; chmod 0600 "$identity_tmp"; mv -f -- "$identity_tmp" "$identity_path"; exec "$@"'; const REMOTE_RUNNER_FAILED_IDENTITY_CLEANUP_SCRIPT = 'set -eu; identity_path=$1; expected_nonce=$2; expected_runner_id=$3; marker_wait=0; while { test ! -f "$identity_path" || test -L "$identity_path"; } && test "$marker_wait" -lt 50; do marker_wait=$((marker_wait + 1)); sleep 0.1; done; test -f "$identity_path" && test ! -L "$identity_path" || exit 3; { IFS= read -r nonce; IFS= read -r pid; IFS= read -r started_at; IFS= read -r runner_id; } < "$identity_path"; test "$nonce" = "$expected_nonce" && test "$runner_id" = "$expected_runner_id" && test -n "$started_at" || exit 4; case "$pid" in ""|*[!0-9]*) exit 4 ;; esac; test "$pid" -gt 0 || exit 4; if kill -0 "$pid" 2>/dev/null; then if test -r "/proc/$pid/cmdline"; then command_line=$(tr "\\000" "\\n" < "/proc/$pid/cmdline"); printf "%s\\n" "$command_line" | grep -Fqx -- "--runner-id" || exit 4; printf "%s\\n" "$command_line" | grep -Fqx -- "$expected_runner_id" || exit 4; fi; signal_target=$pid; if command -v ps >/dev/null 2>&1; then session_id=$(ps -o sid= -p "$pid" 2>/dev/null | tr -d " ") || true; if test "$session_id" = "$pid"; then signal_target="-$pid"; fi; fi; kill -TERM -- "$signal_target" 2>/dev/null || kill -TERM "$pid" 2>/dev/null || true; term_wait=0; while kill -0 "$pid" 2>/dev/null && test "$term_wait" -lt 50; do term_wait=$((term_wait + 1)); sleep 0.1; done; if kill -0 "$pid" 2>/dev/null; then kill -KILL -- "$signal_target" 2>/dev/null || kill -KILL "$pid" 2>/dev/null || true; kill_wait=0; while kill -0 "$pid" 2>/dev/null && test "$kill_wait" -lt 50; do kill_wait=$((kill_wait + 1)); sleep 0.1; done; fi; kill -0 "$pid" 2>/dev/null && exit 5; fi; test -f "$identity_path" && test ! -L "$identity_path" || exit 4; { IFS= read -r final_nonce; IFS= read -r final_pid; IFS= read -r final_started_at; IFS= read -r final_runner_id; } < "$identity_path"; test "$final_nonce" = "$nonce" && test "$final_pid" = "$pid" && test "$final_started_at" = "$started_at" && test "$final_runner_id" = "$runner_id" || exit 4; rm -f -- "$identity_path"'; @@ -9537,8 +9613,8 @@ const REMOTE_RUNNER_FAILED_IDENTITY_CLEANUP_SCRIPT = export function parseRemoteRunnerProcessIdentity( value: string, expected: { nonce: string; runnerInstanceId: string }, -): { pid: number; startedAt: string } | null { - const [nonce, rawPid, startedAt, runnerInstanceId, ...remainder] = value +): { pid: number; startedAt: string; processStartFingerprint?: string } | null { + const [nonce, rawPid, startedAt, runnerInstanceId, processStartFingerprint, ...remainder] = value .trim() .split("\n"); const pid = Number(rawPid); @@ -9549,11 +9625,120 @@ export function parseRemoteRunnerProcessIdentity( !Number.isSafeInteger(pid) || pid <= 0 || !startedAt || - Number.isNaN(new Date(startedAt).getTime()) + Number.isNaN(new Date(startedAt).getTime()) || + (processStartFingerprint !== undefined && !/^linux:[0-9a-f-]+:[0-9]+$/.test(processStartFingerprint)) ) { return null; } - return { pid, startedAt }; + return { pid, startedAt, ...(processStartFingerprint ? { processStartFingerprint } : {}) }; +} + +/** Verify remote ownership before exposing the transport's authenticated adoption path. */ +export async function verifyRemoteRunnerReattachment(input: { + claim: Extract< + NativeRestartRecoveryClaim, + { kind: "reattach_remote_runner" } + >; + target: AdapterExecutionTarget | null | undefined; + identity: Record; + runId: string; + normalizedSessionId: string; +}) { + const { claim, target, identity } = input; + if ( + target?.kind !== "remote" || + target.transport !== "sandbox" || + !target.runner || + claim.runId !== input.runId || + identity.runId !== input.runId || + identity.normalizedSessionId !== input.normalizedSessionId || + claim.remote.providerLeaseId !== + target.sandboxLeaseAcquisition?.providerLeaseId || + target.sandboxLeaseAcquisition.outcome === "replacement" || + claim.remote.remoteCwd !== target.remoteCwd + ) { + throw new Error("native_remote_recovery_lease_mismatch"); + } + const runner = target.runner; + const stateDirectory = posix.join( + target.remoteCwd, + ".paperclip-runtime", + "paperclip-runner", + "sessions", + createHash("sha256").update(input.normalizedSessionId).digest("hex"), + "runner", + ); + const runnerState = await readRemoteRunnerState({ runner, stateDirectory }); + for (const field of [ + "runId", + "normalizedSessionId", + "runnerInstanceId", + "environmentLeaseId", + "turnId", + "itemId", + ]) { + if ( + typeof identity[field] !== "string" || + !identity[field] || + runnerState[field] !== identity[field] + ) { + throw new Error("runner_remote_recovery_identity_mismatch"); + } + } + const identityPath = posix.join(stateDirectory, "runner-process.identity"); + const marker = await runner.execute({ + command: "sh", + args: [ + "-c", + 'test -f "$1" && test ! -L "$1" && cat -- "$1"', + "paperclip-runner-recovery-identity", + identityPath, + ], + bypassSession: true, + timeoutMs: 10_000, + }); + const nonce = marker.stdout.split("\n")[0] ?? ""; + const runnerInstanceId = identity.runnerInstanceId as string; + const processIdentity = + marker.exitCode === 0 && !marker.timedOut && /^[a-zA-Z0-9_-]+$/.test(nonce) + ? parseRemoteRunnerProcessIdentity(marker.stdout, { + nonce, + runnerInstanceId, + }) + : null; + if (!processIdentity?.processStartFingerprint) + throw new Error("runner_remote_process_identity_unavailable"); + const expectedStartFingerprint = processIdentity.processStartFingerprint; + const check = async (signal?: NodeJS.Signals) => { + const result = await runner.execute({ + command: "sh", + args: [ + "-c", + `${REMOTE_RUNNER_IDENTITY_CHECK_SCRIPT}; ${REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT}; test "$process_fingerprint" = "$5" || exit 4` + + (signal ? '; kill -"$6" "$pid"' : ""), + "paperclip-runner-recovery-check", + identityPath, + nonce, + runnerInstanceId, + String(processIdentity.pid), + expectedStartFingerprint, + ...(signal ? [signal.slice(3)] : []), + ], + bypassSession: true, + timeoutMs: 10_000, + }); + return result.exitCode === 0 && !result.timedOut; + }; + if (!(await check())) + throw new Error("runner_remote_process_identity_unavailable"); + // These checks authorize an attempt to reconnect, not the task itself. The + // transport still requires authentication against the exact durable PRP key. + return { + ...processIdentity, + processGroupId: null, + isAlive: () => check(), + signal: check, + }; } async function waitForRemoteRunnerProcessIdentity(input: { @@ -10398,6 +10583,15 @@ async function createRunnerdBackendWithinSessionClaim( ) return; selectedRemoteMode = requiredMode; + if (input.restartRecovery?.kind === "reattach_remote_runner") { + await verifyRemoteRunner(requiredMode); + if (requiresRemoteProviderPack && stagedRemoteProviderPackRoot) { + await verifyRemoteProviderPack(stagedRemoteProviderPackRoot); + activeRemoteProviderPackRoot = stagedRemoteProviderPackRoot; + } + remotePrepared = true; + return; + } let usedPreinstalledRunner = false; const explicitRemoteBinary = input.runnerRemoteBinaryPath?.trim() || null; if (mayUsePreinstalledRunnerArtifact(explicitRemoteBinary)) { @@ -10741,13 +10935,13 @@ async function createRunnerdBackendWithinSessionClaim( ), ); } catch { - throw new Error("runner_harness_state_mismatch"); + throw new Error("runner_harness_state_mismatch: runner_state_invalid_json"); } if ( runnerState.runnerInstanceId !== input.runnerInstanceId || runnerState.normalizedSessionId !== nativeSessionKey(input.execution) ) { - throw new Error("runner_harness_state_mismatch"); + throw new Error("runner_harness_state_mismatch: runner_identity"); } if (runnerState.lifecycle !== "suspended") { return { @@ -10765,7 +10959,7 @@ async function createRunnerdBackendWithinSessionClaim( execution: input.execution, }); } catch { - throw new Error("runner_harness_state_mismatch"); + throw new Error("runner_harness_state_mismatch: provider_state_unreadable"); } const providerSessionIdentity = providerSessionIdentityFromDurableProviderState({ @@ -10773,7 +10967,7 @@ async function createRunnerdBackendWithinSessionClaim( providerState, }); if (!providerSessionIdentityIsPresent(providerSessionIdentity)) { - throw new Error("runner_harness_state_mismatch"); + throw new Error("runner_harness_state_mismatch: provider_identity_incomplete"); } const previousManifest = compatibleNativeHarnessBackupManifests({ root, @@ -10788,7 +10982,7 @@ async function createRunnerdBackendWithinSessionClaim( current: providerSessionIdentity, }) ) { - throw new Error("runner_harness_state_mismatch"); + throw new Error("runner_harness_state_mismatch: provider_identity_changed"); } return { complete: true, @@ -11078,7 +11272,7 @@ async function createRunnerdBackendWithinSessionClaim( ) { if (state.incompleteReason !== "unavailable" || backupAvailable || !(await claimUntouchedSessionInResumedLease())) { - throw new Error("runner_harness_state_mismatch"); + throw new Error(`runner_harness_state_mismatch: resumed_${state.incompleteReason}`); } } else { await recordInPlaceHarnessReuse( @@ -11577,11 +11771,31 @@ async function createRunnerdBackendWithinSessionClaim( } remotePrepared = false; }; - const adoptedProcess = - target.kind === "local" && + const localRecoveryProcess = input.restartRecovery?.kind === "reattach_existing_runner" ? input.restartRecovery.process : null; + const adoptedProcess = + input.restartRecovery?.kind === "reattach_remote_runner" + ? await verifyRemoteRunnerReattachment({ + claim: input.restartRecovery, + target, + identity: durableIdentity ?? {}, + runId: input.execution.binding.runId, + normalizedSessionId: nativeSessionKey(input.execution), + }) + : target.kind === "local" && localRecoveryProcess + ? { + ...localRecoveryProcess, + isAlive: () => verifiedRecoveryProcessIsAlive(localRecoveryProcess), + signal: (signal: NodeJS.Signals) => + signalVerifiedRecoveryProcess(localRecoveryProcess, signal), + } + : undefined; + if (adoptedProcess && remoteTarget) { + const markSpawned = resolveRemoteRunnerProcessSpawned as (() => void) | null; + markSpawned?.(); + } const executeCurrentToolAuthority = ( call: Parameters[0], ) => { @@ -11735,14 +11949,7 @@ async function createRunnerdBackendWithinSessionClaim( : undefined, runnerProcessLauncher: remoteProcessLauncher, runnerReconnectGraceMs: remoteTarget ? 120_000 : undefined, - adoptExistingRunner: adoptedProcess - ? { - ...adoptedProcess, - isAlive: () => verifiedRecoveryProcessIsAlive(adoptedProcess), - signal: (signal) => - signalVerifiedRecoveryProcess(adoptedProcess, signal), - } - : undefined, + adoptExistingRunner: adoptedProcess, environment: effectiveRunnerEnvironment, onDiagnostic: (message) => { void input.onLog?.( diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index c77125217bd..c82779ad086 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -758,6 +758,13 @@ export async function prepareNativeWorkspaceSync(input: { } const target = input.target; const providerLeaseId = providerLeaseIdFor({ target, lease: input.lease }); + // Reattachment may never seed a replacement sandbox over a still-live run. + if (input.restartRecovery?.kind === "reattach_remote_runner" && + (input.restartRecovery.runId !== input.runId || + input.restartRecovery.remote.providerLeaseId !== providerLeaseId || + input.restartRecovery.remote.remoteCwd !== target.remoteCwd)) { + throw new Error("native_remote_recovery_lease_mismatch"); + } const run = await input.db .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) diff --git a/tests/runner-e2e/README.md b/tests/runner-e2e/README.md index ef33a335a50..c689024a78e 100644 --- a/tests/runner-e2e/README.md +++ b/tests/runner-e2e/README.md @@ -787,6 +787,10 @@ lockfile from its own trusted checkout, never from the tested branch. ### Injected interruption diagnostics +The restart supervisor starts Paperclip with the TypeScript loader in the same +Node process it owns. A forced stop therefore cannot leave an old controller +alive to stop the embedded database after the replacement starts. + Everyday restart and Stop scenarios exempt only their recorded cancellation, graceful-shutdown interruption, or process-loss outcome. A later adapter error on that same run still fails immediately and fails the lifecycle grader. The diff --git a/tests/runner-e2e/server.ts b/tests/runner-e2e/server.ts index de963273b94..4a3053b3ddb 100644 --- a/tests/runner-e2e/server.ts +++ b/tests/runner-e2e/server.ts @@ -1,3 +1,4 @@ +import { runnerE2ETypeScriptProcessArgs } from "./web-server-command.js"; import { qualifyLegacyClaudeCli } from "./legacy-claude-cli.js"; import { spawn, type ChildProcess } from "node:child_process"; import { createWriteStream } from "node:fs"; @@ -22,7 +23,6 @@ const paperclipHome = required("PAPERCLIP_HOME"); const configPath = required("PAPERCLIP_CONFIG"); const port = required("PAPERCLIP_RUNNER_E2E_PORT"); const repositoryRoot = path.resolve(import.meta.dirname, "../.."); -const tsxCli = path.join(repositoryRoot, "cli/node_modules/tsx/dist/cli.mjs"); const paperclipCli = path.join(repositoryRoot, "tests/runner-e2e/server-entry.ts"); const { controlDirectory, @@ -111,7 +111,7 @@ function startServer() { } const candidate = spawn( process.execPath, - [tsxCli, paperclipCli, "onboard", "--yes", "--run"], + runnerE2ETypeScriptProcessArgs(repositoryRoot, paperclipCli, ["onboard", "--yes", "--run"]), { cwd: repositoryRoot, env: definedServerEnvironment, diff --git a/tests/runner-e2e/web-server-command.ts b/tests/runner-e2e/web-server-command.ts index 9408f83a559..de88292fcbe 100644 --- a/tests/runner-e2e/web-server-command.ts +++ b/tests/runner-e2e/web-server-command.ts @@ -16,3 +16,8 @@ export function runnerE2EWebServerCommand(repositoryRoot: string) { const server = path.join(repositoryRoot, "tests/runner-e2e/server.ts"); return `node ${shellQuote(tsx)} ${shellQuote(server)}`; } + +// Keep the supervisor's child PID equal to the actual TypeScript process. +export function runnerE2ETypeScriptProcessArgs(repositoryRoot: string, entry: string, args: string[] = []) { + return ["--import", path.join(repositoryRoot, "cli/node_modules/tsx/dist/loader.mjs"), entry, ...args]; +} diff --git a/tests/runner-e2e/web-server-shutdown.test.ts b/tests/runner-e2e/web-server-shutdown.test.ts index 6fc14b5aedb..de2c517ddee 100644 --- a/tests/runner-e2e/web-server-shutdown.test.ts +++ b/tests/runner-e2e/web-server-shutdown.test.ts @@ -5,7 +5,7 @@ import os from "node:os"; import path from "node:path"; import { expect, it } from "vitest"; import { reserveRunnerE2EDatabasePort } from "./ports.js"; -import { runnerE2EWebServerGracefulShutdown } from "./web-server-command.js"; +import { runnerE2EWebServerGracefulShutdown, runnerE2ETypeScriptProcessArgs } from "./web-server-command.js"; const require = createRequire(import.meta.url); @@ -115,3 +115,33 @@ it("lets Playwright reap a restarted server through the production bounded shutd await rm(root, { recursive: true, force: true }); } }, 25_000); + + +it("owns the actual server PID so a forced restart cannot leave a late database closer", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-server-pid-")); + const entry = path.join(root, "server.mts"); + let candidate: ReturnType | undefined; + let actualPid: number | undefined; + try { + await writeFile(entry, `process.send!({ pid: process.pid }); setInterval(() => {}, 1000);`); + candidate = spawn(process.execPath, runnerE2ETypeScriptProcessArgs(path.resolve(import.meta.dirname, "../.."), entry), { stdio: ["ignore", "pipe", "pipe", "ipc"] }); + actualPid = await new Promise((resolve, reject) => { + candidate!.once("message", (message: any) => resolve(message.pid)); + candidate!.once("error", reject); + candidate!.once("exit", () => reject(new Error("Server exited before publishing its PID"))); + }); + expect(actualPid).toBe(candidate.pid); + const exited = new Promise(resolve => candidate!.once("exit", resolve)); + candidate.kill("SIGKILL"); + await exited; + expect(() => process.kill(actualPid!, 0)).toThrow(); + } finally { + if (actualPid) { try { process.kill(actualPid, "SIGKILL"); } catch {} } + if (candidate && candidate.exitCode === null && candidate.signalCode === null) { + const exited = new Promise(resolve => candidate!.once("exit", resolve)); + candidate.kill("SIGKILL"); + await exited; + } + await rm(root, { recursive: true, force: true }); + } +}, 10_000);