feat(plugins): define settled task release and resume (#19468)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Environment plugins provide Runner tasks through the public SDK.
> - A settled turn can leave resources that a later run can reuse.
> - Core needs a shared contract to end active use and request the same
logical session later.
> - The host must retain control of capacity and report when reuse is
unavailable.
> - This pull request adds an optional release and resume contract with
strict validation and documentation.
> - The benefit is a common boundary for later host and Core integration
across Codex and ACPX.

## Linked Issues or Issue Description

**Subsystem affected**

packages/plugins — plugin SDK and its environment task documentation.

**Problem or motivation**

The SDK has stop and completion operations. It has no explicit contract
to release active use after a settled turn and later request the same
logical session without promising retention or the same process.

**Proposed solution**

Add a versioned optional capability, release for the identified session,
and resume with fresh transient authentication. Define two explicit
resume outcomes: resumed or unavailable. Keep authorization, safe
retries and ordinary attachment requirements explicit.

**Alternatives considered**

A provider-specific command would couple Core to one host. Reusing stop,
completion or archive would change their existing meaning. A new
checkpoint format or Runner mode is unnecessary.

**Roadmap alignment**

This extends the existing plugin and sandbox boundaries in ROADMAP.md.
It adds no Core caller, provider implementation or runtime activation.

**Additional context**

The duplicate search found #19371, which fixes Core failure cleanup for
a session resume owner. Its scope is separate. Related recovery work in
#19157 and admission work in #19196 remain unchanged.

## What Changed

- Add strict release and resume operation/result schemas and optional
releaseResumeVersion: 1.
- Reuse the existing task identity, inferred SDK exports and worker RPC
types.
- Define settled release, host retention policy, atomic reuse
reservation, fresh authentication, retries that cannot release a newer
owner and ordinary recovery.
- Add schema and serialized worker RPC tests, including legacy replies
and failure cases.

## Verification

- SDK typecheck: `node_modules/.bin/tsc -p
packages/plugins/sdk/tsconfig.json --noEmit` passed.
- From `packages/plugins/sdk`, `../../../node_modules/.bin/vitest run
--config vitest.config.ts tests/environment-task-release-resume.test.ts
tests/environment-tasks.test.ts tests/environment-task-error.test.ts
tests/environment-task-error-code.test.ts tests/worker-rpc-host.test.ts`
passed: 94 tests across five suites.
- `git diff --check` passed.
- [Full GitHub
CI](https://github.com/paperclipai/paperclip/actions/runs/42238743059)
passed on `a55376801c91af29b68a13304d085706e4d7ed6f`. Standard Greptile
scored that head 5/5 with no comments or unresolved review threads.
- Full regressions run in GitHub CI, per the requested verification
scope. No VM, deployment or live provider test was run. These tests
prove the SDK boundary, not host retention or live attachment.

## Risks

- This is an optional contract. Runtime behavior stays unchanged until
later host and Core work uses it.
- Older strict Core schemas reject the new description field. Providers
must coordinate capability advertisement with a compatible Core.
- Host implementations must enforce settlement, exact authority,
exclusive claims and safe retries. Unknown outcomes remain errors. An
unavailable result never proves physical termination.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning, with repository
inspection, code editing and local test tools. Context window size was
not exposed by the runtime.

## 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 <noreply@paperclip.ing>

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

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<PluginEnvironmentDrive
 /** An accepted stop is not proof of process termination. */
 export function parseEnvironmentTaskResult(operation: PluginEnvironmentTaskOperation, taskId: string, value: unknown): PluginEnvironmentTaskResult {
   const result = environmentTaskResultSchema.parse(value);
-  const expected = 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", "copilot_metadata"].includes(operation.kind) ? operation.kind : "accepted";
+  const expected = operation.kind === "release" ? "released" : operation.kind === "resume" ? "resume" : 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", "copilot_metadata"].includes(operation.kind) ? operation.kind : "accepted";
   if (result.taskId !== taskId || result.kind !== expected) throw new Error("Invalid plugin-provided Runner execution response");
   return result;
 }

packages/plugins/sdk/tests/environment-task-release-resume.test.ts

@@ -0,0 +1,108 @@
+import { createInterface } from "node:readline";
+import { PassThrough } from "node:stream";
+import { describe, expect, it } from "vitest";
+import { definePlugin } from "../src/define-plugin.js";
+import {
+  environmentTaskOperationSchema,
+  environmentTaskResultSchema,
+  parseEnvironmentTaskResult,
+  type PluginEnvironmentTaskOperation,
+  type PluginEnvironmentTaskParams,
+  type PluginEnvironmentTaskResult,
+} from "../src/environment-tasks.js";
+import { createRequest, isJsonRpcResponse, parseMessage, serializeMessage, type JsonRpcResponse } from "../src/protocol.js";
+import { startWorkerRpcHost } from "../src/worker-rpc-host.js";
+
+const identity = { runnerInstanceId: "runner", environmentLeaseId: "lease", runId: "source-run",
+  normalizedSessionId: "session", turnId: "turn", itemId: "item" };
+const release = { kind: "release", identity } as const;
+const resume = { kind: "resume", identity, bootstrapTicket: "fixture-ticket" } as const;
+const description = { kind: "description", taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
+
+async function invoke(operation: unknown, handler: (params: PluginEnvironmentTaskParams) => Promise<PluginEnvironmentTaskResult>) {
+  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<JsonRpcResponse>(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" } });
+  });
+});