Pin plugin Runner cleanup to its physical task (#19300)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Plugin Runner owners manage a physical task through an environment
lease.
> - The existing replacement path can assign a new physical task to that
same lease.
> - A delayed old owner can then stop the replacement or mark it
stopped.
> - Cleanup must keep the original physical task identity across RPC and
database writes.
> - This pull request pins that identity and keeps cleanup available to
the new owner.

## Linked Issues or Issue Description

Refs #19134 and #19157. This fix is based on master and does not enable
repeated stopped-source recovery. A search found no duplicate fix.

**What happened?**

The maintained `createOwnedSandbox` replacement path changes
`providerLeaseId` on one lease. An old task owner uses that mutable
lease for a later Stop. A stopped-state response already in flight can
also update the new task's flags.

**Expected behavior**

The old owner must address its original task. Its late cleanup writes
must not alter the replacement. The replacement's own owner must still
stop and complete it.

**Steps to reproduce**

1. Create a plugin task owner for a stopped task.
2. Replace that task through the existing describe/startup path on the
same lease.
3. Resume an old checkpoint or a held old status response.
4. Observe a Stop for the replacement or an incorrect stopped flag
before this fix.

## What Changed

- Require each task owner to keep the physical ID returned by its
description. Reject a rebind to a different task.
- Carry the ID through the existing internal task context after startup
ends. Reject a changed physical task before RPC dispatch.
- Match the physical task in stopped-state, completion and
credential-handoff writes.
- Hold the exact lease row through the retained-session database update
and local checkpoint save.
- Cover the maintained replacement path with real PostgreSQL tests.
Preserve positive cleanup for the replacement owner. Document this
boundary.

## Verification

- Both replacement regressions failed on unmodified master `10a7d039c8`:
delayed old-owner Stop targeted the replacement, and a held old status
response marked it stopped.
- Six focused files passed 160 tests: task authority, task runtime, warm
tasks, execution target, credential writeback and task owner.
- Four selected executor factory cases passed. They cover task-only
providers and the physical task binding after the startup context ends.
- Runner TypeScript, plugin SDK and adapter-utils builds passed. Server
TypeScript check passed. Diff and added-line secret scans passed.
- All 15 setup-probe cases passed with the actual Runner and fake
provider binaries. The first local attempt lacked cargo on PATH. A
sandboxed native attempt timed out in seven process cases; the unchanged
run with loopback permission passed all 15. A separate initial DB
integration attempt was denied a loopback listener; its
permission-correct run passed.
- Review found three missed constructors in the existing SDK/Core
lifecycle fixture. They now use the actual describe result. All four
cases in that real PostgreSQL and worker RPC file passed.
- The first GitHub attempt failed in the unchanged GitHub recovery
pagination case in `chat-channels.integration.test.ts`: its zero-request
assertion observed a configuration GET. That exact case passed locally
without code or assertion changes. The one failed-job-only retry passed;
the initial failure log is retained.
- [GitHub workflow
42082587576](https://github.com/paperclipai/paperclip/actions/runs/42082587576)
passed on its second attempt. All CI checks completed, and Snyk passed.
Standard Greptile scored exact head `fb914d98b99e` 5/5 with no
unresolved threads. The branch is mergeable and ready for review.

## Risks

The task context change is internal. It adds no plugin wire field,
schema or Runner mode. A stale owner now gets an error and cannot clean
up a replacement. A new owner can still clean up that task. The retained
checkpoint write holds one existing lease row during a bounded local
file write.

Repeated stopped-source incarnation admission, runtime pins and
deployments are outside this change.

## Model Used

OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning. Used repository
tools, TypeScript and real PostgreSQL tests.

## 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 described the issue in-PR using the bug template
- [x] I have not referenced internal or instance-local Paperclip issues
or links
- [x] My branch name describes the change
- [x] I have run the focused tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation
- [x] I have considered and documented risks
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no unresolved findings
- [x] I will address reviewer comments before requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

server/src/__tests__/environment-execution-target.test.ts

@@ -104,6 +104,17 @@ describe("resolveEnvironmentExecutionTarget", () => {
     delete process.env.PAPERCLIP_RUNTIME_API_URL;
   });
 
+  it("forwards the physical task binding through the plugin task target", async () => {
+    const task = vi.fn(async () => ({ kind: "accepted" as const, taskId: "physical-task" }));
+    const target = await resolveEnvironmentExecutionTarget({ db: {} as never, companyId: "company", adapterType: "paperclip_runner",
+      environment: { id: "environment", driver: "plugin", config: {} }, leaseMetadata: {},
+      lease: { id: "lease", companyId: "company", environmentId: "environment", providerLeaseId: "physical-task", metadata: {} } as never,
+      environmentRuntime: { supportsTasks: async () => true, task } as never });
+    if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runnerTask) throw new Error("Plugin task target unavailable");
+    await target.runnerTask({ kind: "stop" }, { expectedTaskId: "physical-task" });
+    expect(task).toHaveBeenCalledWith({ companyId: "company", leaseId: "lease", operation: { kind: "stop" }, expectedTaskId: "physical-task" });
+  });
+
   it("uses a bounded default cwd for sandbox targets when lease metadata omits remoteCwd", async () => {
     mockResolveEnvironmentDriverConfigForRuntime.mockResolvedValue({
       driver: "sandbox",

server/src/__tests__/environment-task-authority.integration.test.ts

@@ -6,6 +6,7 @@ import { resolvePluginTaskResume } from "../services/environment-task-runtime.js
 import { remoteTerminationReceipt } from "../services/remote-execution-termination.js";
 import { environmentRuntimeService } from "../services/environment-runtime.js";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
+import { createPluginRunnerTaskOwner } from "../services/native-runtime/plugin-runner-task.js";
 import { NativeExecutionContext } from "../services/native-runtime/native-execution-context.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 
@@ -21,7 +22,7 @@ databaseSuite("environment task lease authority", () => {
   });
   afterAll(async () => { await stopDb?.(); });
 
-  it("replaces a stopped warm candidate within its tracked startup context", async () => {
+  it.each(["dispatch", "writeback"])("fences stale owner %s through maintained physical task replacement", async mode => {
     const companyId = randomUUID(), agentId = randomUUID(), environmentId = randomUUID(), pluginId = randomUUID(), runId = randomUUID(), leaseId = randomUUID();
     await db.insert(companies).values({ id: companyId, name: "Warm candidate", issuePrefix: `T${companyId.slice(0, 7).replace(/-/g, "").toUpperCase()}` });
     await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "paperclip_runner" });
@@ -33,10 +34,17 @@ databaseSuite("environment task lease authority", () => {
     await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, heartbeatRunId: runId,
       provider: "plugin:fixture:tasks", providerLeaseId: "stopped-task", status: "active", leasePolicy: "reuse_by_environment",
       metadata: { driver: "plugin", pluginId, pluginKey: pluginId, driverKey: "tasks", runnerTaskLease: true, runnerTaskReused: true } });
-    const calls: string[] = [];
+    const calls: string[] = [], stoppedTasks: string[] = [];
+    let holdStatus = false, releaseStatus!: () => void, enterStatus!: () => void;
+    const statusEntered = new Promise<void>(resolve => { enterStatus = resolve; });
+    const statusReleased = new Promise<void>(resolve => { releaseStatus = resolve; });
     const workers = { getWorker: () => ({ supportedMethods: ["environmentTask", "environmentAcquireLease"] }),
       call: async (_plugin: string, method: string, params: any, _timeout?: number, _sink?: unknown, settled?: () => void) => {
         calls.push(method === "environmentTask" ? params.operation.kind : method);
+        if (method === "environmentTask" && params.operation.kind === "stop") stoppedTasks.push(params.taskId);
+        if (method === "environmentTask" && params.operation.kind === "status" && holdStatus) {
+          holdStatus = false; enterStatus(); await statusReleased;
+        }
         try {
           if (method === "environmentAcquireLease") return { providerLeaseId: "replacement-task" };
           if (params.operation.kind === "status") return { kind: "status", taskId: params.taskId, phase: "completed", executionStopped: true };
@@ -47,16 +55,36 @@ databaseSuite("environment task lease authority", () => {
       },
     } as unknown as PluginWorkerManager;
     const context = new NativeExecutionContext({ binding: { companyId, agentId, runId } } as never, 1_000);
+    const runtime = environmentRuntimeService(db, { pluginWorkerManager: workers });
+    const task = (operation: unknown, options?: { expectedTaskId?: string }) => runtime.task({ companyId, leaseId,
+      operation: operation as never, ...options });
+    const identity = { runnerInstanceId: "runner", environmentLeaseId: leaseId, runId, normalizedSessionId: "session", turnId: "turn", itemId: "item" };
+    const oldOwner = createPluginRunnerTaskOwner({ taskId: "stopped-task", task, projectIds: [], harness: "codex", adoptIdentity: identity });
+    const oldRegistration = await oldOwner.registration({} as never, runId);
+    holdStatus = mode === "writeback";
+    const lateStatus = holdStatus ? task({ kind: "status" }, { expectedTaskId: "stopped-task" }) : Promise.resolve();
+    if (holdStatus) await statusEntered;
     try {
       await expect(environmentRuntimeService(db, { pluginWorkerManager: workers }).task({ companyId, leaseId,
         operation: { kind: "describe" }, executionContext: context })).resolves.toMatchObject({ kind: "description", taskId: "replacement-task" });
-      expect(calls).toEqual(["status", "status", "complete", "environmentAcquireLease", "describe"]);
+      expect(calls).toEqual([...(mode === "writeback" ? ["status"] : []), "status", "status", "complete", "environmentAcquireLease", "describe"]);
       expect(context.pending).toBe(false);
       const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId));
       expect(lease.providerLeaseId).toBe("replacement-task");
       expect(lease.metadata?.runnerTaskReused).toBe(false);
       expect(JSON.stringify(lease.metadata)).not.toContain(context.id);
-    } finally { context.finish(); }
+      releaseStatus(); await lateStatus;
+      const [afterLateStatus] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId));
+      expect(afterLateStatus.metadata?.runnerTaskExecutionStopped).toBe(false);
+      const failedCleanup = await oldRegistration.checkpoint("unsettled").then(() => undefined, error => error);
+      expect(stoppedTasks).not.toContain("replacement-task");
+      expect(failedCleanup).toBeInstanceOf(Error);
+      const nextOwner = createPluginRunnerTaskOwner({ taskId: "replacement-task", task, projectIds: [], harness: "codex", adoptIdentity: identity });
+      const nextRegistration = await nextOwner.registration({} as never, runId);
+      try { await expect(nextRegistration.checkpoint("unsettled")).resolves.toBeUndefined(); }
+      finally { await nextRegistration.release("detach"); }
+      expect(stoppedTasks).toEqual(["replacement-task"]);
+    } finally { releaseStatus(); await lateStatus.catch(() => undefined); context.finish(); await oldRegistration.release("detach"); }
   });
 
   it.each(["late", "post-result", "retiring"])("queues a replacement instead of attaching after %s startup cancellation", async mode => {

server/src/__tests__/plugin-runner-task.integration.test.ts

@@ -8,7 +8,7 @@ import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environ
 import { definePlugin } from "../../../packages/plugins/sdk/src/define-plugin.js";
 import { startWorkerRpcHost } from "../../../packages/plugins/sdk/src/worker-rpc-host.js";
 import { createRequest, parseMessage, serializeMessage, isJsonRpcResponse, isJsonRpcSuccessResponse } from "../../../packages/plugins/sdk/src/protocol.js";
-import { environmentTaskOperationSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk";
+import { environmentTaskOperationSchema, environmentTaskResultSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk";
 import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
 import { agentEnvironmentTestService } from "../services/agent-environment-test.js";
 import { heartbeatService } from "../services/heartbeat.js";
@@ -191,11 +191,11 @@ databaseSuite("Core plugin Runner task lifecycle", () => {
     expect(target).toMatchObject({ kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: acquired.lease.id });
     if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runnerTask) throw new Error("Missing plugin target");
     expect(target.runner).toBeUndefined();
-    await expect(target.runnerTask({ kind: "describe" }))
-      .resolves.toMatchObject({ kind: "description", runner: { version: "fixture" } });
+    const description = environmentTaskResultSchema.parse(await target.runnerTask({ kind: "describe" }));
+    expect(description).toMatchObject({ kind: "description", runner: { version: "fixture" } });
     const onDispatched = vi.fn(), checkpoint = vi.fn(async () => {}), close = vi.fn(async () => {});
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close });
-    const owner = createPluginRunnerTaskOwner({ task: target.runnerTask, projectAuthCheckVersion: 1,
+    const owner = createPluginRunnerTaskOwner({ taskId: description.taskId, task: target.runnerTask, projectAuthCheckVersion: 1,
       credentialIdentity: "revision", accountIdentity: "account", credential: { type: "codex-auth-json", authJson: '{"OPENAI_API_KEY":"synthetic-account-key"}' }, projectIds: target.runnerTaskProjectIds ?? [], harness: "codex",
       onSessionRetained: checkpoint, onDispatched, pollMs: 1 });
     const identity = { runnerInstanceId: randomUUID(), environmentLeaseId: acquired.lease.id, runId,
@@ -244,7 +244,7 @@ databaseSuite("Core plugin Runner task lifecycle", () => {
         reattachRemoteLease: { kind: "plugin_task", leaseId: saved.id },
       });
       expect(recovered.lease.id).toBe(saved.id);
-      const adopted = createPluginRunnerTaskOwner({ task: target.runnerTask, projectIds: [projectId], harness: "codex",
+      const adopted = createPluginRunnerTaskOwner({ taskId: description.taskId, task: target.runnerTask, projectIds: [projectId], harness: "codex",
         projectAuthCheckVersion: 1, adoptIdentity: identity });
       await adopted.driver.authorizeRunnerAdoption!(prepared.configurationId, identity);
       const adoptedRegistration = await adopted.registration({} as never, runId);
@@ -254,7 +254,7 @@ databaseSuite("Core plugin Runner task lifecycle", () => {
       expect(calls.slice(callsBefore).some(op => ["prepare", "submit", "stop", "complete"].includes(op.kind))).toBe(false);
       expect(executionStopped).toBe(false);
       projectAuthAllowed = false;
-      const revoked = createPluginRunnerTaskOwner({ task: target.runnerTask, projectIds: [projectId], harness: "codex",
+      const revoked = createPluginRunnerTaskOwner({ taskId: description.taskId, task: target.runnerTask, projectIds: [projectId], harness: "codex",
         projectAuthCheckVersion: 1, adoptIdentity: identity });
       const connectionsBeforeRevocation = calls.filter(op => op.kind === "connection").length;
       await expect(revoked.driver.authorizeRunnerAdoption!(prepared.configurationId, identity)).rejects.toThrow("project authentication conflicts");

server/src/__tests__/plugin-task-credential-writeback.test.ts

@@ -8,7 +8,7 @@ import { agents, companies, companyMemberships, createDb, environmentLeases, env
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { aiConnectionService } from "../services/ai-connections.js";
 import { prepareManagedAiRuntime } from "../services/ai-connection-runtime.js";
-import { registerPluginTaskCredentialHandoff } from "../services/plugin-task-credential-writeback.js";
+import { registerPluginTaskCredentialHandoff, settlePluginTaskCredentialHandoff } from "../services/plugin-task-credential-writeback.js";
 import { executeEnvironmentTask, settleDispatchedEnvironmentTask } from "../services/environment-task-runtime.js";
 import { secretService } from "../services/secrets.js";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
@@ -123,6 +123,14 @@ suite("durable plugin task credential handoff", () => {
     expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
     expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete" });
   });
+  it("does not mark a replacement task's identical credential handoff complete", async () => {
+    const f = await fixture(false), prior = await f.lease();
+    await db.update(environmentLeases).set({ providerLeaseId: "replacement-task" }).where(eq(environmentLeases.id, f.leaseId));
+    await expect(settlePluginTaskCredentialHandoff(db, { companyId: f.companyId, leaseId: f.leaseId, expectedTaskId: "fixture-task" },
+      prior.metadata!, async () => { throw new Error("Undispatched task must not export credentials"); })).rejects.toThrow("handoff changed");
+    expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" });
+  });
+
   it("retires an undispatched task without exporting credentials and blocks a late submit", async () => {
     const f = await fixture(false);
     expect(await f.settle()).toBe(false);