Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-10 12:37:55 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(native): continue on a replacement when a retained Runner retires before claim (#19897)

Browse Source
This commit is contained in:
Nicky Leach authored and GitHub committed 2026-11-07 14:51:32 -08:00
1 parent 91b044858c
commit 94718ce168
5 files changed
+123 -47
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/services/native-runtime/native-session-executor.test.ts
+1 -1
View File
@@ -16086,7 +16086,7 @@ describe("runnerd provider runtime wiring", () => {
state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []);
const readWorkspaceFile = vi.fn(async () => Buffer.from("proof"));
const release = vi.fn(async () => {}), resume = vi.fn(async () => "resumed" as const);
const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, rebind:vi.fn() });
const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retiredBeforeClaim: () => false, rebind:vi.fn() });
const task = vi.fn(async (operation: unknown) => {
environmentTaskOperationSchema.parse(operation);
return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, releaseResumeVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
server/src/services/native-runtime/native-session-executor.ts
+60 -36
View File
@@ -10811,40 +10811,38 @@ async function executePaperclipNativeSessionWithinScope(
const resolveWake = () => timeController("wake_binding", () => resolveCurrentWakeCommentsBinding(input.db, input.execution.binding));
const expectedCurrentWakeComments = await (executionContext ? executionContext.wait(resolveWake) : resolveWake());
let remoteWarmRecoveryPending = remoteWarmRecovery !== undefined;
const createExecutionBackend = () => createRunnerdBackend({
...input,
onDiagnosticLease: ownDiagnosticLease,
providerPackAdmissionSource,
recoveryPurpose: claimedStopCleanup ? "stop_cleanup" : "execution",
executionContext,
onAccountingEvent: producerUsage?.capture,
controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? {
bootId: controller.bootId,
runnerArtifact: restoredController ? { version: restoredController.runnerVersion, digest: restoredController.runnerDigest } : undefined,
} : undefined,
...(overlapGitHubPreparation ? { prepareRunnerEnvironment } : {}),
// Durable scope and prior-run verification use the controller's
// canonical workspace identity. createRunnerdBackend separately
// projects remoteCwd into the provider execution boundary.
execution: input.execution,
runnerInstanceId: effectiveRunnerInstanceId,
pluginTaskDescription,
retainedPluginTaskOwner: existingWarmSession ? pluginTaskSessionOwners.get(existingWarmSession) : undefined,
durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId,
remoteWarmRecovery: remoteWarmRecovery ? {
...remoteWarmRecovery,
revalidate: async () => {
if (remoteWarmRecoveryPending) await remoteWarmRecovery.revalidate(leaseOwner);
},
} : undefined,
trace,
toolTrace,
});
let runnerdBackend =
input.useRunnerd && input.backend === undefined
? await timeController("backend_setup", () => {
const create = () => createRunnerdBackend({
...input,
onDiagnosticLease: ownDiagnosticLease,
providerPackAdmissionSource,
recoveryPurpose: claimedStopCleanup ? "stop_cleanup" : "execution",
executionContext,
onAccountingEvent: producerUsage?.capture,
controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? {
bootId: controller.bootId,
runnerArtifact: restoredController ? { version: restoredController.runnerVersion, digest: restoredController.runnerDigest } : undefined,
} : undefined,
...(overlapGitHubPreparation ? { prepareRunnerEnvironment } : {}),
// Durable scope and prior-run verification use the controller's
// canonical workspace identity. createRunnerdBackend separately
// projects remoteCwd into the provider execution boundary.
execution: input.execution,
runnerInstanceId: effectiveRunnerInstanceId,
pluginTaskDescription,
retainedPluginTaskOwner: existingWarmSession ? pluginTaskSessionOwners.get(existingWarmSession) : undefined,
durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId,
remoteWarmRecovery: remoteWarmRecovery ? {
...remoteWarmRecovery,
revalidate: async () => {
if (remoteWarmRecoveryPending) await remoteWarmRecovery.revalidate(leaseOwner);
},
} : undefined,
trace,
toolTrace,
});
return executionContext ? executionContext.wait(create) : create();
})
? await timeController("backend_setup", () => executionContext ? executionContext.wait(createExecutionBackend) : createExecutionBackend())
: null;
if (remoteWarmRecovery) {
if (executionContext) await executionContext.wait(() => remoteWarmRecovery.revalidate(leaseOwner));
@@ -10877,7 +10875,7 @@ async function executePaperclipNativeSessionWithinScope(
durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, trace, toolTrace });
}
}
const remoteCleanupLease = input.runnerExecutionTarget?.kind === "remote" &&
const readRemoteCleanupLease = async () => input.runnerExecutionTarget?.kind === "remote" &&
input.runnerExecutionTarget.transport === "sandbox" && input.runnerExecutionTarget.leaseId
? await input.db.select({ provider: environmentLeases.provider, providerLeaseId: environmentLeases.providerLeaseId })
.from(environmentLeases).where(and(
@@ -10885,6 +10883,26 @@ async function executePaperclipNativeSessionWithinScope(
eq(environmentLeases.id, input.runnerExecutionTarget.leaseId),
)).then(rows => rows[0])
: null;
let remoteCleanupLease = await readRemoteCleanupLease();
const continueAfterUnavailableReuse = async (error: unknown) => {
const target = input.runnerExecutionTarget;
if (!(error instanceof Error) || error.message !== "native_runner_reuse_unavailable" || !existingWarmSession
|| pluginTaskSessionOwners.get(existingWarmSession)?.retiredBeforeClaim() !== true
|| remoteWarmRecovery || warmConfigDigest === null || target?.kind !== "remote" || target.transport !== "sandbox"
|| !target.runnerTask || runnerdBackend === null) return false;
const retainedTaskId = target.sandboxLeaseAcquisition?.providerLeaseId;
await existingWarmSession.close({ reason: "retained Runner was unavailable for reuse" });
const inspect = () => target.runnerTask!({ kind: "describe" }, executionContext ? { executionContext } : undefined);
const description = environmentTaskResultSchema.parse(await (executionContext ? executionContext.wait(inspect) : inspect()));
if (description.kind !== "description" || description.taskId === retainedTaskId) return false;
existingWarmSession = undefined;
pluginTaskDescription = undefined;
await finishStartupContextForFallback();
persistedWarmSession = loadWarmNativeCheckpoint(input.execution, warmConfigDigest, target.kind);
runnerdBackend = await createExecutionBackend();
remoteCleanupLease = await readRemoteCleanupLease();
return true;
};
nativeSessionExecuteStartedAtMs = Date.now();
native = await trace.measure(
"native.session.execute",
@@ -10894,7 +10912,7 @@ async function executePaperclipNativeSessionWithinScope(
startedAtMs: nativeSessionExecuteStartedAtMs,
});
trace.activate(runnerSessionStartupScope);
const result = await trace.run(runnerSessionStartupScope, () =>
const runExecution = () =>
executeNativeSession({
resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner",
getFreshSessionHandoff: input.getFreshSessionHandoff,
@@ -11274,8 +11292,14 @@ async function executePaperclipNativeSessionWithinScope(
);
}
},
}),
);
});
let result: Awaited<ReturnType<typeof runExecution>>;
try {
result = await trace.run(runnerSessionStartupScope, runExecution);
} catch (error) {
if (!await continueAfterUnavailableReuse(error)) throw error;
result = await trace.run(runnerSessionStartupScope, runExecution);
}
await trace.end(runnerSessionStartupScope, {
outcome:
result.terminal.runTerminalState === "succeeded" ? "ok" : "failed",
server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts
+37 -9
View File
@@ -45,10 +45,12 @@ const parkId = (source: DurableRecoveryIdentity) => `park_${sha(JSON.stringify(O
describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execution entry", () => {
const environment = nativeRestartFixture();
const nativeIt = process.platform === "win32" ? it.skip : realProcessIt;
nativeIt.for([...(["codex", "claude"] as const).flatMap(harness => (["cached", "detached", "reused lease", "prepared copy", "missing stop", "missing state", "foreign state", "working state"] as const).map(mode => [harness, mode] as const))])("continues %s after graceful eviction with a %s Core owner", { timeout: 60_000 }, async ([harness, mode], context) => {
nativeIt.for([...(["codex", "claude"] as const).flatMap(harness => (["cached", "detached", "reused lease", "after inspection", "bare unavailable", "prepared copy", "missing stop", "missing state", "foreign state", "working state"] as const).map(mode => [harness, mode] as const))])("continues %s after graceful eviction with a %s Core owner", { timeout: 60_000 }, async ([harness, mode], context) => {
if (harness === "claude" && !existsSync(fakeAcpxSidecar)) context.skip();
const detached = mode === "detached" || mode === "missing state" || mode === "foreign state";
const denied = mode.startsWith("missing") || mode.endsWith("state");
const unavailableResume = mode === "after inspection" || mode === "bare unavailable";
const reusesLease = mode === "reused lease" || unavailableResume;
const instructionCopy = (id: string) => ({ runId: id, root: workspace, preparationKey: "unchanged-fixture-instructions", hasChanges: async () => false, collectStopped: async () => {} });
const first = await environment.seedRun(`RELEASE-ENTRY-${harness}-${mode}`);
const { db, runId, issueId, native } = first;
@@ -132,6 +134,7 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
const releases: DurableRecoveryIdentity[] = [], claims: DurableRecoveryIdentity[] = [];
let handle: RunnerProcessHandle | undefined, submissions = 0, cleaning = false, sourceReadRequests = 0, sourceReadResponses = 0;
let pendingFollowup: Promise<unknown> | undefined;
let inspectedLive = false, resumeAttempts = 0, acquisitions = 0, evictRetained: (() => Promise<void>) | undefined;
const providerStarts = async () => (await readFile(calls, "utf8")).trim().split(/\r?\n/)
.filter(line => harness === "codex" ? line === "process-start" : JSON.parse(line).processStart).length;
const state = async (taskId: string = taskA) => JSON.parse(await readFile(join(runnerRoot(taskId), "runner-state.json"), "utf8"));
@@ -139,8 +142,14 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
const task = async (raw: unknown, authorityRunId: string, taskId: string): Promise<unknown> => {
const host = hosts.get(taskId)!;
const operation = environmentTaskOperationSchema.parse(raw);
if (operation.kind === "describe") return { kind: "description", taskId, launchPreparationVersion: 1,
projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact };
if (operation.kind === "describe") {
if (unavailableResume && authorityRunId !== runId && !inspectedLive) {
expect(handles.get(taskA)?.isExitConfirmed?.()).toBe(false);
inspectedLive = true;
}
return { kind: "description", taskId, launchPreparationVersion: 1,
projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact };
}
if (operation.kind === "prepare") {
expect(operation.requirements.identity.runId).toBe(authorityRunId);
expect(operation.accountIdentity).toBe("fixture-account");
@@ -205,6 +214,13 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
if (reply.status !== "parked") throw new Error("Fixture native release refused");
releases.push(operation.identity); return { kind: "released", taskId };
}
if (operation.kind === "resume" && unavailableResume) {
resumeAttempts++;
expect(inspectedLive).toBe(true);
if (mode === "after inspection") await evictRetained!();
expect(handles.get(taskA)?.isExitConfirmed?.()).toBe(mode === "after inspection");
return { kind: "resume", taskId, outcome: "unavailable" };
}
if (operation.kind === "resume") {
const retainedProviderCount = await providerStarts();
const reply = await host.request({ op: "claim", id: `claim_${sha(operation.bootstrapTicket)}`, parkId: parkId(operation.identity), bootstrapTicket: operation.bootstrapTicket });
@@ -213,7 +229,10 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
claims.push(operation.identity);
return { kind: "resume", taskId, outcome: "resumed" };
}
if (operation.kind === "stop") { await host.close(); }
if (operation.kind === "stop") {
if (mode === "after inspection" && taskId === taskA) await new Promise(done => setTimeout(done, 300));
await host.close();
}
if (operation.kind === "complete") { expect(handles.get(taskId)?.isExitConfirmed?.()).toBe(true); }
return { kind: "accepted", taskId };
};
@@ -227,6 +246,7 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
};
const proof = async () => (await supervisor()).releasedRunner;
const workers = { getWorker: () => ({ supportedMethods: ["environmentTask"] }), call: async (_id: string, method: string, params: any, _timeout?: number, _unused?: unknown, settled?: () => void) => {
if (method === "environmentAcquireLease") acquisitions++;
if (method === "environmentAcquireLease") return { providerLeaseId: taskB, metadata: { creationOperationId: params.creationOperationId } };
try { return await task(params.operation, params.runId, params.taskId); } finally { settled?.(); }
} } as unknown as PluginWorkerManager;
@@ -237,7 +257,7 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
const execute = async (current: typeof execution, leaseId: string) => {
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: current }, runnerInstanceId: native.runnerInstanceId, completionContractId: contract!.id, completionContractSha256: contract!.canonicalSha256, nativeSessionId: current.session.normalizedSessionId })
.where(eq(heartbeatRuns.id, current.binding.runId));
const target = mode === "reused lease" ? await resolveEnvironmentExecutionTarget({ db, companyId, adapterType: "paperclip_runner",
const target = reusesLease ? await resolveEnvironmentExecutionTarget({ db, companyId, adapterType: "paperclip_runner",
environment: { id: environmentId, driver: "plugin", config: { pluginKey: pluginId, driverKey: "tasks" } }, leaseId,
lease: await environmentService(db).getLeaseById(leaseId), leaseMetadata: metadata,
environmentRuntime: environmentRuntimeService(db, { pluginWorkerManager: workers }) }) : {
@@ -274,7 +294,8 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success" }).where(eq(environmentLeases.id, leaseA));
}
};
if (mode !== "reused lease") await evict();
evictRetained = evict;
if (!reusesLease) await evict();
const admitNext = async () => {
const runB = randomUUID(), leaseB = randomUUID();
await db.insert(heartbeatRuns).values({ id: runB, companyId, agentId, nativeIssueId: issueId, status: "running", runtimeMode: "native" });
@@ -282,8 +303,8 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
await db.update(issues).set({ executionRunId: runB, status: "in_progress" }).where(eq(issues.id, issueId));
const [oldLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseA));
await db.insert(environmentLeases).values({ id: leaseB, companyId, environmentId, heartbeatRunId: runB, issueId, executionWorkspaceId: workspaceId,
provider: providerKey, providerLeaseId: mode === "reused lease" ? taskA : taskB, status: "active", leasePolicy: "reuse_by_environment",
metadata: mode === "reused lease" ? { ...oldLease!.metadata, runnerTaskReused: true, runnerTaskPreviousLeaseId: leaseA,
provider: providerKey, providerLeaseId: reusesLease ? taskA : taskB, status: "active", leasePolicy: "reuse_by_environment",
metadata: reusesLease ? { ...oldLease!.metadata, runnerTaskReused: true, runnerTaskPreviousLeaseId: leaseA,
runnerTaskExecutionStopped: false } : { ...metadata, remoteCwd: workspace } });
if (mode === "reused lease") {
expect(handles.get(taskA)!.isExitConfirmed?.()).toBe(false);
@@ -295,6 +316,12 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
preparationKey: "unchanged-fixture-instructions", adopt: async () => instructionCopy(runB) }) : null;
if (mode === "prepared copy") expect(releasePreparation).toBeTypeOf("function");
pendingFollowup = execute(next, leaseB);
if (mode === "bare unavailable") {
await expect(pendingFollowup).rejects.toThrow("native_runner_reuse_unavailable");
expect({ resumeAttempts, acquisitions, submissions, claims }).toEqual({ resumeAttempts: 1, acquisitions: 0, submissions: 1, claims: [] });
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runB))).toHaveLength(0);
return runB;
}
if (denied) {
const refusal = mode === "working state" ? "checkpoint_not_settled"
: mode === "foreign state" ? "Plugin Runner checkpoint identity mismatch"
@@ -316,12 +343,13 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu
return runB;
};
const runB = await admitNext();
if (denied) return;
if (denied || mode === "bare unavailable") return;
await waitForCondition("automatic release B", () => releases.length === 2);
expect(await proof()).toMatchObject({ source: releases[1], taskId: taskB });
expect((await supervisor()).snapshot.providerSessionId).toBe(providerSessionId);
const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/);
expect(submissions).toBe(2);
if (mode === "after inspection") expect({ resumeAttempts, acquisitions }).toEqual({ resumeAttempts: 1, acquisitions: 1 });
expect(releases[1]!.runId).toBe(runB);
expect(claims).toEqual([]);
if (harness === "codex") {
server/src/services/native-runtime/plugin-runner-task.test.ts
+19 -1
View File
@@ -653,7 +653,25 @@ describe("plugin Runner release/resume facade", () => {
const owner = createPluginRunnerTaskOwner({ ...input("codex"), task });
expect(await owner.driver.adoptExistingRunner!.isAlive()).toBe(false);
await expect(owner.resume(requirements.identity, "transient-ticket")).resolves.toBe("unavailable");
expect(task.mock.calls.map(([operation]) => operation.kind)).toEqual(["status", "resume"]);
expect(task.mock.calls.map(([operation]) => operation.kind)).toEqual(["status", "resume", "status"]);
expect(owner.retiredBeforeClaim()).toBe(false); // stopped, but no saved session state
});
it("records a refused resume as retired only when the exact task stopped with saved session state", async () => {
for (const [status, retired] of [
[{ kind: "status", taskId: "task", phase: "cancelled", executionStopped: true, sessionStateAvailable: true }, true],
[{ kind: "status", taskId: "task", phase: "running", executionStopped: false, sessionStateAvailable: false }, false],
[new Error("fixture status reply lost"), false],
] as const) {
const task = vi.fn(async (operation: any): Promise<unknown> => {
if (operation.kind === "resume") return { kind: "resume", taskId: "task", outcome: "unavailable" };
if (status instanceof Error) throw status;
return status;
});
const owner = createPluginRunnerTaskOwner({ ...input("codex"), task });
await expect(owner.resume(requirements.identity, "transient-ticket")).resolves.toBe("unavailable");
expect(owner.retiredBeforeClaim()).toBe(retired);
}
});
it.each(["release", "resume"] as const)("validates exact task and closed %s result", async operation => {
server/src/services/native-runtime/plugin-runner-task.ts
+6
View File
@@ -73,6 +73,7 @@ export function createPluginRunnerTaskOwner(input: {
let confirmStopped: (code: number) => void = () => {};
let stoppedState: Extract<PluginEnvironmentTaskResult, { kind: "status" }> | undefined;
let completed = false;
let retiredBeforeClaim = false;
let rejectFailure!: (error: unknown) => void;
const failure = new Promise<never>((_resolve, reject) => { rejectFailure = reject; });
void failure.catch(() => undefined);
@@ -413,12 +414,17 @@ export function createPluginRunnerTaskOwner(input: {
const result = await call(environmentTaskOperationSchema.parse({ kind: "resume", identity, bootstrapTicket }));
requireReleaseResume();
if (result.kind !== "resume") throw new Error("Plugin Runner resume result unavailable");
if (result.outcome === "unavailable") {
const state = await call({ kind: "status" }).catch(() => undefined);
retiredBeforeClaim = state?.kind === "status" && state.executionStopped === true && state.sessionStateAvailable === true;
}
return result.outcome;
};
return { launcher, registration, driver, readWorkspaceFile, release, resume,
// This is a lease-bound Launcher receipt, not a terminal phase or a sent
// stop request. Completion also proves credential/writeback obligations ran.
isRetired: () => stoppedState?.executionStopped === true && completed && !detached,
retiredBeforeClaim: () => retiredBeforeClaim,
rebind(next: typeof input) {
if (!launched || stopRequested || stoppedState || detached || next.taskId !== taskId || next.harness !== input.harness ||
input.accountIdentity !== next.accountIdentity || input.credentialIdentity !== next.credentialIdentity ||