diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index 51f03ff31ac..a8caaa4a643 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -480,6 +480,14 @@ class HarnessNativeSession implements NativeSession { }; } + beginRunAttachment(input: { identity: OpenNativeSessionInput["identity"] }): boolean { + const currentIdentity = this.#input.identity; + if (this.#protocolIntegrityFailure !== null || this.#session.beginRunAttachment === undefined + || input.identity.sessionId !== currentIdentity.sessionId || input.identity.companyId !== currentIdentity.companyId + || input.identity.issueId !== currentIdentity.issueId || input.identity.agentId !== currentIdentity.agentId) return false; + return this.#session.beginRunAttachment({ runId: input.identity.runId }); + } + async attachRun(input: { identity: OpenNativeSessionInput["identity"]; currentRunGrant?: NativeRunRuntimeGrant; diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index de5e9c8a29f..d4ccd77bc67 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -564,6 +564,7 @@ export interface HarnessSession { displayId?: string | null; }; events(): AsyncIterable; + beginRunAttachment?(input: { runId: string }): boolean; attachRun?(input: { runId: string; currentRunGrant?: NativeRunRuntimeGrant; completionContract?: { revision: string; criterionIds: readonly string[] } }): Promise | void; startTurn(input: { message: NativeUserMessage; diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 3ac90be6d7d..d4e3a4effea 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -217,6 +217,7 @@ export interface NativeSession { capabilities(): Promise; /** completionContract is the attaching run's own contract; a retained session must not keep the prior run's. */ attachRun?(input: { identity: NativeRunIdentity; currentRunGrant?: NativeRunRuntimeGrant; completionContract?: { revision: string; criterionIds: readonly string[] } }): Promise; + beginRunAttachment?(input: { identity: NativeRunIdentity }): boolean; /** Return the confirmed released source, or false when active use was not released. */ release?(): Promise; /** Relinquish controller authority without suspending provider execution. */ diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index fb7cdbf5958..cbb6186a008 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -80,6 +80,7 @@ export interface CodexAppServerTransport { /** Relinquish controller authority while leaving durable runner work alive. */ detachControllerForRestart?(): Promise; processInfo?(): CodexTransportProcessInfo; + beginRunAttachment?(input: { runId: string }): boolean; attachRun?(input: { runId: string; turnId: string; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index 2c7c2ab0ad4..a3e0a3c0e04 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -110,6 +110,11 @@ export class CodexHarnessSession }; } + beginRunAttachment(input: { runId: string }): boolean { + if (this.protocolIntegrityFailure !== null || this.turnStartPending || this.pendingRunAttachment !== null || !input.runId) return false; + return this.transport.beginRunAttachment?.(input) ?? false; + } + async attachRun(input: { runId: string; currentRunGrant?: NativeRunRuntimeGrant; completionContract?: { revision: string; criterionIds: readonly string[] } }): Promise { this.assertProtocolIntegrity(); const transportOwnsQuiescence = this.transport.attachRun !== undefined; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-release-resume.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-release-resume.test.ts index fc5232c61d6..36b8843de43 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport-release-resume.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-release-resume.test.ts @@ -78,7 +78,7 @@ async function nativeWire(port: number, path: string, closeGate?: Promise) }); } -it.skipIf(process.platform === "win32").each(["cached", "lost claim reply", "cache adoption", "acpx cache adoption", "unavailable", "claim cancellation", "expired ticket", "late claim reply", "held past deadline", "release admission", "no cleanup callback", "adoption no cleanup callback", "resume no cleanup callback", "deferred turn after close", "late wire close", "empty goal snapshot"] as const)( +it.skipIf(process.platform === "win32").each(["cached", "lost claim reply", "cache adoption", "acpx cache adoption", "unavailable", "claim cancellation", "expired ticket", "late claim reply", "held past deadline", "release admission", "no cleanup callback", "adoption no cleanup callback", "resume no cleanup callback", "deferred turn after close", "late wire close", "empty goal snapshot", "early claim", "early claim abandoned", "early claim cancellation", "early claim expired", "early claim stale ticket"] as const)( "releases and resumes the native Codex source before attachment (%s)", async mode => { const root = await mkdtemp(join(tmpdir(), "runner-release-resume-")); const acpx = mode === "acpx cache adoption"; @@ -119,7 +119,7 @@ it.skipIf(process.platform === "win32").each(["cached", "lost claim reply", "cac const heldMeasuredTurn = new Promise(resolve => { finishMeasuredTurn = resolve; }); const sockets = new Set(); const registration = async (core: DurablePrpControlPlane, source = core.store.state.identity) => { - if ((mode === "late claim reply" || mode === "held past deadline") && claimAccepted) throw new Error("Fixture source authentication started after expiry"); + if ((mode === "late claim reply" || mode === "held past deadline" || mode === "early claim expired") && claimAccepted) throw new Error("Fixture source authentication started after expiry"); authority = core; let socket: PrpWireConnection | undefined; return { @@ -165,8 +165,8 @@ it.skipIf(process.platform === "win32").each(["cached", "lost claim reply", "cac parkId: parkId(source), bootstrapTicket: ticket }); if (result.status !== "claimed") throw new Error("Fixture claim receipt unavailable"); claimAccepted = true; acceptClaim(); - if (mode === "claim cancellation" || mode === "held past deadline") await claimReceipt; - if (mode === "expired ticket" || mode === "late claim reply") { + if (mode === "claim cancellation" || mode === "held past deadline" || mode === "early claim" || mode === "early claim cancellation") await claimReceipt; + if (mode === "expired ticket" || mode === "late claim reply" || mode === "early claim expired") { expireClock = vi.spyOn(Date, "now").mockReturnValue(Date.now() + 61_000); if (mode === "expired ticket") throw new Error("Fixture lost claim reply past ticket lifetime"); } @@ -347,12 +347,51 @@ it.skipIf(process.platform === "win32").each(["cached", "lost claim reply", "cac await expect(bundle.transport.request("thread/goal/set", { threadId, objective: "Must stay fenced" })).rejects.toThrow("release_resume_pending"); } if (mode === "unavailable") await fixture.close(); + if (mode.startsWith("early claim")) { + expect(bundle.transport.beginRunAttachment!({ runId: claimantRunId })).toBe(true); + expect(bundle.transport.beginRunAttachment!({ runId: claimantRunId })).toBe(false); + await vi.waitFor(() => expect(pair.resume).toHaveBeenCalledOnce()); + await expect(bundle.transport.release!()).rejects.toThrow("run_attachment_in_progress"); + await expect(bundle.transport.attachRun!({ runId: "other-run", turnId: "other-turn", itemId: "other-item" })).rejects.toThrow("run_attachment_in_progress"); + await expect(bundle.transport.request("turn/start", { input: [] })).rejects.toThrow("run_attachment_in_progress"); + expect(routeRuns).toEqual([identity.runId]); + } + if (mode === "early claim cancellation") { + await vi.waitFor(() => expect(claimAccepted).toBe(true)); + const closing = bundle.transport.close(); + releaseClaim(); + await closing; + await expect(bundle.transport.attachRun!({ runId: claimantRunId, turnId: "next-turn", itemId: "next-item" })).rejects.toThrow("closed"); + } + if (mode === "early claim abandoned" || mode === "early claim stale ticket") { + await vi.waitFor(async () => expect(await pair.resume.mock.results[0]?.value).toBe("resumed")); + if (mode === "early claim abandoned") await bundle.transport.close(); + else { + expireClock = vi.spyOn(Date, "now").mockReturnValue(Date.now() + 61_000); + await expect(bundle.transport.attachRun!({ runId: claimantRunId, turnId: "next-turn", itemId: "next-item" })).rejects.toThrow("reconciliation_expired"); + expireClock.mockRestore(); expireClock = undefined; + await bundle.transport.close(); + } + expect(routeRuns).toEqual([identity.runId]); + expect(pair.release).toHaveBeenCalledOnce(); + } if (mode === "held past deadline") vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); const attaching = Promise.resolve(recoveredSession ? recoveredSession.attachRun!({ runId: claimantRunId }) : bundle.transport.attachRun!({ runId: claimantRunId, turnId: "next-turn", itemId: "next-item" })); void attaching.catch(() => undefined); - if (mode === "held past deadline") { + if (mode === "early claim cancellation" || mode === "early claim abandoned" || mode === "early claim stale ticket") { + await expect(attaching).rejects.toThrow("closed"); + } else if (mode === "early claim") { + await vi.waitFor(() => expect(claimAccepted).toBe(true)); + let attached = false; + void attaching.then(() => { attached = true; }, () => undefined); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(attached).toBe(false); + expect(routeRuns).toEqual([identity.runId]); + releaseClaim(); + await attaching; + } else if (mode === "held past deadline") { await acceptedClaim; let failure: unknown; const settled = attaching.catch(error => { failure = error; }); @@ -372,13 +411,14 @@ it.skipIf(process.platform === "win32").each(["cached", "lost claim reply", "cac releaseClaim(); await expect(attaching).rejects.toThrow("closed"); await closing; - } else if (mode === "unavailable" || mode === "expired ticket" || mode === "late claim reply" || mode === "resume no cleanup callback") { + } else if (mode === "unavailable" || mode === "expired ticket" || mode === "late claim reply" || mode === "resume no cleanup callback" || mode === "early claim expired") { await expect(attaching).rejects.toThrow(mode === "unavailable" ? "reuse_unavailable" : mode === "resume no cleanup callback" ? "released_cleanup_unavailable" : "reconciliation_expired"); expireClock?.mockRestore(); expireClock = undefined; await bundle.transport.close(); } else await attaching; expect(new Set(claimTicketDigests).size).toBe(1); - if (["unavailable", "claim cancellation", "expired ticket", "late claim reply", "held past deadline", "resume no cleanup callback"].includes(mode)) { + if (["unavailable", "claim cancellation", "expired ticket", "late claim reply", "held past deadline", "resume no cleanup callback", + "early claim abandoned", "early claim cancellation", "early claim expired", "early claim stale ticket"].includes(mode)) { expect(pair.resume).toHaveBeenCalledOnce(); expect(authority!.store.state.identity.runId).toBe(identity.runId); expect(launch).toHaveBeenCalledOnce(); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 493b5e8cf91..62b46db025e 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4000,6 +4000,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #detachPromise: Promise | null = null; #startupOperation: Promise | null = null; #attachmentOperation: Promise | null = null; + #earlyClaim: { runId: string; operation: Promise } | null = null; #releaseEpisode: { source: DurableRecoveryIdentity; confirmed: boolean; claimed?: boolean; claimantRunId?: string; ticket?: string; ticketFlushed?: boolean; deadline?: number; @@ -4721,6 +4722,21 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (this.options.adoptExistingRunner && this.#adoptedRunnerMonitor === null) this.#watchAdoptedRunner(this.options.adoptExistingRunner); } + beginRunAttachment(input: { runId: string }): boolean { + if (this.#attachmentOperation !== null || !this.#releaseEpisode?.confirmed || this.#releaseEpisode.claimed || !input.runId) return false; + try { + this.#assertStartupOpen(); + this.#assertRunAttachmentAdmission(); + } catch { + return false; + } + const operation = this.#claimReleasedRunner(input.runId); + void operation.catch(() => undefined); + this.#attachmentOperation = operation; + this.#earlyClaim = { runId: input.runId, operation }; + return true; + } + attachRun(input: { runId: string; turnId: string; @@ -4728,12 +4744,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { currentRunGrant?: NativeRunRuntimeGrant; completionContract?: { revision: string; criterionIds: readonly string[] }; }): Promise { - if (this.#attachmentOperation !== null) { + const earlyClaim = this.#earlyClaim?.runId === input.runId ? this.#earlyClaim.operation : undefined; + if (this.#attachmentOperation !== null && earlyClaim === undefined) { return Promise.reject(new Error("native_runner_run_attachment_in_progress")); } + this.#earlyClaim = null; const operation = Promise.resolve().then(() => { this.#assertStartupOpen(); - return this.#attachRun(input); + return this.#attachRun(input, earlyClaim); }).finally(() => { this.#attachmentOperation = null; }); this.#attachmentOperation = operation; return operation; @@ -4745,20 +4763,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { itemId: string; currentRunGrant?: NativeRunRuntimeGrant; completionContract?: { revision: string; criterionIds: readonly string[] }; - }): Promise { + }, earlyClaim?: Promise): Promise { const previous = record(this.#runAttachTemplate?.provider).runtimeContext; const current = input.currentRunGrant?.runtimeContext; - return this.options.measureOperation ? this.options.measureOperation("runner.transport.attach", () => this.#attachRunWork(input), - previous === undefined ? undefined : {contextReused:current === undefined || commandDigest(current) === commandDigest(previous)}) : this.#attachRunWork(input); + return this.options.measureOperation ? this.options.measureOperation("runner.transport.attach", () => this.#attachRunWork(input, earlyClaim), + previous === undefined ? undefined : {contextReused:current === undefined || commandDigest(current) === commandDigest(previous)}) : this.#attachRunWork(input, earlyClaim); } - async #attachRunWork(input: { - runId: string; - turnId: string; - itemId: string; - currentRunGrant?: NativeRunRuntimeGrant; - completionContract?: { revision: string; criterionIds: readonly string[] }; - }): Promise { + #assertRunAttachmentAdmission(): { core: DurablePrpControlPlane; retainingSource: boolean } { if (this.#pendingWarmRecoveryCompletion !== null) { throw new Error("native_runner_warm_transition_completion_pending"); } @@ -4770,6 +4782,23 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (retainingSource && this.#bridgedRuntimeInputs.size > 0) { throw new NativeRunnerSourceReconciliationRequiredError("runtime"); } + return { core, retainingSource }; + } + + async #attachRunWork(input: { + runId: string; + turnId: string; + itemId: string; + currentRunGrant?: NativeRunRuntimeGrant; + completionContract?: { revision: string; criterionIds: readonly string[] }; + }, earlyClaim?: Promise): Promise { + const { core, retainingSource } = this.#assertRunAttachmentAdmission(); + if (earlyClaim) { + await earlyClaim; + if (this.#releaseEpisode && Date.now() >= this.#releaseEpisode.deadline!) { + throw new Error("native_runner_resume_reconciliation_expired"); + } + } if (this.#releaseEpisode) await this.#resumeReleasedRunner(input.runId); await this.#awaitWarmRunAttachmentReady(); if (retainingSource && !core.semanticToolResultsSettled()) { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index a908267d1c0..3e46964c3f9 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10707,6 +10707,7 @@ async function executePaperclipNativeSessionWithinScope( `native-warm-session:${input.execution.binding.runId}`, ); let existingWarmSession: NativeSession | undefined; + let reusingReleasedRunner = false; let pluginTaskDescription: Extract,{kind:"description"}> | undefined; let managedCredentialSession: NativeSession | undefined; let githubAccess: NativeGitHubAccess | undefined; @@ -10798,6 +10799,7 @@ async function executePaperclipNativeSessionWithinScope( entry.instructionWorkingCopy = input.instructionWorkingCopy?.checkpointWarm ? undefined : input.instructionWorkingCopy; entry.busy = true; entry.ownerToken = warmSessionOwnerToken; + reusingReleasedRunner = entry.released === true; entry.released = false; entry.environmentId = input.runnerExecutionTarget?.environmentId ?? null; @@ -10960,6 +10962,10 @@ 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 beginRetainedRunAttachment = (session: NativeSession) => () => { + session.beginRunAttachment?.({ identity: { runId: input.execution.binding.runId, sessionId: session.identity().sessionId, + companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, agentId: input.execution.binding.agentId } }); + }; const createExecutionBackend = () => createRunnerdBackend({ ...input, onDiagnosticLease: ownDiagnosticLease, @@ -10979,6 +10985,7 @@ async function executePaperclipNativeSessionWithinScope( runnerInstanceId: effectiveRunnerInstanceId, pluginTaskDescription, retainedPluginTaskOwner: existingWarmSession ? pluginTaskSessionOwners.get(existingWarmSession) : undefined, + beginRetainedRunAttachment: existingWarmSession && reusingReleasedRunner ? beginRetainedRunAttachment(existingWarmSession) : undefined, durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, remoteWarmRecovery: remoteWarmRecovery ? { ...remoteWarmRecovery, @@ -14193,6 +14200,7 @@ export async function createRunnerdBackend(input: { durableEnvironmentLeaseId?: string; remoteWarmRecovery?: Omit & { revalidate: () => Promise }; retainedPluginTaskOwner?: ReturnType; + beginRetainedRunAttachment?: () => void; pluginTaskDescription?: Extract,{kind:"description"}>; onSpawn?: (meta: { pid: number; @@ -14487,7 +14495,13 @@ async function createRunnerdBackendWithinSessionClaim( if (relayAssignedMcp && hasAssignedMcp && !assignedGatewayPublicId) { throw new Error("assigned native MCP gateway path is invalid"); } - const [preparedEnvironment, preparedAssignedMcp] = await Promise.allSettled([ + let durableState: ReturnType | undefined; + let pluginTaskOwner: ReturnType | undefined; + const earlyPluginTaskOwner = pluginTask && remoteTarget?.transport === "sandbox" && input.beginRetainedRunAttachment + && input.retainedPluginTaskOwner && !input.remoteWarmRecovery && !input.restartRecovery + ? pluginTaskOwner = preparePluginTaskOwner().then(prepared => { input.beginRetainedRunAttachment!(); return prepared; }) + : undefined; + const [preparedEnvironment, preparedAssignedMcp, preparedPluginTaskOwner] = await Promise.allSettled([ Promise.resolve().then(() => input.prepareRunnerEnvironment?.() ?? input.runnerEnvironment), Promise.resolve().then(() => relayAssignedMcp && !nativeReview && assignedMcpUrl && assignedMcpToken ? timeController("assigned_mcp", () => createAssignedMcpTools({ @@ -14498,9 +14512,11 @@ async function createRunnerdBackendWithinSessionClaim( ...(input.execution.provider.kind === "opencode" ? { delivery: "on_demand" as const } : {}), })) : undefined), + earlyPluginTaskOwner, ]); if (preparedEnvironment.status === "rejected") throw preparedEnvironment.reason; if (preparedAssignedMcp.status === "rejected") throw preparedAssignedMcp.reason; + if (preparedPluginTaskOwner.status === "rejected") throw preparedPluginTaskOwner.reason; const runnerEnvironment = preparedEnvironment.value; const assignedMcpTools = preparedAssignedMcp.value; const dotAttachmentActor = input.execution.schema === "paperclip.native-execution-input.v6" @@ -14565,13 +14581,7 @@ async function createRunnerdBackendWithinSessionClaim( authorityEpoch.revoke(); throw error; } - const root = runnerdStateRoot(input.execution); - const durableIdentity = readRunnerdDurableIdentity(root); - const durableBinding = retainedTransition - ? loadRunnerdDurableBinding(input.execution, retainedTransition) - : durableIdentityMatchesSession(durableIdentity, input.execution) - ? durableIdentity - : null; + const { root, durableIdentity, durableBinding } = durableState ??= readDurableState(); const stagedRecovery = input.restartRecovery?.kind === "reattach_remote_runner" ? await verifyStagedAttachmentRestart(input) : null; const retainedSystemInstructions = input.restartRecovery?.kind === "reattach_remote_runner" @@ -14632,7 +14642,21 @@ async function createRunnerdBackendWithinSessionClaim( input.durableEnvironmentLeaseId ?? input.execution.binding.executionWorkspaceId; mkdirSync(root, { recursive: true, mode: 0o700 }); - if (pluginTask && remoteTarget?.transport === "sandbox" && input.execution.schema !== "paperclip.native-execution-input.v6" && input.execution.schema !== "paperclip.native-execution-input.v8") { + function readDurableState() { + const root = runnerdStateRoot(input.execution); + const durableIdentity = readRunnerdDurableIdentity(root); + const durableBinding = retainedTransition + ? loadRunnerdDurableBinding(input.execution, retainedTransition) + : durableIdentityMatchesSession(durableIdentity, input.execution) + ? durableIdentity + : null; + return { root, durableIdentity, durableBinding }; + } + async function preparePluginTaskOwner() { + if (!pluginTask || remoteTarget?.transport !== "sandbox" || input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8") { + throw new Error("Plugin Runner tasks require a workspace session"); + } + const { root, durableIdentity, durableBinding } = durableState ??= readDurableState(); const provider = input.execution.provider; if (provider.kind !== "codex" && provider.kind !== "opencode" && provider.kind !== "acpx") throw new Error("Plugin Runner tasks require a workspace harness"); if (remoteTarget.sandboxLeaseAcquisition?.outcome === "resumed" && !durableBinding && !input.retainedPluginTaskOwner && !input.remoteWarmRecovery) { @@ -14719,6 +14743,10 @@ async function createRunnerdBackendWithinSessionClaim( if (input.execution.session.lifecyclePolicy.mode === "warm" && (!credentials.accountIdentity || description.warmTaskVersion !== 1)) throw new Error("Plugin warm task capability or account unavailable"); const owner = retainedOwner ?? createPluginRunnerTaskOwner(ownerInput); if (retainedOwner) retainedOwner.rebind(ownerInput); + return { provider, description, prior, adoptionIdentity, adopting, owner }; + } + if (pluginTask && remoteTarget?.transport === "sandbox" && input.execution.schema !== "paperclip.native-execution-input.v6" && input.execution.schema !== "paperclip.native-execution-input.v8") { + const { provider, description, prior, adoptionIdentity, adopting, owner } = await (pluginTaskOwner ??= preparePluginTaskOwner()); if (description.workspaceFileReadVersion === 1) { authority.binding.readRemoteApiFile = owner.readWorkspaceFile; authority.binding.readRemoteTaskFile = file => owner.readWorkspaceFile(file.contentRef, file.byteSize); diff --git a/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts index fed25e693ab..79d0a44655b 100644 --- a/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts +++ b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts @@ -6,10 +6,10 @@ import https, { createServer } from "node:https"; import { createServer as createTcpServer } from "node:net"; import { join, resolve } from "node:path"; import { getCACertificates, setDefaultCACertificates } from "node:tls"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import WebSocket, { WebSocketServer, type RawData } from "ws"; import { expect, it, vi } from "vitest"; -import { completionContracts, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { completionContracts, environmentLeases, environments, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; import { environmentTaskOperationSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import { readRunnerdArtifactBinding, defaultCapabilityRunnerdBinary, parseNativeExecutionInput, parseNativeRuntimeContext, QUALIFIED_ACPX_PROFILES, type DurableRecoveryIdentity, type RunnerProcessHandle, type RunnerProcessLaunchSpec } from "../../vendor/paperclip-runner/index.js"; @@ -23,6 +23,8 @@ import * as remoteWarmRecovery from "./remote-warm-session-recovery.js"; import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js"; import { buildNativeCompletionContract } from "./completion-contracts.js"; import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; +import { PROCESS_START_REQUESTED } from "../native-local-process-stop.js"; +import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; const { createPrivateHostControlFixture } = await import(new URL("../../../../packages/paperclip-runner/test/support/private-host-control.ts", import.meta.url).href) as { createPrivateHostControlFixture(): { @@ -42,9 +44,11 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio const environment = nativeRestartFixture(); const nativeIt = process.platform === "win32" ? it.skip : realProcessIt; nativeIt.for([...(["codex", "claude"] as const).flatMap(harness => (["cached", "adopted"] as const).map(mode => [harness, mode] as const)), ["codex", "adopted cancellation"] as const, ["codex", "adopted legacy cancellation"] as const, + ["codex", "cached cancellation"] as const, ["codex", "cached held claim"] as const, ["claude", "cached held claim"] as const, ["codex", "cached managed GitHub"] as const, ["codex", "adopted managed GitHub"] as const])("completes %s runs through a retained %s process", { timeout: 60_000 }, async ([harness, mode], context) => { if (harness === "claude" && !existsSync(fakeAcpxSidecar)) context.skip(); - const adopted = mode.startsWith("adopted"), cancellation = mode.includes("cancellation"), qualifiedCancellation = mode === "adopted cancellation"; + const adopted = mode.startsWith("adopted"), cancellation = mode.includes("cancellation"), heldClaimMode = mode === "cached held claim"; + const qualifiedCancellation = mode === "adopted cancellation" || mode === "cached cancellation"; const managedGitHub = mode.endsWith("managed GitHub"), brokerUrl = "https://core.example.test/api", brokerToken = (runId: string) => `fixture-broker-${runId}`; const preparedBrokers: Array<{ runId: string; token?: string }> = []; const first = await environment.seedRun(`RELEASE-ENTRY-${harness}-${mode}`); @@ -121,6 +125,18 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio const host = createPrivateHostControlFixture(), runnerRoot = join(root, "native"), binary = defaultCapabilityRunnerdBinary(); const artifact = readRunnerdArtifactBinding(binary), preparations = new Map; launch: { configurationId: string } }>(); const releases: DurableRecoveryIdentity[] = [], claims: DurableRecoveryIdentity[] = []; + const operationsDuringClaim: string[] = []; + let claimHeld = false, holdPreparationFor: string | undefined, claimedDuringPreparation = false; + const toolDefinitions = PaperclipRunnerToolAuthority.prototype.definitions; + const preparation = vi.spyOn(PaperclipRunnerToolAuthority.prototype, "definitions").mockImplementation(function (this: PaperclipRunnerToolAuthority) { + const definitions = toolDefinitions.call(this); + if (holdPreparationFor === undefined || this.binding.runId !== holdPreparationFor) return definitions; + holdPreparationFor = undefined; + return waitForCondition("claim during preparation", () => claims.length === 1) + .then(() => { claimedDuringPreparation = true; return definitions; }) as unknown as typeof definitions; + }); + const processStartRequested = async (id: string) => (await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.runId, id), eq(heartbeatRunEvents.eventType, PROCESS_START_REQUESTED)))).length > 0; let handle: RunnerProcessHandle | undefined, submissions = 0, stops = 0, completes = 0; let releaseClaimReply!: () => void, observeHeldClaim!: () => void; const claimReply = new Promise(done => { releaseClaimReply = done; }); @@ -132,6 +148,7 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio const sameSource = (a: DurableRecoveryIdentity, b: DurableRecoveryIdentity) => Object.entries(a).every(([key, value]) => b[key as keyof DurableRecoveryIdentity] === value); const task = async (raw: unknown, authorityRunId: string): Promise => { const operation = environmentTaskOperationSchema.parse(raw); + if (claimHeld) operationsDuringClaim.push(operation.kind); if (operation.kind === "describe") return { kind: "description", taskId, launchPreparationVersion: 1, projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact, ...(qualifiedCancellation ? { executionContextVersion: 1 } : {}), ...(managedGitHub ? { githubCredentialBrokerVersion: 1 } : {}) }; @@ -190,6 +207,13 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio if (reply.status !== "claimed") throw new Error("Fixture native claim refused"); expect(await providerStarts()).toBe(retainedProviderCount); claims.push(operation.identity); + if (heldClaimMode) { + expect(await processStartRequested(authorityRunId)).toBe(false); + claimHeld = true; + await waitForCondition("preparation completes while the claim is held", () => processStartRequested(authorityRunId)); + await new Promise(done => setTimeout(done, 200)); + claimHeld = false; + } if (cancellation) { observeHeldClaim(); await claimReply; } return { kind: "resume", taskId, outcome: "resumed" }; } @@ -247,6 +271,7 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio await db.insert(environmentLeases).values({ id: leaseB, companyId, environmentId, heartbeatRunId: runB, issueId, executionWorkspaceId: workspaceId, provider: "plugin:fixture:tasks", providerLeaseId: taskId, status: "active", leasePolicy: "reuse_by_environment", metadata: { remoteCwd: workspace } }); const next = { ...execution, binding: { ...execution.binding, runId: runB } }; + if (heldClaimMode) holdPreparationFor = runB; pendingFollowup = execute(next, leaseB); if (cancellation) { void pendingFollowup.catch(() => undefined); @@ -290,6 +315,7 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio expect(adoption).toHaveBeenCalledTimes(1); } expect(claims).toEqual(releases.slice(0, -1)); expect(releases[1]!.runId).toBe(runB); + if (heldClaimMode) expect({ claimedDuringPreparation, operationsDuringClaim }).toEqual({ claimedDuringPreparation: true, operationsDuringClaim: [] }); expect(releases[1]!.environmentLeaseId).toBe(leaseA); expect(handle!.child.pid).toBe(pid); expect(submissions).toBe(1); expect([...new Set(preparedBrokers.map(prepared => prepared.runId))]).toEqual(releases.map(release => release.runId)); const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/); @@ -316,7 +342,7 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio const failures = cleanup.filter((result): result is PromiseRejectedResult => result.status === "rejected"); if (failures.length) throw new AggregateError(failures.map(result => result.reason), "Fixture cleanup failed"); } finally { - adoption.mockRestore(); + adoption.mockRestore(); preparation.mockRestore(); https.globalAgent.destroy(); https.globalAgent = oldAgent; setDefaultCACertificates(oldCAs); } }