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>

doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md

@@ -157,6 +157,11 @@ An accepted stop does not prove termination. Set `executionStopped: true`
 only after the Runner and all descendants have stopped. A terminal phase alone
 does not permit Core to release execution authority.
 
+Core binds each Runner owner to the physical task ID returned by its description.
+An old owner cannot follow a changed lease to stop or complete its replacement.
+Late stopped-state, retained-session and credential-handoff writes must still
+match that same physical task. A new owner can clean up the replacement.
+
 Core fences new submissions when cleanup starts. It stops the saved task,
 checks termination, saves credential results, and completes the task before it
 releases or destroys the environment lease. An uncertain result leaves a saved

packages/adapter-utils/src/execution-target.ts

@@ -185,8 +185,8 @@ export interface AdapterSandboxExecutionTarget extends AdapterExecutionTargetWor
   kind: "remote";
   transport: "sandbox";
   providerKey?: string | null;
-  /** Host-bound plugin task RPC. Both sides validate the wire contract. */
+  /** Host-bound plugin task RPC. expectedTaskId pins an owner to its physical task after startup. */
-  runnerTask?: (operation: unknown, context?: { executionContext?: unknown }) => Promise<unknown>;
+  runnerTask?: (operation: unknown, context?: { executionContext?: unknown; expectedTaskId?: string }) => Promise<unknown>;
   runnerTaskProjectIds?: readonly string[];
   /** Private Core-issued identity for a bounded, tool-free setup probe. */
   runnerTaskProbe?: { runId: string; sessionId: string };

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" }))
+    const description = environmentTaskResultSchema.parse(await target.runnerTask({ kind: "describe" }));
-      .resolves.toMatchObject({ kind: "description", runner: { version: "fixture" } });
+    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);

server/src/services/environment-execution-target.ts

@@ -277,7 +277,8 @@ export async function resolveEnvironmentExecutionTarget(input: {
       runnerTask: async (operation, context) => {
         const result = await runtime.task({ companyId: input.companyId, leaseId: lease.id,
           operation: environmentTaskOperationSchema.parse(operation),
-          ...(context?.executionContext ? { executionContext: context.executionContext } : {}) });
+          ...(context?.executionContext ? { executionContext: context.executionContext } : {}),
+          ...(context?.expectedTaskId !== undefined ? { expectedTaskId: context.expectedTaskId } : {}) });
         if (result.kind === "description" && result.taskId !== acquisition.providerLeaseId) {
           acquisition.outcome = "replacement";
           acquisition.providerLeaseId = result.taskId;

server/src/services/environment-runtime.ts

@@ -3787,7 +3787,7 @@ export function environmentRuntimeService(
     },
 
     /** Dispatch plugin-provided Runner execution, including resource preparation and task lifecycle. */
-    async task(input: { companyId: string; leaseId: string; operation: PluginEnvironmentTaskOperation; executionContext?: unknown }) {
+    async task(input: { companyId: string; leaseId: string; operation: PluginEnvironmentTaskOperation; executionContext?: unknown; expectedTaskId?: string }) {
       if (!options.pluginWorkerManager) throw new Error("Plugin-provided Runner execution worker manager unavailable");
       return executeEnvironmentTask(db, options.pluginWorkerManager, input);
     },

server/src/services/environment-task-runtime.ts

@@ -35,6 +35,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
   leaseId: string;
   operation: PluginEnvironmentTaskOperation;
   executionContext?: unknown;
+  /** Physical task captured by an existing owner, independent of the mutable lease. */
+  expectedTaskId?: string;
 }) {
   const operation = environmentTaskOperationSchema.parse(input.operation);
   const [row] = await db.select({ lease: environmentLeases, environment: environments, run: heartbeatRuns })
@@ -52,6 +54,7 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
     context.assertScope({ companyId: input.companyId, agentId: run.agentId, runId: run.id });
   }
   const taskId = row.lease.providerLeaseId;
+  if (input.expectedTaskId !== undefined && input.expectedTaskId !== taskId) throw new Error("Plugin Runner physical task changed");
   const metadata = lease.metadata ?? {};
   const probe = metadata.runnerTaskProbe as { runId?: unknown; sessionId?: unknown; expiresAtMs?: unknown } | undefined;
   const activeProbe = lease.heartbeatRunId === null && lease.issueId === null && probe
@@ -338,8 +341,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
       }
     }
     if (operation.kind === "complete") {
-      await settlePluginTaskCredentialHandoff(db, { companyId: input.companyId, leaseId: lease.id }, metadata, async accountIdentity => {
+      await settlePluginTaskCredentialHandoff(db, { companyId: input.companyId, leaseId: lease.id, expectedTaskId: taskId }, metadata, async accountIdentity => {
-        const snapshot = await executeEnvironmentTask(db, workers, { ...input, operation: { kind: "credentials", accountIdentity } });
+        const snapshot = await executeEnvironmentTask(db, workers, { ...input, expectedTaskId: taskId, operation: { kind: "credentials", accountIdentity } });
         if (snapshot.kind !== "credentials") throw new Error("Plugin Runner credential snapshot unavailable");
         return snapshot.credential;
       });
@@ -405,7 +408,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
         metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ runnerTaskExecutionStopped: true,
           runnerTaskSessionRetained: parsed.kind === "status" && parsed.sessionStateAvailable === true,
           ...(parsed.kind === "status" && parsed.preparationStopped === true ? { runnerTaskPreparationStopped: true } : {}) })}::jsonb`,
-      }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId)));
+      }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId),
+        eq(environmentLeases.providerLeaseId, taskId)));
     }
     if (operation.kind === "complete") {
       if (freshWorkspace && metadata.runnerTaskExecutionStopped === true)
@@ -420,13 +424,13 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
           await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({
             ...(owner.status === "retained" ? {status:"released" as const,releasedAt:new Date(),cleanupStatus:"success" as const}:{}),
             metadata:sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({remoteExecutionTermination:receipt,runnerTaskExecutionStopped:true})}::jsonb`,
