From 9046884811a7b06131ec60ce9c9f0255c8e84465 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sun, 8 Nov 2026 22:56:08 -0800 Subject: [PATCH] perf(native-runtime): claim a parked Runner while the follow-up run prepares (#20195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Paperclip Runner executes native agent runs. A host can park an idle Runner (release it and freeze it) between runs, and resume it for the next run on the same task. > - A follow-up run that reuses a parked Runner must claim it first. The claim is a bootstrap ticket, a checkpoint flush, and the plugin resume that thaws the Runner. > - Core made the claim only inside `attachRun`, after all of its per-run preparation. The claim does not depend on that preparation, so the two ran in series. > - This pull request starts the claim while Core still prepares the run. `attachRun` then waits for the claim that is already in progress. > - The benefit is a shorter start for every follow-up run on a parked Runner. The model below gives about 285 ms per run at P50. ## Linked Issues or Issue Description No public issue exists. The issue is described inline. **What happened?** A follow-up run that reuses a parked Runner claims the Runner only in `attachRun`. Core calls `attachRun` after backend setup: runner environment, assigned MCP tools, tool authority and the plugin task owner. The claim waits for all of this, but it uses none of it. **Expected behavior** The claim runs at the same time as backend setup. Everything that needs the new run's tool authority stays at attach. **Steps to reproduce** 1. Run a Codex or Claude task on a host that parks Runners between runs. 2. Let the run finish and the Runner park. 3. Send a follow-up. Before this change, the plugin resume starts only after backend setup ends. **Paperclip version or commit** `master` at 7086019d4b. **Deployment mode** Hosted, with parked Runner tasks. ## What Changed - **Transport** (`runnerd-codex-transport.ts`): - New `beginRunAttachment({ runId })`. It runs the attach admission checks, then starts the existing `#claimReleasedRunner` as the held attachment operation. It returns `false` and does nothing when the Runner is not released and confirmed, a claim already exists, or admission fails. - The admission checks move into `#assertRunAttachmentAdmission`. `beginRunAttachment` runs them first, and `attachRun` runs them again. - The next `attachRun` for the same run waits for the early claim. An `attachRun` for a different run, `release` and `turn/start` still fail with `native_runner_run_attachment_in_progress` while the claim is held. - After an early claim, `attachRun` fails with `native_runner_resume_reconciliation_expired` before it registers if the claim deadline has passed. - **Pass-throughs:** `HarnessNativeSession`, `CodexHarnessSession` and the session, harness and transport contracts get an optional `beginRunAttachment`. Each layer does the same identity, protocol-integrity and busy checks that its attach does. - **Core** (`native-session-executor.ts`): - The executor records when the in-memory warm entry it reuses was parked (`released`). - `createRunnerdBackend` gets `beginRetainedRunAttachment`. Core gives it only for that parked in-memory warm entry. - The plugin task owner preparation moves into `preparePluginTaskOwner`. On the early path, Core prepares and rebinds the retained owner at the same time as the runner environment and the assigned MCP tools, then starts the claim. - Cold, adopted, remote warm recovery and restart paths keep the old order. ### What stays at attach, and why - Registration, the quiescence probe, `run.attach` and the authority rotation stay in `attachRun`. Checkpoint registration needs the new run's tool authority epoch, and Core creates that epoch during backend setup. - The admission checks run again at attach, so a state change during preparation still blocks the attachment. ## Verification - `vitest run src/live/runnerd-codex-transport-release-resume.test.ts` (runner): 21 passed. Five new modes: - `early claim`: `attachRun` waits for the held claim and does not route the new run before the claim ends; - `early claim abandoned`: a close after the claim ends fails the waiting `attachRun` with `closed`, and the new run is never routed; - `early claim cancellation`: a close during the claim cancels it, and the next `attachRun` fails with `closed`; - `early claim expired`: the deadline passes during the claim, and attach fails before it registers; - `early claim stale ticket`: the deadline passes after the claim, before attach. - `vitest run plugin-runner-release-resume.integration.test.ts plugin-runner-eviction.integration.test.ts` (server, with real Runner processes from `cargo build --workspace --bins`): 2 files, 51 passed, none skipped. New modes: - `cached held claim` (Codex and Claude): the plugin resume is held, and the test proves Core's tool preparation runs while the claim is in progress; - `cached cancellation` (Codex): a cancellation after an early claim retires the Runner and the run stops cleanly. - `pnpm run typecheck:typescript` (runner) and `pnpm typecheck` (server): pass. - Evidence (modeled, not measured live): traces of 8 resumed runs on parked Runners in a hosted test deployment, on current `master`. The claim takes about 300 ms, and it fits inside backend setup in 7 of 8 runs. The modeled saving is about 285 ms per follow-up on a parked Runner at P50 (range 208–331 ms). I did not run this change on a hosted deployment. - Not run: the full server suite (CI runs it), smoke tests, and the hosted park, claim and attach loop with this change. ## Risks - **New failure mode:** Core now claims the parked Runner before backend setup ends. If that run then fails preparation, or is cancelled before attach, the existing retire path closes the claimed session. The Runner is retired, so that agent's next follow-up starts cold. Before this change, a preparation failure happened before the claim, while the Runner was still parked. - **Ordering:** the retained plugin task owner is rebound to the new run earlier, before Core creates tool authority. Nothing in the owner preparation reads that authority. A failure in tool authority setup now happens after the rebind, and the retire path handles it. - **Scope:** only an in-memory warm entry that was parked takes the early path. Cold launches, adopted Runners, remote warm recovery and restart recovery are unchanged. - **Deadline:** the claim deadline now starts earlier. A very slow backend setup can make the claim expire before attach. Attach then fails with `native_runner_resume_reconciliation_expired` before it registers, the same failure as a slow claim today. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, git, GitHub CLI) and subagents. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 --- .../src/backends/harness-driver-backend.ts | 8 +++ .../src/contracts/harness-driver.ts | 1 + .../src/contracts/native-session-backend.ts | 1 + .../src/drivers/codex/app-server-transport.ts | 1 + .../drivers/codex/codex-harness-session.ts | 5 ++ ...erd-codex-transport-release-resume.test.ts | 54 ++++++++++++++++--- .../src/live/runnerd-codex-transport.ts | 53 +++++++++++++----- .../native-runtime/native-session-executor.ts | 46 ++++++++++++---- ...-runner-release-resume.integration.test.ts | 34 ++++++++++-- 9 files changed, 171 insertions(+), 32 deletions(-) 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); } }