From 1391f6455185e94f87a31540d17e07c99028af59 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 6 Nov 2026 11:42:32 -0800 Subject: [PATCH] Add gated release/resume methods to plugin Runner owners (#19584) 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. > - Core owns Runner tasks through a lease-bound plugin owner. > - The SDK already defines release and resume in #19468. > - The owner still expects an accepted result for both operations, so it cannot use those SDK results. > - This pull request adds both methods behind the existing capability, warm-task and account checks. > - It uses the SDK parser to verify the result and task ID, and preserves the owner's cancellation checks. > - This prepares the owner boundary for later lifecycle integration. It does not start automatic release or resume. ## Linked Issues or Issue Description **Problem or motivation** The plugin Runner owner cannot forward the existing SDK release/resume pair. Its local result mapping rejects their valid results. Later lifecycle code needs both operations on the same checked task owner. **Proposed solution** Pass the existing description capability to the owner. Add explicit release and resume methods and use the SDK's operation-result parser. Require warm-task support and a selected account. Reject calls and late receipts after owner cancellation or detachment. **Alternatives considered** Adding automatic release here would also require connection quieting, settled-event drain, pending-ticket retry ownership and safe disposal. Keep those changes together in the next integration. Do not add a new flag, schema, timer or lifecycle mode here. Related: #19468. I checked ROADMAP.md and found no duplicate planned item. An open-PR title search for release/resume found no overlapping pull request. ## What Changed - Add the capability-gated owner facade for Codex and ACPX. Resume returns only `resumed` or `unavailable`. - Use the SDK parser for exact task-ID and operation-result validation. - Pass `releaseResumeVersion` from the checked description and test that it does not cause an automatic operation. - Cover account-preserving owner rebind, missing prerequisites, malformed results, ambiguous errors, cancellation and late replies. - Document the caller's remaining settlement, connection, ticket and attachment duties. ## Verification Passed locally on Node 26.12.0 with the pinned workspace dependencies: - In `server/`: `pnpm exec vitest run src/services/native-runtime/plugin-runner-task.test.ts` — 74 tests passed. - In `server/`: `pnpm exec vitest run src/services/native-runtime/native-session-executor.test.ts -t 'uses a task-only'` — 3 cases passed; 1,160 unrelated cases were not selected. - Built the real workspace SDK dependencies and Runner TypeScript. A focused strict TypeScript program for the owner and its tests passed. - `git diff --check` and the added-line secret/PII scan passed. The first root test command selected a nonexistent Vitest project name and ran no tests. Running the test from the server package fixed the command. The first focused TypeScript setup could not find Node types; pointing it to the package's installed type definitions fixed that setup. The full regression suite and release checks run in GitHub. ## Risks The facade has no automatic caller. Existing warm timeout, bootstrap, outbound connection and pool behavior stay in place. No plugin capability is advertised or task control channel enabled by this pull request. Future integration must serialize release with new work and disposal. It must quiet the owned connection, keep the exact pending ticket after an ambiguous reply, authenticate the source again and finish ordinary run attachment before new work. This change does not qualify provider parking, cold recovery, retention policy or deployment. ## Model Used OpenAI GPT-6.1 Sol, exact model ID `gpt-6.1-sol`, High reasoning. Used code execution and targeted tests. The context-window size was not exposed. ## 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 - [x] 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 Co-authored-by: Paperclip --- doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md | 15 +++ .../native-session-executor.test.ts | 9 +- .../native-runtime/native-session-executor.ts | 1 + .../native-runtime/plugin-runner-task.test.ts | 95 +++++++++++++++++++ .../native-runtime/plugin-runner-task.ts | 30 ++++-- 5 files changed, 140 insertions(+), 10 deletions(-) diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index 2c59e814b2b..987edb8386a 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -117,6 +117,21 @@ warm or source recovery capabilities do not imply support. Older strict Core schemas reject this description field, so coordinate advertisement with a compatible Core version. +Core's lease-bound task owner exposes capability-gated `release` and `resume` +methods for both Codex and ACPX. They require `releaseResumeVersion: 1`, warm-task +support and a selected account, and validate the operation result and exact task +ID with the SDK parser. These methods are a forwarding facade; Core does not yet +call them automatically or change its warm-session timeout policy. + +The caller must serialize settled release against new work and disposal, quiet +its owned connection's liveness checks and reconnect loop before release, and +retain the same pending ticket for ambiguous resume retries. Ordinary owner +rebind must first check the new run's account, credentials and project scope. +After resume, authenticate the source with the fresh ticket before warm readiness +and ordinary `run.attach`. An RPC error does not prove that release or resume had +no effect; it must not trigger reconnect, a replacement ticket or turn replay. +The facade does not supply this orchestration or enable provider retention. + ## Error codes Core does not read the text of a task error. To report a failure that Core can diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 47c09f6bde5..47e155341a4 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -15073,10 +15073,11 @@ describe("runnerd provider runtime wiring", () => { const driver = { prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration }; state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []); const readWorkspaceFile = vi.fn(async () => Buffer.from("proof")); - const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, isRetired: () => false, rebind:vi.fn() }); + 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 task = vi.fn(async (operation: unknown) => { environmentTaskOperationSchema.parse(operation); - return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; + return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, releaseResumeVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; }); const target = { kind: "remote", transport: "sandbox", providerKey: "plugin", leaseId: "plugin-task-lease", remoteCwd: "/unused", runnerTask: task, runnerTaskProjectIds: ["10000000-0000-4000-8000-000000000001"] } as const; @@ -15093,7 +15094,7 @@ describe("runnerd provider runtime wiring", () => { expect(state.toolAuthorityDefinitions).toHaveBeenCalledWith(expect.objectContaining({ assignedMcpTools })); expect(state.createAssignedMcpTools).toHaveBeenCalledWith(expect.objectContaining({ gatewayPublicId: "assigned-test", bearerToken: "fixture-mcp-secret" })); expect(owner).toHaveBeenCalledWith(expect.objectContaining({ task, projectIds: target.runnerTaskProjectIds, - credentialIdentity: undefined })); + credentialIdentity: undefined, releaseResumeVersion: 1 })); expect(state.createBackend.mock.calls.at(-1)![1]).toMatchObject({ workingDirectoryAuthority: "runner_prepared_launch", environment: {} }); state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); const options = state.createTransport.mock.calls.at(-1)![0]; @@ -15107,6 +15108,8 @@ describe("runnerd provider runtime wiring", () => { expect(options).not.toHaveProperty(key); } expect(task.mock.calls).toEqual([[{ kind: "describe" }]]); + expect(release).not.toHaveBeenCalled(); + expect(resume).not.toHaveBeenCalled(); const binding = state.toolAuthorityDefinitions.mock.calls.at(-1)![0] as unknown as { readRemoteApiFile?: (path: string, maxBytes: number) => Promise; readRemoteTaskFile?: (file: {contentRef: string;byteSize: number;sha256: string}) => Promise; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6e1833712bb..27f2abb0856 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -13685,6 +13685,7 @@ async function createRunnerdBackendWithinSessionClaim( // Warm retirement can outlive the run's temporary credential home; complete uses the durable handoff. onCredentials: input.execution.session.lifecyclePolicy.mode === "warm" ? undefined : credentials.onCredentials, warmTaskVersion: description.warmTaskVersion, + releaseResumeVersion: description.releaseResumeVersion, adoptIdentity: adoptionIdentity ?? (adopting && prior ? { runnerInstanceId: prior.runnerInstanceId, environmentLeaseId: prior.environmentLeaseId, runId: prior.runId, normalizedSessionId: prior.normalizedSessionId, diff --git a/server/src/services/native-runtime/plugin-runner-task.test.ts b/server/src/services/native-runtime/plugin-runner-task.test.ts index b9ac804b10d..0e866b013a8 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -529,6 +529,101 @@ describe("plugin native Runner ownership", () => { }); +describe("plugin Runner release/resume facade", () => { + const input = (harness: string) => ({ taskId: "task", projectIds: ["project"], harness, + warmTaskVersion: 1 as const, releaseResumeVersion: 1 as const, + accountIdentity: "selected-account", credentialIdentity: "selected-credential", + adoptIdentity: requirements.identity, authorityRunId: "run" }); + const lifecycleTask = () => vi.fn(async (operation: any): Promise => operation.kind === "release" + ? { kind: "released", taskId: "task" } + : { kind: "resume", taskId: "task", outcome: "resumed" }); + + it.each(["codex", "claude"])("forwards both %s operations through the current checked owner", async harness => { + const firstTask = lifecycleTask(); + const currentTask = lifecycleTask(); + const original = { ...input(harness), task: firstTask }; + const owner = createPluginRunnerTaskOwner(original); + await expect(owner.release(requirements.identity)).resolves.toBeUndefined(); + expect(firstTask).toHaveBeenCalledExactlyOnceWith({ kind: "release", identity: requirements.identity }, { expectedTaskId: "task" }); + expect(() => owner.rebind({ ...original, task: currentTask, authorityRunId: "next", accountIdentity: "changed" })).toThrow("cannot be rebound"); + expect(currentTask).not.toHaveBeenCalled(); + owner.rebind({ ...original, task: currentTask, authorityRunId: "next" }); + await expect(owner.resume(requirements.identity, "transient-ticket")).resolves.toBe("resumed"); + expect(currentTask).toHaveBeenCalledExactlyOnceWith({ kind: "resume", identity: requirements.identity, bootstrapTicket: "transient-ticket" }, { expectedTaskId: "task" }); + currentTask.mockResolvedValueOnce({ kind: "resume", taskId: "task", outcome: "unavailable" }); + await expect(owner.resume(requirements.identity, "transient-ticket")).resolves.toBe("unavailable"); + expect(firstTask).toHaveBeenCalledTimes(1); + }); + + it.each(["codex", "claude"])("requires capability, warm support and account for %s", async harness => { + for (const missing of ["releaseResumeVersion", "warmTaskVersion", "accountIdentity"] as const) { + const task = lifecycleTask(); + const owner = createPluginRunnerTaskOwner({ ...input(harness), task, [missing]: undefined }); + await expect(owner.release(requirements.identity)).rejects.toThrow("release/resume unavailable"); + await expect(owner.resume(requirements.identity, "transient-ticket")).rejects.toThrow("release/resume unavailable"); + expect(task).not.toHaveBeenCalled(); + } + }); + + it.each(["released", "detached", "cancelled"] as const)("does not dispatch after the owner is %s", async state => { + const task = provider().task; + const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); + const owner = createPluginRunnerTaskOwner({ ...input("codex"), task, executionContext: context }); + try { + if (state === "cancelled") context.abort(new Error("fixture cancellation")); + else await (await owner.registration({} as never, "run")).release(state === "detached" ? "detach" : undefined); + task.mockClear(); + await expect(owner.release(requirements.identity)).rejects.toThrow("operation cancelled"); + await expect(owner.resume(requirements.identity, "transient-ticket")).rejects.toThrow("operation cancelled"); + expect(task).not.toHaveBeenCalled(); + } finally { context.finish(); } + }); + + it.each(["release", "resume"] as const)("does not accept a late %s receipt after detachment", async operation => { + let reply!: (value: unknown) => void; + const task = vi.fn(() => new Promise(resolve => { reply = resolve; })); + const owner = createPluginRunnerTaskOwner({ ...input("codex"), task }); + const pending = operation === "release" ? owner.release(requirements.identity) : owner.resume(requirements.identity, "transient-ticket"); + await (await owner.registration({} as never, "run")).release("detach"); + reply(operation === "release" ? { kind: "released", taskId: "task" } : { kind: "resume", taskId: "task", outcome: "resumed" }); + await expect(pending).rejects.toThrow("operation cancelled"); + expect(task).toHaveBeenCalledTimes(1); + }); + + it("can receive unavailable after an observed natural exit", async () => { + const task = vi.fn(async (operation: any): Promise => operation.kind === "status" + ? { kind: "status", taskId: "task", phase: "completed", executionStopped: true, exitCode: 0 } + : { kind: "resume", taskId: "task", outcome: "unavailable" }); + 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"]); + }); + + it.each(["release", "resume"] as const)("validates exact task and closed %s result", async operation => { + const task = lifecycleTask(); + const owner = createPluginRunnerTaskOwner({ ...input("codex"), task }); + const invoke = () => operation === "release" ? owner.release(requirements.identity) : owner.resume(requirements.identity, "transient-ticket"); + for (const result of [ + { kind: operation === "release" ? "released" : "resume", taskId: "other-task", ...(operation === "resume" ? { outcome: "resumed" } : {}) }, + { kind: "accepted", taskId: "task" }, + { kind: "resume", taskId: "task", outcome: "unknown" }, + ]) { + task.mockResolvedValueOnce(result); + await expect(invoke()).rejects.toThrow(); + } + }); + + it.each(["release", "resume"] as const)("preserves ambiguous %s errors without another operation", async operation => { + const task = lifecycleTask(); + const owner = createPluginRunnerTaskOwner({ ...input("codex"), task }); + const error = new Error("fixture transport lost its reply"); + task.mockRejectedValueOnce(error); + await expect(operation === "release" ? owner.release(requirements.identity) : owner.resume(requirements.identity, "transient-ticket")).rejects.toBe(error); + expect(task).toHaveBeenCalledTimes(1); + }); +}); + describe("plugin task workspace file transfer", () => { it("reads only a submitted task, validates bytes, and rejects late responses after release", async () => { const {task}=provider(); const perform=task.getMockImplementation()!; diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index fa14bd0ea55..bf99a5ed958 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -1,7 +1,7 @@ import { measureRunPhase } from "../execution-preparation-timing.js"; import { createHash } from "node:crypto"; import { setTimeout as delay } from "node:timers/promises"; -import { environmentTaskOperationSchema, environmentTaskResultSchema, type PluginEnvironmentTaskOperation, type PluginEnvironmentTaskResult, type PluginEnvironmentTaskCredential } from "@paperclipai/plugin-sdk"; +import { environmentTaskOperationSchema, environmentTaskResultSchema, parseEnvironmentTaskResult, type PluginEnvironmentTaskOperation, type PluginEnvironmentTaskResult, type PluginEnvironmentTaskCredential } from "@paperclipai/plugin-sdk"; import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-connectivity"; import { RUNNER_PRP_PROTOCOL_MIN, RUNNER_PRP_PROTOCOL_MAX, type RunnerProcessHandle, type RunnerProcessLaunchSpec, type DurablePrpControlPlane, type RunnerExecutionDriver, type RunnerLaunchPreparer, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js"; @@ -47,6 +47,7 @@ export function createPluginRunnerTaskOwner(input: { onDispatched?: () => Promise | void; pollMs?: number; warmTaskVersion?: 1; + releaseResumeVersion?: 1; githubBroker?: { url: string; token: string }; adoptIdentity?: DurableRecoveryIdentity; }) { @@ -78,12 +79,8 @@ export function createPluginRunnerTaskOwner(input: { const call = async (operation: PluginEnvironmentTaskOperation) => { const phase = ({ prepare: "execution.plugin_prepare", check_project_auth: "execution.plugin_project_auth", submit: "execution.plugin_submit", connection: "execution.plugin_connection" } as const)[operation.kind as "prepare" | "check_project_auth" | "submit" | "connection"]; - const result = environmentTaskResultSchema.parse(await (phase + return parseEnvironmentTaskResult(operation, taskId, await (phase ? measureRunPhase(phase, () => input.task(operation, { expectedTaskId: taskId })) : input.task(operation, { expectedTaskId: taskId }))); - if (result.kind !== (operation.kind === "read_file" ? "file" : operation.kind === "describe" ? "description" : operation.kind === "prepare" ? "prepared" : operation.kind === "check_project_auth" ? "project_auth" : ["read_state", "archive_state"].includes(operation.kind) ? "state" : ["status", "connection", "credentials"].includes(operation.kind) ? operation.kind : "accepted")) { - throw new Error("Plugin Runner returned an invalid operation result"); - } - return result; }; let removeStartupAbort = () => {}; const bindStartupAbort = () => { @@ -366,7 +363,26 @@ export function createPluginRunnerTaskOwner(input: { createHash("sha256").update(bytes).digest("hex") !== result.sha256) throw new Error("Plugin Runner file integrity check failed"); return bytes; }; - return { launcher, registration, driver, readWorkspaceFile, + const requireReleaseResume = () => { + if (released || stopRequested || detached) throw new Error("Plugin Runner lifecycle operation cancelled"); + if (input.releaseResumeVersion !== 1 || input.warmTaskVersion !== 1 || !input.accountIdentity) { + throw new Error("Plugin Runner release/resume unavailable"); + } + }; + // The caller owns settlement, connection quieting and exact-ticket retries. + const release = async (identity: DurableRecoveryIdentity): Promise => { + requireReleaseResume(); + await call(environmentTaskOperationSchema.parse({ kind: "release", identity })); + requireReleaseResume(); + }; + const resume = async (identity: DurableRecoveryIdentity, bootstrapTicket: string): Promise<"resumed" | "unavailable"> => { + requireReleaseResume(); + const result = await call(environmentTaskOperationSchema.parse({ kind: "resume", identity, bootstrapTicket })); + requireReleaseResume(); + if (result.kind !== "resume") throw new Error("Plugin Runner resume result unavailable"); + 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,