diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index 32f05cc9b1d..2c59e814b2b 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -45,6 +45,8 @@ See [the schemas](../../packages/plugins/sdk/src/environment-tasks.ts) for all f | `credentials` | `credentials` | Return the selected account after process termination. | | `stop` | `accepted` | Stop the task and its descendants. Fence a delayed submission. | | `complete` | `accepted` | Release task resources after Core saves required results. | +| `release` | `released` | End active use of an exact, settled Runner session. Requires `releaseResumeVersion: 1`. | +| `resume` | `resume` | Request the same logical session with a fresh bootstrap ticket. Requires `releaseResumeVersion: 1`. | For a refused `submit` with `resumeFrom`, a driver can throw an error with `status: 409`, `code: "SESSION_STATE_UNAVAILABLE"`, and `reason` set to @@ -67,6 +69,54 @@ authorized project IDs, and optional launch limits. It can also contain a saved task ID for session reuse and a typed Codex or Claude credential. Core does not supply an executable command or a guest path. +## Release and resume + +`release` ends Core's active use of an exact, settled Runner session. Core calls +it only after the turn, required events, journal writes, tool results and turn +cleanup have settled. The provider verifies the six-field `identity` and refuses +stale, busy or unsettled requests. `released` does not cancel work or replace +`stop`, `complete`, `archive_state` or the allocation's `environmentReleaseLease`. +Hosts should retain reusable resources when capacity allows, but can reclaim +them under retention and capacity policy. Release guarantees neither retention +nor physical termination. The same contract applies to Codex and ACPX, with no +same-process-ID guarantee. + +`resume` requests that same logical session using the source `identity` and a +fresh `bootstrapTicket`. Both operations retain the outer request's authorized +company, environment, task, lease and run scope. Resume requires an authenticated +current claimant run and lease; reject a null `runId`. Verify the source, account, +workspace and current authority before installing the ticket or reserving reuse. +The ticket is bounded transient authentication input, as on submission. Never +copy its bytes into status, metadata, logs, results or durable replay records. + +The `resume` result has a closed `outcome`: + +- `resumed`: the same logical session is exclusively reserved for this claimant, + atomically against eviction and competing claims. Core obtains `connection`, + authenticates with the fresh ticket and performs ordinary `run.attach` with its + new identity. No new turn is permitted before attachment succeeds. +- `unavailable`: the host knows reuse is impossible, for example after eviction. + Core may select ordinary recovery, preserving its authorization, credential, + workspace and stop checks. This is not proof of physical termination. + +Source mismatches, authentication failures, conflicting claimants, transport +failures and ambiguous claims remain errors. They must not become `unavailable`, +authorize replacement, or silently create a new logical session. + +One release episode is allowed per exact source identity; ordinary `run.attach` +rotates authority before a later release. Release retries replay the original +result without releasing a resumed owner again. Resume retries with the same +source, claimant and ticket reconcile the same claim and original outcome. +Changed claimants or tickets conflict. A lost reply requires reconciliation, +not a replacement ticket or an assumption of eviction. Preserve replay fencing +after reclamation so delayed requests cannot change a newer owner. + +Advertise `releaseResumeVersion: 1` from `describe` only when both operations are +implemented. Core must check it before use. Absence keeps the existing lifecycle; +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. + ## Error codes Core does not read the text of a task error. To report a failure that Core can diff --git a/packages/plugins/sdk/src/environment-tasks.ts b/packages/plugins/sdk/src/environment-tasks.ts index 017a0d0fa92..3dd691d4423 100644 --- a/packages/plugins/sdk/src/environment-tasks.ts +++ b/packages/plugins/sdk/src/environment-tasks.ts @@ -113,12 +113,16 @@ export const environmentTaskOperationSchema = z.discriminatedUnion("kind", [ z.object({ kind: z.literal("credentials"), accountIdentity: z.string().min(1).max(1024) }).strict(), z.object({ kind: z.literal("status") }).strict(), z.object({ kind: z.literal("connection") }).strict(), + z.object({ kind: z.literal("release"), identity: durableIdentity }).strict(), + z.object({ kind: z.literal("resume"), identity: durableIdentity, bootstrapTicket: credentialValue }).strict(), z.object({ kind: z.literal("complete") }).strict(), z.object({ kind: z.literal("stop") }).strict(), ]); export const environmentTaskResultSchema = z.discriminatedUnion("kind", [ z.object({ kind: z.literal("accepted"), taskId: providerId }).strict(), + z.object({ kind: z.literal("released"), taskId: providerId }).strict(), + z.object({ kind: z.literal("resume"), taskId: providerId, outcome: z.enum(["resumed", "unavailable"]) }).strict(), z.object({ kind: z.literal("copilot_metadata"), taskId: providerId, metadata: z.record(z.string(), z.unknown()), executionStopped: z.literal(true) }).strict(), z.object({ kind: z.literal("project_auth"), taskId: providerId, configurationId: providerId.max(1024), @@ -135,6 +139,7 @@ export const environmentTaskResultSchema = z.discriminatedUnion("kind", [ z.object({ kind: z.literal("description"), taskId: providerId, runner: z.object({ version: z.string().min(1), digest: z.string().regex(/^sha256:[a-f0-9]{64}$/) }).strict(), warmTaskVersion: z.literal(1).optional(), + releaseResumeVersion: z.literal(1).optional(), launchPreparationVersion: z.literal(1).optional(), executionContextVersion: z.literal(1).optional(), /** Each version attests stopped-source recovery for its provider independently. */ @@ -204,7 +209,7 @@ export interface PluginEnvironmentTaskParams extends Omit Promise) { + const stdin = new PassThrough(), stdout = new PassThrough(), lines = createInterface({ input: stdout }); + const worker = startWorkerRpcHost({ stdin, stdout, plugin: definePlugin({ async setup() {}, onEnvironmentTask: handler }) }); + try { + const response = new Promise(resolve => lines.on("line", line => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) resolve(message); + })); + stdin.write(serializeMessage(createRequest("environmentTask", { companyId: "company", taskId: "task", + runId: "claimant-run", lease: { providerLeaseId: "task" }, operation }, "lifecycle"))); + return await response; + } finally { worker.stop(); lines.close(); stdin.destroy(); stdout.destroy(); } +} + +describe("release and resume task contract", () => { + it("keeps the capability optional and independent of warm and source recovery support", () => { + for (const capabilities of [{}, { releaseResumeVersion: 1 }, { warmTaskVersion: 1 }, + { codexSourceRecoveryVersion: 1 }, { acpxSourceRecoveryVersion: 1 }]) { + expect(environmentTaskResultSchema.parse({ ...description, ...capabilities })).toEqual({ ...description, ...capabilities }); + } + expect(environmentTaskResultSchema.safeParse({ ...description, releaseResumeVersion: 2 }).success).toBe(false); + }); + + it.each([release, resume])("requires the exact durable identity for $kind", operation => { + expect(environmentTaskOperationSchema.parse(operation)).toEqual(operation); + for (const field of Object.keys(identity)) { + expect(environmentTaskOperationSchema.safeParse({ ...operation, identity: { ...identity, [field]: undefined } }).success).toBe(false); + } + expect(environmentTaskOperationSchema.safeParse({ ...operation, identity: { ...identity, targetRunId: "other" } }).success).toBe(false); + expect(environmentTaskOperationSchema.safeParse({ ...operation, cancelActiveWork: true }).success).toBe(false); + }); + + it.each([undefined, "", "\0", "x".repeat(65_537)])("rejects a missing or invalid resume ticket", bootstrapTicket => { + expect(environmentTaskOperationSchema.safeParse({ ...resume, bootstrapTicket }).success).toBe(false); + }); + + it("matches each operation to its own closed result and task", () => { + const released = { kind: "released", taskId: "task" }; + expect(parseEnvironmentTaskResult(release, "task", released)).toEqual(released); + for (const outcome of ["resumed", "unavailable"]) { + const result = { kind: "resume", taskId: "task", outcome }; + expect(parseEnvironmentTaskResult(resume, "task", result)).toEqual(result); + expect(() => parseEnvironmentTaskResult(release, "task", result)).toThrow(); + } + for (const operation of [release, resume]) { + expect(() => parseEnvironmentTaskResult(operation, "task", { kind: "accepted", taskId: "task" })).toThrow(); + const result = operation.kind === "release" ? released : { kind: "resume", taskId: "task", outcome: "resumed" }; + expect(() => parseEnvironmentTaskResult(operation, "other-task", result)).toThrow(); + } + expect(() => parseEnvironmentTaskResult(resume, "task", { kind: "resume", taskId: "task", outcome: "unknown" })).toThrow(); + expect(environmentTaskResultSchema.safeParse({ ...released, retained: true }).success).toBe(false); + expect(environmentTaskResultSchema.safeParse({ kind: "resume", taskId: "task", outcome: "resumed", bootstrapTicket: "fixture-ticket" }).success).toBe(false); + }); + + it.each([release, resume])("dispatches $kind through the serialized worker RPC", async operation => { + const result: PluginEnvironmentTaskResult = operation.kind === "release" ? { kind: "released", taskId: "task" } + : { kind: "resume", taskId: "task", outcome: "resumed" }; + const response = await invoke(operation, async params => { + expect(params.operation).toEqual(operation); + expect(params.runId).toBe("claimant-run"); + return result; + }); + expect(response).toMatchObject({ result }); + }); + + it("passes known absence through RPC without treating it as a successful reservation", async () => { + expect(await invoke(resume, async () => ({ kind: "resume", taskId: "task", outcome: "unavailable" }))) + .toMatchObject({ result: { kind: "resume", taskId: "task", outcome: "unavailable" } }); + }); + + it.each(["source mismatch", "claim conflict", "authentication failed", "transport timeout"])("keeps %s as an RPC error", async message => { + expect(await invoke(resume, async () => { throw new Error(message); })).toHaveProperty("error"); + }); + + it("rejects cross-task and wrong-kind replies at the worker boundary", async () => { + for (const result of [{ kind: "resume", taskId: "other", outcome: "resumed" }, { kind: "accepted", taskId: "task" }] as PluginEnvironmentTaskResult[]) { + expect(await invoke(resume, async () => result)).toHaveProperty("error"); + } + }); + + it("preserves legacy describe and stop replies through RPC", async () => { + expect(await invoke({ kind: "describe" }, async () => description as PluginEnvironmentTaskResult)).toMatchObject({ result: description }); + const operation: PluginEnvironmentTaskOperation = { kind: "stop" }; + expect(await invoke(operation, async () => ({ kind: "accepted", taskId: "task" }))) + .toMatchObject({ result: { kind: "accepted", taskId: "task" } }); + }); +});