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,