-          }).where(eq(environmentLeases.id,owner.id)), { companyId: owner.companyId, runId: owner.heartbeatRunId ?? undefined });
+          }).where(and(eq(environmentLeases.id,owner.id),eq(environmentLeases.providerLeaseId,taskId))), { companyId: owner.companyId, runId: owner.heartbeatRunId ?? undefined });
         }
       }
       await db.update(environmentLeases).set({
         metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskDispatched":false}'::jsonb`,
       }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId),
-        sql`${environmentLeases.metadata}->>'runnerTaskExecutionStopped' = 'true'`));
+        eq(environmentLeases.providerLeaseId, taskId), sql`${environmentLeases.metadata}->>'runnerTaskExecutionStopped' = 'true'`));
     }
     return parsed;
   } catch {

server/src/services/native-runtime/native-session-executor.test.ts

@@ -13430,6 +13430,33 @@ describe("runnerd provider runtime wiring", () => {
     return state.createTransport.mock.calls[0]![0].stateDirectory!;
   }
 
+  it("keeps physical task cleanup binding after the startup context finishes", async () => {
+    const current = scopedNativeExecution("physical-task-cleanup");
+    const context = new NativeExecutionContext(current, 1_000);
+    const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner");
+    const db = leaseDb(current), update = db.update.bind(db);
+    db.update = ((table: unknown) => table === environmentLeases
+      ? { set: () => ({ where: () => ({ returning: async () => [] }) }) }
+      : update(table as never)) as typeof db.update;
+    db.transaction = (async callback => callback(db as never)) as typeof db.transaction;
+    const task = vi.fn(async (_operation: unknown, _options?: { executionContext?: unknown; expectedTaskId?: string }) => ({
+      kind: "description", taskId: "physical-task", launchPreparationVersion: 1, projectAuthCheckVersion: 1,
+      runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` },
+    }));
+    try {
+      await createRunnerdBackend({ db, execution: current, runnerInstanceId: "physical-task-runner",
+        executionContext: context, runnerIngressAuthorized: true, runnerExecutionTarget: {
+          kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: "lease", runnerTask: task,
+        } });
+      const input = owner.mock.calls.at(-1)![0];
+      expect(input.taskId).toBe("physical-task");
+      context.finish();
+      await input.task({ kind: "stop" }, { expectedTaskId: "physical-task" });
+      expect(task).toHaveBeenLastCalledWith({ kind: "stop" }, { expectedTaskId: "physical-task" });
+      await expect(input.onSessionRetained!({})).rejects.toThrow("physical task changed");
+    } finally { context.finish(); owner.mockRestore(); }
+  });
+
   it("rejects a late backend factory after its startup context has finished", async () => {
     const current = scopedNativeExecution("late-context-factory");
     state.createBackend.mockClear();

server/src/services/native-runtime/native-session-executor.ts

@@ -13257,10 +13257,11 @@ async function createRunnerdBackendWithinSessionClaim(
   const cloudArtifactRoot = input.execution.provider.kind === "cursor_cloud" ? join(runnerdStateRoot(input.execution), "cursor-cloud", "artifacts") : undefined;
   if (input.execution.provider.kind === "cursor_cloud" && remoteTarget) throw new Error("cursor_cloud_local_controller_required");
   const rawPluginTask = remoteTarget?.transport === "sandbox" ? remoteTarget.runnerTask : undefined;
-  const pluginTask = rawPluginTask && input.executionContext ? (operation: unknown) => {
+  const pluginTask = rawPluginTask && input.executionContext ? (operation: unknown, options?: { expectedTaskId?: string }) => {
     const startupOperation = ["describe", "prepare", "check_project_auth", "connection"].includes(String(record(operation).kind));
     if (startupOperation) input.executionContext?.signal.throwIfAborted();
-    return rawPluginTask(operation, input.executionContext?.active && startupOperation ? { executionContext: input.executionContext } : undefined);
+    return rawPluginTask(operation, { ...options,
+      ...(input.executionContext?.active && startupOperation ? { executionContext: input.executionContext } : {}) });
   } : rawPluginTask;
   let runnerControlPlane: DurablePrpControlPlane | undefined;
   if (pluginTask && input.runnerIngressAuthorized !== true) throw new Error("runner_ingress_unavailable");
@@ -13493,7 +13494,7 @@ async function createRunnerdBackendWithinSessionClaim(
       await registerPluginTaskCredentialHandoff(input.db, { companyId: input.execution.binding.companyId, leaseId: remoteTarget.leaseId },
         credentials.accountIdentity, input.managedAiCredentialWriteback);
     }
-    const ownerInput: Parameters<typeof createPluginRunnerTaskOwner>[0] = { ...credentials,
+    const ownerInput: Parameters<typeof createPluginRunnerTaskOwner>[0] = { ...credentials, taskId: description.taskId,
       executionContext: input.executionContext?.pluginSupported ? input.executionContext : undefined,
       // 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,
@@ -13511,14 +13512,19 @@ async function createRunnerdBackendWithinSessionClaim(
       } : undefined,
       credentialIdentity: input.managedAiCredentialIdentity,
       onSessionRetained: async state => {
-        savePluginRunnerCheckpoint(root, state, state);
+        await input.db.transaction(async tx => {
-        await input.db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({
+          const saved = await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({
-          runnerTaskSessionRetained: true, runnerTaskIdentity: {
+            runnerTaskSessionRetained: true, runnerTaskIdentity: {
-            runnerId: state.runnerInstanceId, leaseId: state.environmentLeaseId,
+              runnerId: state.runnerInstanceId, leaseId: state.environmentLeaseId,
-            runId: state.runId, sessionId: state.normalizedSessionId, turnId: state.turnId, itemId: state.itemId,
+              runId: state.runId, sessionId: state.normalizedSessionId, turnId: state.turnId, itemId: state.itemId,
-          },
+            },
-        })}::jsonb` })
+          })}::jsonb` })
-          .where(and(eq(environmentLeases.id, remoteTarget.leaseId!), eq(environmentLeases.companyId, input.execution.binding.companyId)));
+            .where(and(eq(environmentLeases.id, remoteTarget.leaseId!), eq(environmentLeases.companyId, input.execution.binding.companyId),
+              eq(environmentLeases.providerLeaseId, description.taskId))).returning({ id: environmentLeases.id });
+          if (saved.length !== 1) throw new Error("Plugin Runner physical task changed");
+          // Keep the lease row locked until its local checkpoint has been saved.
+          savePluginRunnerCheckpoint(root, state, state);
+        });
       }, onDispatched: input.onTaskDispatched,
     };
     if (input.execution.session.lifecyclePolicy.mode === "warm" && (!credentials.accountIdentity || description.warmTaskVersion !== 1)) throw new Error("Plugin warm task capability or account unavailable");

server/src/services/native-runtime/plugin-runner-probe.test.ts

@@ -229,7 +229,7 @@ describe("plugin Runner setup", () => {
         await mkdir(resources.provider_state!, { recursive: true });
         await writeFile(join(resources.provider_state!, "AGENTS.md"), "Installed persistent instructions");
         const retained = vi.fn(async () => {});
-        const owner = createPluginRunnerTaskOwner({ task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", onSessionRetained: retained, pollMs: 1 });
+        const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", onSessionRetained: retained, pollMs: 1 });
         const first = createRunnerdCodexTransport({ ...owner.driver, stateDirectory, runnerArtifact: artifact,
           runtimeContext: context("first"), prpIdentity: identity, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } });
         let thread: Record<string, unknown>;
@@ -245,7 +245,7 @@ describe("plugin Runner setup", () => {
         for (const run of ["same", "next"]) {
           if (run !== "same") await writeFile(join(resources.provider_state!, "AGENTS.md"), `Installed persistent instructions ${run}`);
           const priorRun = "run-first";
-          const nextOwner = createPluginRunnerTaskOwner({ task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", resumeFrom: "task", pollMs: 1,
+          const nextOwner = createPluginRunnerTaskOwner({ taskId: "task", task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", resumeFrom: "task", pollMs: 1,
             priorIdentity: { ...identity, runId: priorRun, turnId: identity.turnId, itemId: identity.itemId } });
           const second = createRunnerdCodexTransport({ ...nextOwner.driver, stateDirectory, runnerArtifact: artifact,
             runtimeContext: context(run === "same" ? "first" : run),

server/src/services/native-runtime/plugin-runner-probe.ts

@@ -43,7 +43,7 @@ export async function probePluginRunnerEnvironment(context: AdapterEnvironmentTe
       identity: managed?.identity, accountIdentity: managed?.accountIdentity,
       home: managed?.home, environment,
     });
-    const owner = createPluginRunnerTaskOwner({ ...credentials, projectAuthCheckVersion: description.projectAuthCheckVersion, task, projectIds: [],
+    const owner = createPluginRunnerTaskOwner({ ...credentials, taskId: description.taskId, projectAuthCheckVersion: description.projectAuthCheckVersion, task, projectIds: [],
       harness: profile.provider === "acpx" ? profile.acpxAgent! : profile.provider,
       credentialIdentity: managed?.identity });
     probeStarted = true;

server/src/services/native-runtime/plugin-runner-task.test.ts

@@ -54,12 +54,17 @@ function provider() {
 }
 
 describe("plugin native Runner ownership", () => {
+  it("requires an explicit physical task binding", () => {
+    const { task } = provider();
+    expect(() => createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" } as never)).toThrow("physical task unavailable");
+  });
+
   it.each(["finish", "rebind"])("does not send startup Stop after the context %s", mode => {
     const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000);
     const next = new NativeExecutionContext(context.execution, 1_000);
     const remove = vi.spyOn(context.signal, "removeEventListener");
     const { task } = provider();
-    const input = { task, projectIds: [], harness: "codex", executionContext: context, adoptIdentity: requirements.identity };
+    const input = { taskId: "task", task, projectIds: [], harness: "codex", executionContext: context, adoptIdentity: requirements.identity };
     const owner = createPluginRunnerTaskOwner(input);
     try {
       if (mode === "finish") context.finish();
@@ -74,7 +79,7 @@ describe("plugin native Runner ownership", () => {
     let allowed = true;
     task.mockImplementation(async op => op.kind === "check_project_auth"
       ? { ...await perform(op) as object, allowed } : perform(op));
-    const input = { task, projectIds: ["project-retained"], harness: "codex", projectAuthCheckVersion: 1 as const,
+    const input = { taskId: "task", task, projectIds: ["project-retained"], harness: "codex", projectAuthCheckVersion: 1 as const,
       adoptIdentity: requirements.identity };
     const owner = createPluginRunnerTaskOwner(input);
     await owner.driver.authorizeRunnerAdoption!("retained-configuration", requirements.identity);
@@ -90,7 +95,7 @@ describe("plugin native Runner ownership", () => {
     let finish!: () => void;
     const gate = new Promise<void>(resolve => { finish = resolve; });
     task.mockImplementation(async op => { if (mode === "late-release" && op.kind === "check_project_auth") await gate; return perform(op); });
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: ["project-current"], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: ["project-current"], harness: "codex",
       projectAuthCheckVersion: mode === "unsupported" ? undefined : 1, adoptIdentity: requirements.identity });
     const registration = await owner.registration({} as never, "run");
     if (mode === "released") await registration.release("detach");
@@ -107,7 +112,7 @@ describe("plugin native Runner ownership", () => {
     const { task } = provider();
     const close = vi.fn(async () => {});
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close });
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex",
       adoptIdentity: requirements.identity });
     const registration = await owner.registration({} as never, "run");
     registration.activate(); await registration.ready();
@@ -131,7 +136,7 @@ describe("plugin native Runner ownership", () => {
       return result;
     });
     ingress.connect.mockClear();
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex",
       adoptIdentity: requirements.identity });
     const registration = await owner.registration({} as never, "run", abort.signal);
     registration.activate();
@@ -144,7 +149,7 @@ describe("plugin native Runner ownership", () => {
     const { task } = provider();
     const retained = vi.fn(async () => {});
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} });
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex",
       adoptIdentity: requirements.identity, onSessionRetained: retained });
     const registration = await owner.registration({} as never, "run");
     registration.activate(); await registration.ready();
@@ -155,7 +160,7 @@ describe("plugin native Runner ownership", () => {
   it.each(["settled", "unsettled"] as const)("proves retirement independently of %s checkpoint reuse", async settlement => {
     const { task } = provider();
     const retained = vi.fn(async () => {});
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex",
       adoptIdentity: requirements.identity, onSessionRetained: retained });
     const registration = await owner.registration({} as never, "run");
     expect(owner.isRetired()).toBe(false);
@@ -176,7 +181,7 @@ describe("plugin native Runner ownership", () => {
       if (operation.kind === "credentials") return { kind: "credentials", taskId: "task", credential: null };
       return perform(operation);
     });
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", accountIdentity: "account",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", accountIdentity: "account",
       adoptIdentity: requirements.identity, onCredentials: async () => {}, onSessionRetained: async () => {} });
     const registration = await owner.registration({} as never, "run");
     await registration.release();
@@ -187,7 +192,7 @@ describe("plugin native Runner ownership", () => {
 
   it("submits only a prepared identity and configuration ID", async () => {
     const { task } = provider();
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 });
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 });
     expect(() => owner.launcher(launchSpec())).toThrow("requires preparation");
     await prepare(owner);
     expect(task.mock.calls[0][0]).toEqual({ kind: "prepare", requirements, projectIds: [] });
@@ -206,7 +211,7 @@ describe("plugin native Runner ownership", () => {
     const perform = task.getMockImplementation()!;
     task.mockImplementation(async op => op.kind === "prepare"
       ? { ...await perform(op) as object, projectAuthCheckAtSubmit: true } : perform(op));
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1,
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1,
       task, projectIds: [], harness: "codex", pollMs: 1 });
     await prepare(owner);
     expect(task.mock.calls.map(([op]) => op.kind)).toEqual(["prepare"]);
@@ -216,7 +221,7 @@ describe("plugin native Runner ownership", () => {
     registration.activate(); await registration.ready();
     await registration.release(); await registration.checkpoint("unsettled"); await handle.completion;
     expect(task.mock.calls.filter(([op]) => op.kind === "submit")).toHaveLength(1);
-    const unsupported = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" });
+    const unsupported = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex" });
     await expect(prepare(unsupported)).rejects.toThrow("validation unavailable");
   });
 
@@ -225,7 +230,7 @@ describe("plugin native Runner ownership", () => {
     const gate = new Promise<void>(resolve => { finish = resolve; });
     const { task } = provider(); const perform = task.getMockImplementation()!;
     task.mockImplementation(async op => { if (op.kind === "prepare") await gate; return perform(op); });
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex" });
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex" });
     const prepared = prepare(owner);
     const registration = await owner.registration({} as never, "run");
     await registration.release(); finish();
@@ -238,11 +243,11 @@ describe("plugin native Runner ownership", () => {
     let release!: () => void;
     const gate = new Promise<void>(resolve => { release = resolve; });
     task.mockImplementation(async operation => { if (operation.kind === "prepare") await gate; return perform(operation); });
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, executionContext: context });
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, executionContext: context });
     const preparing = prepare(owner);
-    await vi.waitFor(() => expect(task).toHaveBeenCalledWith(expect.objectContaining({ kind: "prepare" })));
+    await vi.waitFor(() => expect(task).toHaveBeenCalledWith(expect.objectContaining({ kind: "prepare" }), { expectedTaskId: "task" }));
     context.abort(new Error("operator Stop"));
-    expect(task).toHaveBeenCalledWith({ kind: "stop" });
+    expect(task).toHaveBeenCalledWith({ kind: "stop" }, { expectedTaskId: "task" });
     release();
     await expect(preparing).rejects.toThrow("preparation cancelled");
     expect(task.mock.calls.some(([operation]) => ["check_project_auth", "connection", "submit"].includes(operation.kind))).toBe(false);
@@ -254,7 +259,7 @@ describe("plugin native Runner ownership", () => {
       ? { kind: "project_auth", taskId: "task", allowed: mode !== "denied",
           configurationId: mode === "wrong-configuration" ? "other" : op.configurationId,
           provider: mode === "wrong-provider" ? "anthropic" : op.provider } : perform(op));
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex",
       projectAuthCheckVersion: mode === "unsupported" ? undefined : 1,
       credentialIdentity: "revision", accountIdentity: "account", credential: { type: "codex-auth-json", authJson: '{"OPENAI_API_KEY":"never-transfer"}' } });
     await expect(prepare(owner)).rejects.toThrow(mode === "unsupported" ? "validation unavailable" : "conflicts");
@@ -264,7 +269,7 @@ describe("plugin native Runner ownership", () => {
   });
   it("rejects an installed account when the plugin lacks project validation", async () => {
     const { task } = provider();
-    const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" });
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex" });
     await expect(prepare(owner)).rejects.toThrow("validation unavailable");
     expect(task.mock.calls.some(([op]) => op.kind === "submit")).toBe(false);
   });
@@ -282,7 +287,7 @@ describe("plugin native Runner ownership", () => {
       return perform(operation);
     });
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} });
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", credential,
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", credential,
       credentialIdentity: "revision", accountIdentity: "stable-account", pollMs: 1,
       onCredentials: async value => { expect(value).toEqual(credential); copied = true; } });
     await prepare(owner);
@@ -297,7 +302,7 @@ describe("plugin native Runner ownership", () => {
     const perform = task.getMockImplementation()!;
     task.mockImplementation(async op => op.kind === "credentials" ? { kind: "credentials", taskId: "task", credential: { type: "codex-auth-json", authJson: "{}" } } : perform(op));
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} });
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", accountIdentity: "account", pollMs: 1,
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", accountIdentity: "account", pollMs: 1,
       onCredentials: async () => { throw new Error("credential handoff failed"); } });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run");
@@ -324,7 +329,7 @@ describe("plugin native Runner ownership", () => {
     });
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} });
     const checkpoint = vi.fn(async () => {});
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: checkpoint, pollMs: 1,
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: checkpoint, pollMs: 1,
       });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run");
@@ -349,7 +354,7 @@ describe("plugin native Runner ownership", () => {
     let fail!: (error: Error) => void;
     const failure = new Promise<never>((_resolve, reject) => { fail = reject; });
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), failure, close: async () => {} });
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",  pollMs: 1 });
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",  pollMs: 1 });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run");
     const observed = registration.failure.catch(error => error);
@@ -370,7 +375,7 @@ describe("plugin native Runner ownership", () => {
       return perform(op);
     });
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} });
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
        pollMs: 1 });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run");
@@ -388,7 +393,7 @@ describe("plugin native Runner ownership", () => {
   it("retains task grants when retention recording fails after route release", async () => {
     const { task } = provider();
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} });
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: async () => { throw new Error("retention write failed"); }, pollMs: 1 });
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: async () => { throw new Error("retention write failed"); }, pollMs: 1 });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run");
     const process = owner.launcher(launchSpec());
@@ -403,7 +408,7 @@ describe("plugin native Runner ownership", () => {
     const { task } = provider();
     const close = vi.fn(async () => {});
     ingress.connect.mockReturnValue({ ready: Promise.resolve(), close });
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
        pollMs: 1 });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run");
@@ -417,7 +422,7 @@ describe("plugin native Runner ownership", () => {
     let finish!: () => void;
     const prepared = new Promise<void>(resolve => { finish = resolve; });
     const { task } = provider();
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
       onDispatched: () => prepared, pollMs: 1 });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run");
@@ -433,7 +438,7 @@ describe("plugin native Runner ownership", () => {
     const { task } = provider();
     ingress.connect.mockClear();
     const controller = new AbortController(); controller.abort();
-    const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex",
        pollMs: 1 });
     await prepare(owner);
     const registration = await owner.registration({} as never, "run", controller.signal);
@@ -456,7 +461,7 @@ describe("plugin task workspace file transfer", () => {
     let file={kind:"file",taskId:"task",base64:bytes.toString("base64"),byteSize:bytes.length,sha256:createHash("sha256").update(bytes).digest("hex")};
     let afterRead=async()=>{};
     task.mockImplementation(async op=>{if(op.kind==="read_file"){await afterRead();return file;}return perform(op);});
-    const owner=createPluginRunnerTaskOwner({projectAuthCheckVersion:1,task,projectIds:[],harness:"codex",pollMs:1});
+    const owner=createPluginRunnerTaskOwner({ taskId: "task",projectAuthCheckVersion:1,task,projectIds:[],harness:"codex",pollMs:1});
     await expect(owner.readWorkspaceFile("proof.txt",1024)).rejects.toThrow("unavailable");
     await prepare(owner);
     ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}});
@@ -480,7 +485,7 @@ it.each(["codex", "claude"])("reuses a warm %s process across run authorities wi
   const {task} = provider();
   ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}});
   const retained = vi.fn(async()=>{});
-  const input = {task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const,
+  const input = {taskId:"task",task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const,
     accountIdentity:"account",credentialIdentity:"revision",pollMs:1,onSessionRetained:retained, githubBroker:{url:"https://core.example.test",token:"first-run-capability"}};
   const owner = createPluginRunnerTaskOwner(input);
   await prepare(owner, launchRequirements(harness));
@@ -512,7 +517,7 @@ it.each(["codex", "claude"])("reuses a warm %s process across run authorities wi
 it.each(["codex", "claude"])("adopts a verified retained %s task without submitting a replacement", async harness => {
   const {task}=provider();
   ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}});
-  const owner=createPluginRunnerTaskOwner({task,projectIds:[],harness,projectAuthCheckVersion:1,
+  const owner=createPluginRunnerTaskOwner({ taskId: "task",task,projectIds:[],harness,projectAuthCheckVersion:1,
     warmTaskVersion:1,accountIdentity:"account",adoptIdentity:requirements.identity,pollMs:1});
   const recovered = {configurationId:"saved",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}};
   await expect(owner.driver.prepareLaunch!(launchRequirements(harness),recovered)).resolves.toEqual(recovered);
@@ -531,7 +536,7 @@ it.each(["codex", "claude"])("denies retained %s process adoption when current p
   task.mockImplementation(async op=>op.kind === "check_project_auth"
     ? {kind:"project_auth",taskId:"task",configurationId:op.configurationId,provider:op.provider,allowed:false} as never
     : {kind:"accepted",taskId:"task"} as never);
-  const owner=createPluginRunnerTaskOwner({task,projectIds:["project"],harness,projectAuthCheckVersion:1,
+  const owner=createPluginRunnerTaskOwner({ taskId: "task",task,projectIds:["project"],harness,projectAuthCheckVersion:1,
     warmTaskVersion:1,accountIdentity:"account",adoptIdentity:requirements.identity});
   await expect(owner.driver.prepareLaunch!(launchRequirements(harness),{configurationId:"saved",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}}))
     .rejects.toThrow("project authentication conflicts");
@@ -540,12 +545,13 @@ it.each(["codex", "claude"])("denies retained %s process adoption when current p
 
 it.each(["codex", "claude"])("refuses warm %s account, harness, project, or credential generation changes", async harness=>{
   const {task}=provider();
-  const input={task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const,
+  const input={taskId:"task",task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const,
     accountIdentity:"account",credentialIdentity:"revision",pollMs:1};
   const owner=createPluginRunnerTaskOwner(input);
   await prepare(owner, launchRequirements(harness));
   const spec=launchSpec();spec.nativeRunner!.lifecyclePolicy={mode:"warm",idleTimeoutMs:300000};
   const process=owner.launcher(spec);
+  expect(()=>owner.rebind({...input,taskId:"replacement"})).toThrow("cannot be rebound");
   expect(()=>owner.rebind({...input,accountIdentity:"other"})).toThrow("cannot be rebound");
   expect(()=>owner.rebind({...input,credentialIdentity:"rotated"})).toThrow("cannot be rebound");
   expect(()=>owner.rebind({...input,harness:harness === "codex" ? "claude" : "codex"})).toThrow("cannot be rebound");
@@ -560,7 +566,7 @@ it.each(["codex", "claude"])("refuses warm %s account, harness, project, or cred
   task.mockImplementation(async operation => operation.kind === "describe"
     ? {kind:"description",taskId:"task",runner:{version:"fixture",digest:`sha256:${"a".repeat(64)}`}}
     : implementation(operation));
-  const owner = createPluginRunnerTaskOwner({task,projectIds:[],harness:"codex",
+  const owner = createPluginRunnerTaskOwner({ taskId: "task",task,projectIds:[],harness:"codex",
     githubBroker:{url:"https://core.example.test",token:"runtime-capability"}});
   await expect(prepare(owner)).rejects.toThrow("credential broker v1");
   expect(task.mock.calls.map(([op])=>op.kind)).toEqual(["describe"]);

server/src/services/native-runtime/plugin-runner-task.ts

@@ -29,7 +29,8 @@ export async function verifyPluginRunnerTaskReattachment(
  */
 export function createPluginRunnerTaskOwner(input: {
   executionContext?: NativeExecutionContext;
-  task: (operation: unknown) => Promise<unknown>;
+  taskId: string;
+  task: (operation: unknown, options?: { expectedTaskId?: string }) => Promise<unknown>;
   projectIds: readonly string[];
   harness: string;
   projectAuthCheckVersion?: 1;
@@ -46,6 +47,8 @@ export function createPluginRunnerTaskOwner(input: {
   githubBroker?: { url: string; token: string };
   adoptIdentity?: DurableRecoveryIdentity;
 }) {
+  const taskId = input.taskId;
+  if (!taskId) throw new Error("Plugin Runner physical task unavailable");
   const pollMs = input.pollMs ?? 500;
   let registrationGeneration = 0;
   let activeOutbound: ReturnType<typeof connectRunnerPrpIngress> | null = null;
@@ -72,7 +75,7 @@ export function createPluginRunnerTaskOwner(input: {
     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
-      ? measureRunPhase(phase, () => input.task(operation)) : input.task(operation)));
+      ? 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");
     }
@@ -353,7 +356,7 @@ export function createPluginRunnerTaskOwner(input: {
     // stop request. Completion also proves credential/writeback obligations ran.
     isRetired: () => stoppedState?.executionStopped === true && completed && !detached,
     rebind(next: typeof input) {
-      if (!launched || stopRequested || stoppedState || detached || next.harness !== input.harness ||
+      if (!launched || stopRequested || stoppedState || detached || next.taskId !== taskId || next.harness !== input.harness ||
           input.accountIdentity !== next.accountIdentity || input.credentialIdentity !== next.credentialIdentity ||
           JSON.stringify(input.projectIds) !== JSON.stringify(next.projectIds)) throw new Error("Plugin warm task cannot be rebound");
       input = next;

server/src/services/plugin-task-credential-writeback.ts

@@ -24,7 +24,7 @@ const handoffSchema = z.object({
   state: z.enum(["pending", "complete"]),
 }).strict();
 const key = "runnerTaskCredentialHandoff";
-type Scope = { companyId: string; leaseId: string };
+type Scope = { companyId: string; leaseId: string; expectedTaskId?: string };
 
 /** Save authority, never credential bytes or a controller-local temporary path. */
 export async function registerPluginTaskCredentialHandoff(
@@ -126,13 +126,15 @@ export async function settlePluginTaskCredentialHandoff(
       metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ [key]: { ...handoff, state: "complete" } })}::jsonb`,
     }).where(and(
       eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId),
+      scope.expectedTaskId === undefined ? undefined : eq(environmentLeases.providerLeaseId, scope.expectedTaskId),
       sql`${environmentLeases.metadata}->${key} = ${JSON.stringify(handoff)}::jsonb`,
     )).returning({ id: environmentLeases.id });
     if (completed.length !== 1) {
-      const [current] = await db.select({ metadata: environmentLeases.metadata }).from(environmentLeases).where(and(
+      const [current] = await db.select({ metadata: environmentLeases.metadata, taskId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and(
         eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId),
       ));
-      if (!isDeepStrictEqual(handoffSchema.safeParse(current?.metadata?.[key]).data, { ...handoff, state: "complete" })) {
+      if (scope.expectedTaskId !== undefined && current?.taskId !== scope.expectedTaskId ||
+          !isDeepStrictEqual(handoffSchema.safeParse(current?.metadata?.[key]).data, { ...handoff, state: "complete" })) {
         throw new Error("Plugin task credential handoff changed");
       }
     }