Admit exact stopped Runner sources under the current run lease (#19134)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native execution keeps provider state and a durable controller
journal.
> - A completed process can stop before the next run attaches.
> - The current run must own recovery of that exact source.
> - An old controller must lose permission to change the consumed source
journal.
> - This pull request adds explicit source admission and protected task
dispatch.
> - The benefit is a narrow recovery boundary that preserves the
original successful result.

## Linked Issues or Issue Description

Native prerequisites #19132 and #19153 are merged into master. This PR
targets master and updates the TypeScript Runner driver contract
alongside Core and SDK admission. This change adds the SDK and Core
admission boundary. Executor selection and fresh source preparation are
a separate change. Related #17916 supplies the workspace export holds
preserved here.

## What Changed

- Plugins advertise `codexSourceRecoveryVersion: 1` for Codex App Server
and `acpxSourceRecoveryVersion: 1` for ACPX. Hosts can qualify either,
both, or neither. Core selects the matching field, because a
stopped-source check for one provider cannot establish the other
provider’s recovery guarantees. ACPX orchestration is tracked separately
in #19196. Use the existing exact source identity and `resumeFrom` under
the authenticated current run. No public recovery mode is added. Require
the claimed source, current account and prepared launch at protected
dispatch.
- Keep the first source binding pinned on the consumer lease, including
after uncertain dispatch. A newly acquired coordinator can refresh only
the same source.
- Claim one stopped successful source under the current run's acquired
native lease. Store the token-free consumer reference in existing lease
metadata. Preserve the source status and result.
- Check the exact six source identity fields, encrypted journal header,
company, task, agent, environment, driver, workspace, account selection,
and current coordinator generation. Keep workspace export and legacy
settlement holds.
- Serialize claims and checkpoint writers in the same coordinator lock
order. Reject stale source writes after consumption. Check the current
consumer before recovered source writes.
- Check source admission before and after protected worker dispatch.
Keep the request and lease owned by the current run while the native
identity remains the source.
- Allow the same owner to prepare B after a fresh source submission,
using the exact consumed-source proof instead of a generic warm-reuse
flag. Recheck that proof after dispatch.
- Preserve ordinary cached adoption and repeated preparation after
launch. The current authenticated owner can differ from the adopted PRP
identity without requesting a new source launch. Keep ordinary suspended
recovery and live task attachment checks. Extract the shared source
resolver to avoid a module cycle, and make ordinary selection honor the
same durable source consumer.

## Verification

- Current head `cac9cf188a` includes merged #19300 and the symmetric
capabilities introduced in `b7e25a64b4`. The task-owner and warm-rebind
fixtures now identify their physical task. All 29 SDK schema cases, 95
Core ownership and source-admission cases, 25 native Codex and ACPX
recovery cases, and the binary metadata check passed on this head. The
SDK and Runner builds and server typecheck passed. Both native providers
reject the other provider’s capability. GitHub CI is green and [standard
Greptile scored this commit
5/5](https://github.com/paperclipai/paperclip/pull/19134#issuecomment-6645574114)
with no open threads.

- Previous head `bda49271a4ae362a15baf3fe6d6d297e25fd2000` is directly
based on master after #19132 and #19153 landed. [GitHub workflow
42072192023](https://github.com/paperclipai/paperclip/actions/runs/42072192023)
passed on its first attempt. All check runs completed successfully or as
intentional workflow skips; Snyk passed. Standard Greptile scored this
exact head 5/5. That review had no unresolved findings.
- Runner and SDK TypeScript builds and server typecheck passed. The
focused SDK, task-owner, task-runtime, PostgreSQL source and authority
suites passed all 161 cases. The production transport-factory getter
regression and 14 native-lane selection checks also passed.
- The Linux native lane executed the new ordinary warm A→B process case.
It kept one Runner/provider process and completed two turns on the same
thread. The eager-getter mutation fails with
`native_runner_source_attachment_pending`; preserving the live getter
passes. The production factory test changes the current owner after
transport creation and checks that same options object.
- PostgreSQL cases cover both concurrent source-writer/claim orders,
exact source submission under the current caller, source-switch denial
after uncertain dispatch, same-source coordinator refresh, stale
same-boot generation, corrupted journal headers, competing consumers,
account changes and export holds.
- The protected dispatch sequence submits A, prepares B and retries B
preparation. It rejects undispatched sources, changed accounts and
foreign target bindings. Source status and accepted result remain
unchanged.
- Ordinary adopted A preparation under current B, repeated ordinary
preparation after launch and ordinary suspended continuation retain
their existing path. Absent or false claims preserve ordinary identity
denial; malformed claims fail closed.
- Earlier development attempts exposed ordinary preparation regressions
and the live-getter omission; the final regressions cover those fixes.
The first embedded PostgreSQL attempt needed missing library aliases
repaired in ignored dependencies. Those attempts are not passing
evidence.
- Historical, superseded CI heads had unchanged UI fake-timer and
Playwright health-response-lifetime failures. Their isolated cases and
scoped retries passed. The response-lifetime cause remains unproven. No
UI/test timeout was changed. The final rebased-head workflow above
passed without a retry.
- Whitespace and added-line secret/PII scans passed. Full workspace
checks ran on GitHub. No live vendor probe or deployment was performed.

## Risks

- Any driver that used the earlier Codex TypeScript option must use
`codexSourceRecoveryVersion`. The new plugin description rejects the
generic spelling. The Launcher’s existing Codex-specific HTTP header can
map to this property when the plugin adopts the new SDK. The change
requires no journal migration or ROM change.

- This is an inactive admission foundation. Executor orchestration is
separate in #19157. Native restore integrity and provider
stopped-source/account validation must be present before executor
activation. The native prerequisite uses ordinary recovery followed by
attachment. The composed executor proofs are in #19157.
- The source claim is durable. A later controller must recover the same
consumer or keep the hold. It cannot select another source after an
ambiguous launch.
- Admission adds bounded database checks around protected startup calls.
Busy issue locks fail closed; other lock waits have a one-second bound.
- No migration, snapshot format, proactive receipt, recurring worker, or
deployment is included. Full workspace checks and platform coverage run
in CI. These tests do not prove vendor history durability or a hosted
rollout.

## Model Used

OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning, through Codex with
repository tools and code execution. The runtime does not expose a
context-window size.

## 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 on the current head
- [x] Standard Greptile is 5/5 with no unresolved findings on the
current head
- [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

@@ -84,6 +84,39 @@ authority as submission.
 Core requires preparation version 1 for native plugin execution. It validates
 and saves the result before it submits its configuration ID.
 
+### Stopped source reconciliation
+
+A plugin’s `describe` response advertises `codexSourceRecoveryVersion: 1` for
+Codex App Server and `acpxSourceRecoveryVersion: 1` for ACPX. It can advertise
+either, both, or neither. Each field means the host can verify and restore a
+stopped source for that provider without starting another turn. Core requires
+the field for the selected provider and verifies the run’s eligibility.
+
+The authenticated request belongs to
+the current run while `prepare` and `submit` carry the predecessor Runner identity.
+Core derives this relationship from its saved source-consumer claim; a mismatched
+run ID alone never grants recovery. Both operations must name the same `resumeFrom`, exact source Runner identity, and current
+`accountIdentity`. The request and lease still belong to the current run.
+Preparation must save this source binding with its new configuration. Submission
+must verify the same binding before it starts the source process.
+
+Core requires the source run to have a committed successful result, an encrypted
+PRP journal, confirmed physical termination, and no other source consumer. Core
+claims the source under the current native controller lease. This does not change
+the source run's status or result. A stale source controller cannot write its
+journal after this claim. Workspace export and settlement holds still apply.
+After source submission, the same current owner can prepare its attachment with
+the exact source as `priorIdentity`. This does not grant ordinary warm reuse.
+
+The provider must verify the original account, workspace, runtime, and source
+ownership before launch. The provider must reject saved active goals, pending tools, ambiguous starts, and
+uncertain work before launch. Runner verifies that a saved idle provider remains
+idle when restored.
+Until the existing authenticated `run.attach` operation changes authority, the
+Core retains semantic inputs and sends no fresh predecessor business commands
+while the restored source reconciles its journal and known cleanup. Ordinary suspended recovery and live attachment keep their existing
+checks. Providers that omit the capability do not support predecessor recovery.
+
 ## Launch authentication validation
 
 Declare `projectAuthCheckVersion: 1` in `describe` for each native task launch.

packages/paperclip-runner/runner/DURABLE_TRANSPORT.md

@@ -76,7 +76,7 @@ the conversation. Core uses ordinary recovery for A, then attaches B with
 authenticated `run.attach`. If the original process is alive, its owner can
 verify and adopt it instead.
 
-**Is A idle?** Before launch, the host driver (`sourceRecoveryVersion: 1`)
+**Is A idle?** Before launch, the host driver (`codexSourceRecoveryVersion: 1`)
 checks that A’s process stopped and its saved session was idle. After launch,
 the Runner (`recoveryIntegrityVersion: 1`) verifies that Codex restored an idle
 session when no command effect is uncertain. It reads the goal, then the

packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs

@@ -675,6 +675,7 @@ mod tests {
         let metadata = build_metadata();
         assert_eq!(metadata["recoveryIntegrityVersion"], 1);
         assert!(metadata.get("sourceRecoveryVersion").is_none());
+        assert!(metadata.get("codexSourceRecoveryVersion").is_none());
         assert_eq!(metadata["acpxRecoveryIntegrityVersion"], 1);
         assert!(metadata.get("acpxSourceRecoveryVersion").is_none());
         assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA);

packages/paperclip-runner/src/control-plane/runner-execution-driver.ts

@@ -9,9 +9,9 @@ import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js";
  * specifications stay private to the driver; plugins receive versioned requirements.
  */
 export interface RunnerExecutionDriver {
-  /** Attests stopped ownership and dormant prelaunch admission for the selected provider. */
-  sourceRecoveryVersion?: 1;
-  /** Same trusted prelaunch attestation for a qualified ACPX provider. */
+  /** The driver verifies a stopped, idle Codex App Server source before launch. */
+  codexSourceRecoveryVersion?: 1;
+  /** The driver verifies a stopped, idle source for the selected ACPX provider before launch. */
   acpxSourceRecoveryVersion?: 1;
   /** Current authenticated lease/checkpoint-writer owner; may change during warm reuse. */
   authorityRunId?: string;

packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts

@@ -56,7 +56,7 @@ async function fixture(agent: "codex" | "claude" = "codex", nativeSource = false
   let recovering = false;
   const options: CapabilityRunnerdCodexTransportOptions = {
     provider: "acpx", acpxAgent: agent, runnerBinary: binary, stateDirectory: root, prpIdentity: identity,
-    runnerArtifact: readRunnerdArtifactBinding(binary), sourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1,
+    runnerArtifact: readRunnerdArtifactBinding(binary), codexSourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1,
     lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, runnerProcessReconnectGraceMs: null, runnerReconnectGraceMs: 3_000,
     controlPlaneRegistration: async next => {
       const generation = ++registrationGeneration; authority = next;

packages/paperclip-runner/src/live/runnerd-codex-transport.ts

@@ -5613,7 +5613,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       throw new Error("native_runner_source_reconciliation_requires_exact_authority");
     }
     if (lostPredecessor && (!["codex", "acpx"].includes(this.options.provider ?? "codex") || this.options.acpxCandidateAdmission)) throw new Error("native_runner_recovery_integrity_unsupported");
-    const recoveryCapability = this.options.provider === "acpx" ? this.options.acpxSourceRecoveryVersion : this.options.sourceRecoveryVersion;
+    const recoveryCapability = this.options.provider === "acpx" ? this.options.acpxSourceRecoveryVersion : this.options.codexSourceRecoveryVersion;
     if (lostPredecessor && recoveryCapability !== 1) {
       throw new Error("native_runner_recovery_integrity_unsupported");
     }

packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts

@@ -90,7 +90,7 @@ async function sourceFixture(registration?: CapabilityRunnerdCodexTransportOptio
     await vi.waitFor(() => { expect(absent(evidence.runnerPid!)).toBe(true); expect(absent(evidence.codexPid!)).toBe(true); });
     expect((await read(runnerFile)).lifecycle).toBe("ready");
     const recover = (overrides: Partial<CapabilityRunnerdCodexTransportOptions> = {}) => {
-      const bundle = createCapabilityRunnerdCodexTransport({ ...options, sourceRecoveryVersion: 1, resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return Object.hasOwn(overrides, "authorityRunId") ? overrides.authorityRunId : "run-target"; } });
+      const bundle = createCapabilityRunnerdCodexTransport({ ...options, codexSourceRecoveryVersion: 1, resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return Object.hasOwn(overrides, "authorityRunId") ? overrides.authorityRunId : "run-target"; } });
       bundles.push(bundle); return bundle;
     };
     return { root, options, identity, thread, runnerFile, providerFile, coreFile, calls, complete, settled, recover, dispose, nativeIdentity, execution, nativeSnapshot };
@@ -101,6 +101,8 @@ it.each(["isolated", "preconfigured"] as const)("recovers the real native backen
   const f = await sourceFixture(undefined, true, homeMode);
   try {
     expect(f.nativeSnapshot!.sessionId).toBe(f.thread.id);
+    await expect(f.recover({ codexSourceRecoveryVersion: undefined, acpxSourceRecoveryVersion: 1 })
+      .transport.request("thread/read", {})).rejects.toThrow("recovery_integrity_unsupported");
     let owner = "run-target";
     const source = f.recover({ ...(homeMode === "isolated" ? { environment: { ...process.env, PAPERCLIP_RUNNER_HOME_MODE: "preconfigured" } } : {}), get authorityRunId() { return owner; } });
     await expect(source.transport.request("thread/read", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch");
@@ -198,7 +200,7 @@ it.each([["launch", "default"], ["adopt", "default"], ["adopt-denied", "default"
     expect((await prepare(requirements)).provider).toEqual(provider); prepare.mockClear();
     const binary = fixture.options.runnerBinary;
     let owner: string | undefined = "prepared-target";
-    const sourceOptions: Partial<CapabilityRunnerdCodexTransportOptions> = { sourceRecoveryVersion: 1, get authorityRunId() { return owner; },
+    const sourceOptions: Partial<CapabilityRunnerdCodexTransportOptions> = { codexSourceRecoveryVersion: 1, get authorityRunId() { return owner; },
       recoveryLaunch: { requirements }, prepareLaunch: prepare,
       disableCodexWebSearch: policy === "conflict" ? false : undefined,
       runnerArtifact: { version: "0.3.0", digest: `sha256:${createHash("sha256").update(await readFile(binary)).digest("hex")}` },

packages/plugins/sdk/src/environment-tasks.ts

@@ -137,6 +137,9 @@ export const environmentTaskResultSchema = z.discriminatedUnion("kind", [
     warmTaskVersion: z.literal(1).optional(),
     launchPreparationVersion: z.literal(1).optional(),
     executionContextVersion: z.literal(1).optional(),
+    /** Each version attests stopped-source recovery for its provider independently. */
+    codexSourceRecoveryVersion: z.literal(1).optional(),
+    acpxSourceRecoveryVersion: z.literal(1).optional(),
     freshWorkspaceVersion: z.literal(1).optional(),
     projectAuthCheckVersion: z.literal(1).optional(),
     workspaceFileReadVersion: z.literal(1).optional(),

packages/plugins/sdk/tests/environment-tasks.test.ts

@@ -1,5 +1,5 @@
 import {describe,expect,it} from "vitest";
-import {environmentTaskOperationSchema} from "../src/environment-tasks.js";
+import {environmentTaskOperationSchema,environmentTaskResultSchema} from "../src/environment-tasks.js";
 const submit={kind:"submit",projectIds:[],bootstrapTicket:"synthetic",runner:{harness:"codex",protocolMin:1,protocolMax:3,runnerId:"runner",leaseId:"lease",runId:"run",sessionId:"session",turnId:"turn",itemId:"item"},
  launch:{runnerVersion:"fixture",runnerDigest:`sha256:${"a".repeat(64)}`,maxOutboxBytes:262144,p0ReserveBytes:65536,lifecyclePolicy:{mode:"warm",idleTimeoutMs:300000}}};
 describe("task lifecycle policy",()=>{
@@ -24,6 +24,31 @@ const prepare = {
   },
 };
 
+describe("retained source launch fields", () => {
+  it("uses the existing source identity and resume reference without a recovery mode", () => {
+    for (const operation of [prepare, submit]) {
+      const recovery = { ...operation, accountIdentity: "selected-account", resumeFrom: "source-task" };
+      expect(environmentTaskOperationSchema.parse(recovery)).toEqual(recovery);
+      expect(environmentTaskOperationSchema.safeParse({ ...recovery, recoveryPurpose: "source_reconcile" }).success).toBe(false);
+    }
+  });
+});
+
+describe("provider-specific source recovery capability", () => {
+  const description = { kind: "description", taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
+  it.each(["codexSourceRecoveryVersion", "acpxSourceRecoveryVersion"] as const)("accepts %s independently", key => {
+    const advertised = { ...description, [key]: 1 };
+    expect(environmentTaskResultSchema.parse(advertised)).toEqual(advertised);
+    expect(environmentTaskResultSchema.safeParse({ ...description, [key]: 2 }).success).toBe(false);
+  });
+  it("accepts both or neither and rejects an unqualified recovery flag", () => {
+    const both = { ...description, codexSourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1 };
+    expect(environmentTaskResultSchema.parse(both)).toEqual(both);
+    expect(environmentTaskResultSchema.parse(description)).toEqual(description);
+    expect(environmentTaskResultSchema.safeParse({ ...description, sourceRecoveryVersion: 1 }).success).toBe(false);
+  });
+});
+
 describe("task hosted search restriction", () => {
   it("preserves Core's explicit restriction through plugin launch preparation", () => {
     const operation = { ...prepare, requirements: { ...prepare.requirements,

scripts/__tests__/run-vitest-stable-shard.test.mjs

@@ -323,6 +323,7 @@ const nativeRunnerSuitePath =
 const ingressRetrySuitePath = "server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts";
 const usageDeliverySuitePath = "server/src/services/native-runtime/native-usage-delivery.integration.test.ts";
 const warmCheckpointOwnershipSuitePath = "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts";
+const pluginWarmOwnerSuitePath = "server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts";
 const dotRunnerSuitePath = "server/src/__tests__/dot-runner.test.ts";
 const restartRecoverySuitePath =
   "server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts";
@@ -371,7 +372,8 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes
   assert.ok(!files.includes(restartRecoverySuitePath));
   assert.ok(!files.includes(usageDeliverySuitePath));
   assert.ok(!files.includes(warmCheckpointOwnershipSuitePath));
-  assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
+  assert.ok(!files.includes(pluginWarmOwnerSuitePath));
+  assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
   assert.equal(new Set(files).size, files.length);
   const defaultRun = dryRunJson([], prEnv);
   assert.ok(defaultRun.generalServerSuiteCount === full.generalServerSuiteCount);
@@ -392,6 +394,7 @@ for (const [caller, envOverrides] of [["Release", { GITHUB_WORKFLOW: "Release" }
     assert.ok(files.includes(dotRunnerSuitePath));
     assert.ok(files.includes(restartRecoverySuitePath));
     assert.ok(files.includes(warmCheckpointOwnershipSuitePath));
+    assert.ok(files.includes(pluginWarmOwnerSuitePath));
     assert.deepEqual([...files, chatSuitePath].sort(), full.selectedGeneralServerSuites.sort());
     assert.equal(new Set(files).size, files.length);
   });
@@ -425,6 +428,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => {
     ...restartWaitSuites,
     usageDeliverySuitePath,
     warmCheckpointOwnershipSuitePath,
+    pluginWarmOwnerSuitePath,
   ]);
 });
 

scripts/run-vitest-stable.mjs

@@ -104,6 +104,7 @@ const nativeRunnerSuites = [
   "server/src/services/native-runtime/native-runner-restart-diskless.integration.test.ts",
   "server/src/services/native-runtime/native-usage-delivery.integration.test.ts",
   "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts",
+  "server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts",
 ];
 // In the PR workflow (pr.yml, the caller of pr-trusted.yml — reusable
 // workflows inherit the caller's GITHUB_WORKFLOW), the last Verify Paperclip

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

@@ -1,17 +1,20 @@
+import { resolvePluginTaskResume } from "./plugin-task-resume.js";
+export { resolvePluginTaskResume } from "./plugin-task-resume.js";
 import { attachSandboxCreation, environmentService } from "./environments.js";
 import { createOwnedSandbox, queueSandboxCreationCleanup, recordSandboxCreationResource } from "./sandbox-creation-operations.js";
 import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js";
 import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js";
 import { readWorkspaceProviderBinding } from "@paperclipai/shared";
 import { settlePluginTaskCredentialHandoff } from "./plugin-task-credential-writeback.js";
-import { and, eq, inArray, or, sql } from "drizzle-orm";
+import { and, eq, inArray, sql } from "drizzle-orm";
 import { environmentLeases, executionWorkspaces, environments, heartbeatRuns, issues, projects, type Db } from "@paperclipai/db";
 import { freshWorkspaceSchema, environmentTaskOperationSchema, parseEnvironmentTaskResult,
   type HostToWorkerMethodName, type HostToWorkerMethods, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk";
 import { pluginRegistryService } from "./plugin-registry.js";
-import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt } from "./remote-execution-termination.js";
+import { remoteTerminationReceipt } from "./remote-execution-termination.js";
 import type { PluginWorkerManager } from "./plugin-worker-manager.js";
 import { NativeExecutionContext } from "./native-runtime/native-execution-context.js";
+import { assertPluginTaskSourceAttachment, assertPluginTaskSourceRecovery } from "./native-runtime/plugin-task-source-recovery.js";
 import { pluginAgentRuntimeIdentity } from "./plugin-agent-runtime-state.js";
 
 export async function pluginEnvironmentSupportsTasks(db: Db, workers: PluginWorkerManager, lease: { companyId: string; metadata: Record<string, unknown> | null }) {
@@ -56,6 +59,11 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
   const taskId = row.lease.providerLeaseId;
   if (input.expectedTaskId !== undefined && input.expectedTaskId !== taskId) throw new Error("Plugin Runner physical task changed");
   const metadata = lease.metadata ?? {};
+  // A different launch identity is authorized by the saved consumer claim, never by the mismatch alone.
+  const sourceRecovery = run && ((operation.kind === "prepare" && operation.requirements.identity.runId !== run.id && metadata.runnerTaskSourceRecovery) ||
+      (operation.kind === "submit" && operation.runner.runId !== run.id && metadata.runnerTaskSourceRecovery))
+    ? await assertPluginTaskSourceRecovery(db, { companyId: input.companyId, leaseId: lease.id,
+      context: context instanceof NativeExecutionContext ? context : undefined, operation }) : undefined;
   const probe = metadata.runnerTaskProbe as { runId?: unknown; sessionId?: unknown; expiresAtMs?: unknown } | undefined;
   const activeProbe = lease.heartbeatRunId === null && lease.issueId === null && probe
     && typeof probe.runId === "string" && typeof probe.sessionId === "string"
@@ -94,15 +102,19 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
   if ((operation.kind === "read_file" || operation.kind === "describe" || operation.kind === "prepare" || operation.kind === "submit" || (operation.kind === "connection" && !retainedConnection) || (operation.kind === "read_state" && metadata.runnerTaskExecutionStopped !== true && !retainedCheckpointRead) || operation.kind === "archive_state" || operation.kind === "check_project_auth" || operation.kind === "copilot_metadata") &&
       (metadata.runnerTaskRetiring === true || !environment || (!activeProbe && run?.status !== "running") || lease.status !== "active" || (lease.expiresAt && lease.expiresAt.getTime() <= Date.now()))) throw new Error("Plugin-provided Runner execution lease is not active");
   if (operation.kind === "copilot_metadata" && !activeProbe) throw new Error("Copilot metadata requires a private setup lease");
-  const warmAttachment = !!prior && metadata.runnerTaskReused === true &&
+  const sourceReference = metadata.runnerTaskSourceRecovery as { identity?: { runId?: unknown } } | undefined;
+  const sourceAttachment = operation.kind === "prepare" && prior && sourceReference?.identity?.runId === prior.runId
+    ? await assertPluginTaskSourceAttachment(db, { companyId: input.companyId, leaseId: lease.id,
+      context: context instanceof NativeExecutionContext ? context : undefined, operation }) : undefined;
+  const warmAttachment = !!prior && (metadata.runnerTaskReused === true || !!sourceAttachment) &&
     (matches(prior, savedIdentity) || matches(prior, priorSaved) && operation.kind === "prepare" && matches(operation.requirements.identity, savedIdentity)) && runner?.runnerId === prior.runnerInstanceId &&
     runner?.leaseId === prior.environmentLeaseId && runner?.sessionId === prior.normalizedSessionId;
   if (prior && !warmAttachment) throw new Error("Plugin warm task prior identity mismatch");
-  if (runner && ((!checkpointIdentity || !resumeFrom) && !(ownCheckpoint && operation.kind !== "read_file") && runner.runId !== (activeProbe ? probe.runId : run?.id) || (!resumeFrom && runner.leaseId !== lease.id && !ownCheckpoint && !warmAttachment)
+  if (runner && !sourceRecovery && ((!checkpointIdentity || !resumeFrom) && !(ownCheckpoint && operation.kind !== "read_file") && runner.runId !== (activeProbe ? probe.runId : run?.id) || (!resumeFrom && runner.leaseId !== lease.id && !ownCheckpoint && !warmAttachment)
     || (activeProbe && (runner.sessionId !== probe.sessionId || (launches && operation.projectIds.length > 0))))) {
     throw new Error("Plugin-provided Runner execution identity mismatch");
   }
-  if (runner && resumeFrom) {
+  if (runner && resumeFrom && !sourceRecovery) {
     const inspectCheckpoint = operation.kind === "read_state";
     const sources = await db.select({ id: environmentLeases.id, runId: environmentLeases.heartbeatRunId, metadata: environmentLeases.metadata }).from(environmentLeases).where(and(
       eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.provider, lease.provider!),
@@ -373,6 +385,14 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
     const result = await callWorker("environmentTask", params, timeout);
     if (executionContext) context!.assertOpen();
     const parsed = parseEnvironmentTaskResult(operation, taskId, result);
+    if (sourceRecovery && (operation.kind === "prepare" || operation.kind === "submit")) {
+      await assertPluginTaskSourceRecovery(db, { companyId: input.companyId, leaseId: lease.id,
+        context: context instanceof NativeExecutionContext ? context : undefined, operation });
+    }
+    if (sourceAttachment && operation.kind === "prepare") {
+      await assertPluginTaskSourceAttachment(db, { companyId: input.companyId, leaseId: lease.id,
+        context: context instanceof NativeExecutionContext ? context : undefined, operation });
+    }
     if (operation.kind === "read_file") {
       const [current] = await db.select({ lease: environmentLeases, run: heartbeatRuns }).from(environmentLeases)
         .leftJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), eq(heartbeatRuns.companyId, environmentLeases.companyId)))
@@ -470,48 +490,3 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor
     throw new Error("Plugin Runner termination is unconfirmed; the saved lease requires cleanup");
   }
 }
-
-/** A saved provider task can resume only within the same authorized agent workspace. */
-export async function resolvePluginTaskResume(db: Db, input: {
-  companyId: string; leaseId: string;
-  priorIdentity: { runnerInstanceId: string; environmentLeaseId: string; runId: string; normalizedSessionId: string };
-  /** Read-only discovery; launching still requires an attested retained checkpoint. */
-  inspectCheckpoint?: boolean;
-}): Promise<string> {
-  const rows = await db.select({ lease: environmentLeases, run: heartbeatRuns }).from(environmentLeases)
-    .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), eq(heartbeatRuns.companyId, input.companyId)))
-    .where(and(eq(environmentLeases.companyId, input.companyId),
-      or(eq(environmentLeases.id, input.leaseId), eq(heartbeatRuns.id, input.priorIdentity.runId))));
-  const next = rows.find(row => row.lease.id === input.leaseId);
-  // A retained PRP lease is logical; later provider tasks have different physical lease IDs.
-  const candidates = rows.filter(row => {
-    const identity = row.lease.metadata?.runnerTaskIdentity as Record<string, unknown> | undefined;
-    return row.lease.id !== input.leaseId && row.run.id === input.priorIdentity.runId &&
-      (input.inspectCheckpoint || row.lease.metadata?.runnerTaskSessionRetained === true) &&
-      identity?.runnerId === input.priorIdentity.runnerInstanceId && identity.leaseId === input.priorIdentity.environmentLeaseId &&
-      identity.sessionId === input.priorIdentity.normalizedSessionId && identity.runId === input.priorIdentity.runId &&
-      !rows.some(successor => successor.lease.id !== input.leaseId &&
-        successor.lease.metadata?.runnerTaskResumeFrom === row.lease.providerLeaseId);
-  });
-  const old = candidates.length === 1 ? candidates[0] : undefined;
-  const identity = old?.lease.metadata?.runnerTaskIdentity as Record<string, unknown> | undefined;
-  if (!old || !next || !old.lease.providerLeaseId || !identity ||
-      identity.runnerId !== input.priorIdentity.runnerInstanceId || identity.leaseId !== input.priorIdentity.environmentLeaseId ||
-      identity.sessionId !== input.priorIdentity.normalizedSessionId || identity.runId !== input.priorIdentity.runId ||
-      old.lease.metadata?.driver !== "plugin" || next.lease.metadata?.driver !== "plugin" ||
-      !next.lease.providerLeaseId || next.lease.providerLeaseId === old.lease.providerLeaseId ||
-      (old.run.id !== next.run.id && !["succeeded", "interrupted", "failed", "cancelled", "timed_out"].includes(old.run.status)) ||
-      old.run.agentId !== next.run.agentId || old.lease.issueId !== next.lease.issueId ||
-      old.lease.executionWorkspaceId !== next.lease.executionWorkspaceId ||
-      !old.lease.environmentId || old.lease.environmentId !== next.lease.environmentId ||
-      !next.lease.provider?.startsWith("plugin:") || old.lease.provider !== next.lease.provider ||
-      old.lease.metadata?.pluginId !== next.lease.metadata?.pluginId || old.lease.metadata?.driverKey !== next.lease.metadata?.driverKey ||
-      next.lease.status !== "active" || next.lease.releasedAt || next.run.status !== "running" ||
-      (next.lease.expiresAt && next.lease.expiresAt.getTime() <= Date.now()) ||
-      !(old.run.id === next.run.id
-        ? rows.filter(row => row.lease.id !== next.lease.id).every(row => hasRemoteTerminationReceipt(row.lease))
-        : await remoteExecutionHasStopped(db, input.companyId, old.run.id))) {
-    throw new Error("Plugin Runner retained session is not authorized");
-  }
-  return old.lease.providerLeaseId;
-}

server/src/services/native-runtime/native-execution-context.ts

@@ -15,6 +15,7 @@ export class NativeExecutionContext {
   #transport?: Promise<PluginEnvironmentTaskExecutionContext["transport"]>;
   #target?: string;
   #finished = false;
+  #nativeLease?: { bootId: string; leaseOwner: string; generation: number };
   pluginSupported = false;
 
   constructor(execution: NativeExecutionInput, timeoutMs: number) {
@@ -42,6 +43,19 @@ export class NativeExecutionContext {
     this.assertOpen();
   }
 
+  /** Bind only the coordinator lease actually acquired by this execution. */
+  bindNativeLease(owner: { bootId: string; leaseOwner: string; generation: number }): void {
+    this.assertOpen();
+    if (this.#nativeLease) { this.assertNativeLease(owner); return; }
+    this.#nativeLease = { ...owner };
+  }
+
+  assertNativeLease(owner: { bootId: string; leaseOwner: string; generation: number }): void {
+    this.assertOpen();
+    if (!this.#nativeLease || this.#nativeLease.bootId !== owner.bootId || this.#nativeLease.leaseOwner !== owner.leaseOwner ||
+        this.#nativeLease.generation !== owner.generation) throw new Error("native_execution_context_owner_changed");
+  }
+
   async wait<T>(operation: () => Promise<T>): Promise<T> {
     this.assertOpen();
     let remove = () => {};

server/src/services/native-runtime/native-runner-checkpoint.ts

@@ -4,6 +4,7 @@ import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunn
 import { inspectDurablePrpCheckpointHeader, inspectUnclaimedBootstrapCheckpoint, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js";
 import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js";
 import { legacyControllerBootId } from "../legacy-controller-lease.js";
+import { assertNativeSourceCheckpointConsumer, lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js";
 
 type Binding = { companyId: string; issueId: string; runId: string };
 export const NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES = 1024 * 1024;
@@ -85,6 +86,7 @@ export async function validateNativeRunnerCheckpointOwner(
   const activeOwner = owner.leaseOwner !== null && current.leaseOwner === owner.leaseOwner
     && current.leaseExpiresAt !== null && current.leaseExpiresAt > new Date(owned.now);
   if (!activeOwner) {
+    await assertNativeSourceCheckpointConsumer(tx, owner);
     const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues)
       .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId)));
     const [successor] = task?.executionRunId && task.executionRunId !== owner.runId
@@ -119,6 +121,10 @@ export function nativeRunnerCheckpointWriter(db: Db, owner: NativeRunnerCheckpoi
     if (checkpoint.runId !== owner.runId) throw new Error("native_runner_checkpoint_binding_mismatch");
     const changed = encryptParts(owner, checkpoint.changedRecords);
     await persistenceDb.transaction(async tx => {
+      if (owner.runId !== owner.authorityRunId) {
+        await lockSourceRecoveryCoordinators(tx, owner.companyId, [owner.runId, owner.authorityRunId]);
+        await assertNativeSourceCheckpointConsumer(tx, owner, checkpoint);
+      }
       await assertNativeRunnerCheckpointOwner(tx, owner, checkpoint.workCommandIds.some(id => !previousWork.has(id)));
       if (checkpoint.predecessor) {
         const [prior] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(

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

@@ -164,6 +164,7 @@ type BackendFactoryOptions = {
 };
 
 type RunnerTransportOptions = {
+  authorityRunId?: string;
   onAccountingEvent?: (event: PrpEvent) => Promise<void>;
   onRecoveryAccountingEvent?: unknown;
   environment?: NodeJS.ProcessEnv;
@@ -7924,7 +7925,8 @@ describe("native terminal-turn accounting", () => {
       binding: { ...execution.binding, runId: "plugin-usage-run", executionWorkspaceId: "plugin-usage-workspace" },
       session: { ...execution.session, normalizedSessionId: "plugin-usage-session" } } as NativeExecutionInput;
     const launcher = vi.fn(), registration = vi.fn();
-    const driver = { prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration };
+    let authorityRunId = current.binding.runId;
+    const driver = { get authorityRunId() { return authorityRunId; }, prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration };
     const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue(
       { launcher, registration, driver, readWorkspaceFile: vi.fn(), isRetired: () => false, rebind: vi.fn() } as never);
     const task = vi.fn(async (operation: unknown) => {
@@ -7944,6 +7946,9 @@ describe("native terminal-turn accounting", () => {
       await options.onAccountingEvent!(report);
       expect(capture).toHaveBeenCalledExactlyOnceWith(report);
       expect(options.onRecoveryAccountingEvent).toEqual(expect.any(Function));
+      expect(options.authorityRunId).toBe(current.binding.runId);
+      authorityRunId = "next-plugin-owner";
+      expect(options.authorityRunId).toBe("next-plugin-owner");
     } finally {
       owner.mockRestore();
       if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;

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

@@ -13595,7 +13595,7 @@ async function createRunnerdBackendWithinSessionClaim(
         ...(input.recoveryPurpose === "stop_cleanup" ? { recoveryPurpose: "stop_cleanup" as const } : {}),
         measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined,
         recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined,
-        ...owner.driver, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? description.runner,
+        ...owner.driver, get authorityRunId() { return owner.driver.authorityRunId; }, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? description.runner,
         controlPlaneRegistration: input.controllerPersistence ? async (authority, identity, signal) => {
           await nativeRunnerCheckpointRegistration({
             ...nativeRunnerCheckpointPersistence(input.db, input.controllerPersistence!.bootId,

server/src/services/native-runtime/native-source-recovery-ownership.ts

@@ -0,0 +1,44 @@
+import { and, asc, eq, inArray, sql } from "drizzle-orm";
+import { environmentLeases, nativeRunFinalizations, type Db } from "@paperclipai/db";
+import { inspectDurablePrpCheckpointHeader, type DurablePrpCheckpoint } from "../../vendor/paperclip-runner/index.js";
+
+type Transaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
+const record = (value: unknown): Record<string, unknown> => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
+
+/** Claims and recovered-source writers lock both coordinators in this order. */
+export async function lockSourceRecoveryCoordinators(tx: Transaction, companyId: string, runIds: readonly string[]) {
+  return tx.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId),
+    inArray(nativeRunFinalizations.runId, [...new Set(runIds)]))).orderBy(asc(nativeRunFinalizations.runId)).for("update");
+}
+
+export async function readNativeSourceConsumers(tx: Transaction, companyId: string, runId: string) {
+  return tx.select({ id: environmentLeases.id, metadata: environmentLeases.metadata }).from(environmentLeases).where(and(
+    eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, runId),
+    sql`${environmentLeases.metadata}->'runnerTaskSourceConsumer' is not null`));
+}
+
+/** Physical source consumption revokes the completed controller's late writer. */
+export async function assertNativeSourceCheckpointConsumer(tx: Transaction, owner: {
+  companyId: string; runId: string; authorityRunId: string; bootId: string; generation: number;
+}, checkpoint?: DurablePrpCheckpoint) {
+  const rows = await readNativeSourceConsumers(tx, owner.companyId, owner.runId);
+  if (!rows.length) return;
+  const deny = (): never => { throw new Error("native_runner_checkpoint_owner_lost"); };
+  if (owner.runId === owner.authorityRunId || rows.length !== 1 || !checkpoint) return deny();
+  const source = rows[0]!, consumer = record(source.metadata?.runnerTaskSourceConsumer);
+  if (typeof consumer.leaseId !== "string" || consumer.runId !== owner.authorityRunId) return deny();
+  const [lease] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, owner.companyId),
+    eq(environmentLeases.id, consumer.leaseId)));
+  const recovery = record(lease?.metadata?.runnerTaskSourceRecovery);
+  const header = checkpoint.records.get("header");
+  const identity = header ? inspectDurablePrpCheckpointHeader(header) : null;
+  const expected = record(recovery.identity);
+  if (!lease || lease.heartbeatRunId !== owner.authorityRunId || lease.status !== "active" || lease.releasedAt ||
+      lease.expiresAt && lease.expiresAt.getTime() <= Date.now() ||
+      recovery.schema !== "paperclip.runner.source-recovery.v1" || recovery.sourceLeaseId !== source.id ||
+      recovery.consumerLeaseId !== lease.id || recovery.consumerRunId !== owner.authorityRunId ||
+      typeof recovery.accountIdentity !== "string" || recovery.accountIdentity !== consumer.accountIdentity ||
+      recovery.controllerBootId !== owner.bootId || recovery.controllerGeneration !== owner.generation ||
+      ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].some(key =>
+        typeof expected[key] !== "string" || expected[key] !== record(identity)[key] || expected[key] !== record(consumer.identity)[key])) return deny();
+}

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

@@ -0,0 +1,83 @@
+import { existsSync } from "node:fs";
+import { mkdtemp, readFile, rm } from "node:fs/promises";
+import { spawn } from "node:child_process";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { expect, it } from "vitest";
+import { createRunnerdCodexTransport, createRunnerProviderLaunch, defaultCapabilityRunnerdBinary,
+  runnerLaunchRequirementsDigest, readRunnerdArtifactBinding, type RunnerProcessLaunchSpec, type RunnerProviderLaunchRequirements } from "@paperclipai/paperclip-runner";
+import { createPluginRunnerTaskOwner } from "./plugin-runner-task.js";
+
+const fakeCodex = resolve(import.meta.dirname, "../../../../packages/paperclip-runner/runner/target/debug/fake-codex-app-server");
+
+it.skipIf(!existsSync(defaultCapabilityRunnerdBinary()) || !existsSync(fakeCodex))(
+  "keeps the current plugin owner through a real ordinary warm attachment", async () => {
+    const root = await mkdtemp(join(tmpdir(), "plugin-warm-owner-")), calls = join(root, "calls");
+    const source = { runnerInstanceId: "warm-runner", environmentLeaseId: "warm-lease", runId: "first-run",
+      normalizedSessionId: "warm-session", turnId: "first-turn", itemId: "first-item" };
+    const target = { ...source, runId: "second-run", turnId: "second-turn", itemId: "second-item" };
+    let spec: RunnerProcessLaunchSpec | undefined, child: ReturnType<typeof spawn> | undefined;
+    let submits = 0, stopped = false;
+    const task = async (operation: unknown): Promise<unknown> => {
+      const op = operation as { kind: string; requirements: RunnerProviderLaunchRequirements; configurationId: string; provider: string };
+      if (op.kind === "prepare") return { kind: "prepared", taskId: "warm-task", launch: {
+        configurationId: `configuration-${op.requirements.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(op.requirements),
+        provider: createRunnerProviderLaunch(op.requirements, { options: { codexArgs: ["--state-file", join(root, "fake.json"), "--call-log", calls, "--record-process-start"] },
+          root, cwd: root, codexHome: join(root, "codex-home"), codexCommand: fakeCodex, providerNodeCommand: process.execPath,
+          opencodeProxyPath: "", acpxSidecarPath: "" }),
+      } };
+      if (op.kind === "check_project_auth") return { kind: "project_auth", taskId: "warm-task", configurationId: op.configurationId, provider: op.provider, allowed: true };
+      if (op.kind === "submit") {
+        submits++;
+        child = spawn(spec!.command, [...spec!.args], { cwd: spec!.cwd, env: spec!.environment, stdio: "ignore", detached: true });
+      }
+      if (op.kind === "stop") { stopped = true; if (child?.pid) process.kill(-child.pid, "SIGTERM"); }
+      if (op.kind === "status") return { kind: "status", taskId: "warm-task", phase: stopped ? "cancelled" : "running",
+        executionStopped: stopped && child?.exitCode != null, sessionStateAvailable: false };
+      if (op.kind === "describe") return { kind: "description", taskId: "warm-task", warmTaskVersion: 1,
+        runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
+      return { kind: "accepted", taskId: "warm-task" };
+    };
+    const input = { taskId: "warm-task", task, harness: "codex", projectIds: [], projectAuthCheckVersion: 1 as const,
+      authorityRunId: source.runId, accountIdentity: "synthetic-account", warmTaskVersion: 1 as const, pollMs: 5 };
+    const owner = createPluginRunnerTaskOwner(input);
+    const launcher = owner.driver.runnerProcessLauncher!;
+    const cores = new Set<import("@paperclipai/paperclip-runner").DurablePrpControlPlane>();
+    const bundle = createRunnerdCodexTransport({ ...owner.driver,
+      get authorityRunId() { return owner.driver.authorityRunId; },
+      runnerProcessLauncher: next => { spec = next; return launcher(next); },
+      stateDirectory: root, runnerBinary: defaultCapabilityRunnerdBinary(), runnerArtifact: readRunnerdArtifactBinding(defaultCapabilityRunnerdBinary()), prpIdentity: source,
+      codexCommand: fakeCodex, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 },
+      // This fixture isolates ordinary owner rotation from hosted ingress timing.
+      controlPlaneRegistration: async core => {
+        if (!cores.has(core)) { cores.add(core); await core.start(); }
+        return { connectUrl: core.connectUrl, release: () => {} };
+      },
+    });
+    try {
+      const opened = await bundle.transport.request("thread/start", { cwd: root, dynamicTools: [] });
+      const turn = async () => {
+        await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Complete once." }] });
+        for await (const event of bundle.transport.notifications()) if (event.method === "turn/completed") return;
+        throw new Error("Warm fixture ended without completion");
+      };
+      await turn();
+      const pid = child!.pid;
+      owner.rebind({ ...input, authorityRunId: target.runId });
+      await bundle.transport.attachRun!(target);
+      await turn();
+      const retained = await bundle.transport.request("thread/read", {});
+      expect(retained.thread).toMatchObject({ id: (opened.thread as { id: string }).id });
+      expect(child!.pid).toBe(pid); expect(submits).toBe(1);
+      const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/);
+      expect(methods.filter(method => method === "process-start")).toHaveLength(1);
+      expect(methods.filter(method => method === "turn/start")).toHaveLength(2);
+    } finally {
+      const providerPid = bundle.evidence().providerPid;
+      await bundle.detachControllerForRestart().catch(() => undefined);
+      for (const pid of [providerPid, child?.pid]) if (pid) try { process.kill(-pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; }
+      for (const core of cores) await core.stop();
+      await rm(root, { recursive: true, force: true });
+    }
+  }, 30_000,
+);

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

@@ -59,6 +59,81 @@ describe("plugin native Runner ownership", () => {
     expect(() => createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" } as never)).toThrow("physical task unavailable");
   });
 
+  it("publishes current trusted recovery authority through warm owner rebind", () => {
+    const { task } = provider();
+    const input = { taskId: "task", task, projectIds: [], harness: "codex", authorityRunId: "first-owner", codexSourceRecoveryVersion: 1 as const, adoptIdentity: requirements.identity };
+    const owner = createPluginRunnerTaskOwner(input);
+    expect(owner.driver.authorityRunId).toBe("first-owner");
+    expect(owner.driver.codexSourceRecoveryVersion).toBe(1);
+    owner.rebind({ ...input, authorityRunId: "next-owner", codexSourceRecoveryVersion: undefined });
+    expect(owner.driver.authorityRunId).toBe("next-owner");
+    expect(owner.driver.codexSourceRecoveryVersion).toBeUndefined();
+  });
+  it.each(["capability", "source", "identity", "account", "provider", "adoption"])("refuses an unqualified predecessor preparation: %s", async mode => {
+    const { task } = provider();
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: mode === "provider" ? "claude" : "codex",
+      authorityRunId: "next", codexSourceRecoveryVersion: mode === "capability" ? undefined : 1,
+      resumeFrom: mode === "source" ? undefined : "source-task", priorIdentity: mode === "identity" ? undefined : requirements.identity,
+      accountIdentity: mode === "account" ? undefined : "selected-account", ...(mode === "adoption" ? { adoptIdentity: requirements.identity } : {}),
+    });
+    await expect(prepare(owner)).rejects.toThrow("source reconciliation unavailable");
+    expect(task).not.toHaveBeenCalled();
+  });
+  it("reuses exact adopted A preparation under current B without a source launch", async () => {
+    const { task } = provider();
+    const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "next" } } as never, 1_000);
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1,
+      executionContext: context, adoptIdentity: requirements.identity });
+    const recovery = { configurationId: "retained-configuration", requirementsDigest: `sha256:${"a".repeat(64)}` as const, provider: { kind: "codex" } };
+    try {
+      await expect(owner.driver.prepareLaunch!(requirements, recovery)).resolves.toEqual(recovery);
+      expect(task.mock.calls.map(([operation]) => operation.kind)).toEqual(["check_project_auth"]);
+    } finally { context.finish(); }
+  });
+  it.each(["same-run", "suspended-predecessor"])("keeps ordinary %s preparation independent of predecessor capability", async mode => {
+    const { task } = provider();
+    const identity = mode === "same-run" ? requirements.identity : { ...requirements.identity, runId: "next" };
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1,
+      authorityRunId: identity.runId, priorIdentity: requirements.identity, resumeFrom: "source-task" });
+    await prepare(owner, { ...requirements, identity });
+    expect(task.mock.calls[0][0]).toMatchObject({ kind: "prepare", resumeFrom: "source-task", requirements: { identity } });
+    expect(task.mock.calls[0][0]).not.toHaveProperty("recoveryPurpose");
+  });
+  it("keeps the resume parent on ordinary repeated preparation after launch", async () => {
+    const { task } = provider();
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, pollMs: 1,
+      authorityRunId: "run", resumeFrom: "source-task", priorIdentity: requirements.identity });
+    await prepare(owner);
+    const registration = await owner.registration({} as never, "run");
+    try {
+      owner.launcher(launchSpec());
+      await vi.waitFor(() => expect(task.mock.calls.some(([op]) => op.kind === "submit")).toBe(true));
+      await prepare(owner);
+      expect(task.mock.calls.map(([op]) => op).filter(op => op.kind === "prepare").at(-1)).toMatchObject({ resumeFrom: "source-task" });
+    } finally { await registration.release("detach"); }
+  });
+  it("prepares and submits exact A, then prepares B without copying or reconciling A again", async () => {
+    const { task } = provider();
+    const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, pollMs: 1,
+      authorityRunId: "next", codexSourceRecoveryVersion: 1, resumeFrom: "source-task", priorIdentity: requirements.identity,
+      accountIdentity: "selected-account" });
+    await expect(prepare(owner, { ...requirements, identity: { ...requirements.identity, runId: "foreign" } })).rejects.toThrow("source reconciliation unavailable");
+    await prepare(owner);
+    const registration = await owner.registration({} as never, "run");
+    try {
+      owner.launcher(launchSpec());
+      await vi.waitFor(() => expect(task.mock.calls.some(([op]) => op.kind === "submit")).toBe(true));
+      const initial = task.mock.calls.map(([op]) => op).filter(op => ["prepare", "submit"].includes(op.kind));
+      expect(initial).toEqual([expect.objectContaining({ kind: "prepare", resumeFrom: "source-task", requirements: expect.objectContaining({ identity: requirements.identity }) }),
+        expect.objectContaining({ kind: "submit", resumeFrom: "source-task", runner: expect.objectContaining({ runId: "run", turnId: "turn", itemId: "item" }) })]);
+      await prepare(owner);
+      expect(task.mock.calls.map(([op]) => op).filter(op => op.kind === "prepare").at(-1)).toMatchObject({ resumeFrom: "source-task" });
+      await prepare(owner, { ...requirements, identity: { ...requirements.identity, runId: "next", turnId: "next-turn", itemId: "next-item" } });
+      const attached = task.mock.calls.map(([op]) => op).filter(op => op.kind === "prepare").at(-1);
+      expect(attached).toMatchObject({ priorIdentity: requirements.identity, requirements: { identity: { runId: "next" } } });
+      expect(attached).not.toHaveProperty("recoveryPurpose"); expect(attached).not.toHaveProperty("resumeFrom");
+    } finally { await registration.release("detach"); }
+  });
   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);

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

@@ -36,6 +36,9 @@ export function createPluginRunnerTaskOwner(input: {
   projectAuthCheckVersion?: 1;
   resumeFrom?: string;
   priorIdentity?: DurableRecoveryIdentity;
+  /** Current authenticated lease owner; the restored PRP identity may belong to its predecessor. */
+  authorityRunId?: string;
+  codexSourceRecoveryVersion?: 1;
   credentialIdentity?: string;
   accountIdentity?: string;
   credential?: PluginEnvironmentTaskCredential;
@@ -55,6 +58,7 @@ export function createPluginRunnerTaskOwner(input: {
   let preparedLaunchId: string | undefined;
   let preparedIdentity: DurableRecoveryIdentity | undefined = input.adoptIdentity;
   let launched = !!input.adoptIdentity;
+  let launchIdentity = input.adoptIdentity;
   let stopRequested = false;
   let released = false;
   let submitted = !!input.adoptIdentity;
@@ -119,6 +123,13 @@ export function createPluginRunnerTaskOwner(input: {
       await authorizeRunnerAdoption(recovery.configurationId, requirements.identity);
       return recovery;
     }
+    const authorityRunId = input.authorityRunId ?? input.executionContext?.execution.binding.runId;
+    const sourceRecovery = !!authorityRunId && requirements.identity.runId !== authorityRunId;
+    if (sourceRecovery && (input.codexSourceRecoveryVersion !== 1 || !input.resumeFrom || !input.priorIdentity || !input.accountIdentity ||
+        input.harness !== "codex" || input.adoptIdentity || Object.entries(input.priorIdentity).some(([key, value]) =>
+          requirements.identity[key as keyof DurableRecoveryIdentity] !== value))) {
+      throw new Error("Plugin Runner source reconciliation unavailable");
+    }
     if (input.githubBroker) {
       const description = await call({ kind: "describe" });
       if (description.kind !== "description" || description.githubCredentialBrokerVersion !== 1) {
@@ -128,7 +139,8 @@ export function createPluginRunnerTaskOwner(input: {
     const result = await call(environmentTaskOperationSchema.parse({ kind: "prepare", requirements,
       ...(input.githubBroker ? { githubBroker: input.githubBroker } : {}),
       projectIds: [...input.projectIds], ...(recovery ? { recovery } : {}),
-      ...(input.resumeFrom ? { resumeFrom: input.resumeFrom } : {}),
+      ...(input.resumeFrom && (!authorityRunId || !launchIdentity || launchIdentity.runId === authorityRunId || sourceRecovery)
+        ? { resumeFrom: input.resumeFrom } : {}),
       ...(input.accountIdentity ? { accountIdentity: input.accountIdentity } : {}),
       ...(launched && preparedIdentity && requirements.identity.runId !== preparedIdentity.runId
         ? { priorIdentity: preparedIdentity } : {}),
@@ -202,6 +214,7 @@ export function createPluginRunnerTaskOwner(input: {
         reconnectGraceMs: native.reconnectGraceMs, lifecyclePolicy: native.lifecyclePolicy,
       },
     });
+    launchIdentity = native.identity;
     launched = true;
     const child: RunnerProcessHandle["child"] = { exitCode: null, signalCode: null,
       kill: () => {
@@ -320,6 +333,8 @@ export function createPluginRunnerTaskOwner(input: {
     };
   };
   const driver: RunnerExecutionDriver = {
+    get authorityRunId() { return input.authorityRunId ?? input.executionContext?.execution.binding.runId; },
+    get codexSourceRecoveryVersion() { return input.codexSourceRecoveryVersion; },
     prepareLaunch, authorizeRunnerAdoption, readRunnerState, runnerProcessLauncher: launcher,
     // The provider resolves the authorized retained task before each state operation.
     prepareExternalRunnerState: async () => {},

server/src/services/native-runtime/plugin-task-source-recovery.integration.test.ts

@@ -0,0 +1,337 @@
+import { randomUUID } from "node:crypto";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+import { agents, companies, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations,
+  nativeRunResults, nativeRunnerCheckpoints, plugins } from "@paperclipai/db";
+import { DurablePrpControlPlane, type DurablePrpCheckpoint, type NativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
+import { recoveryPostgresSupport, startRecoveryPostgres } from "../../__tests__/helpers/recovery-postgres.js";
+import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js";
+import { executeEnvironmentTask, resolvePluginTaskResume } from "../environment-task-runtime.js";
+import type { PluginWorkerManager } from "../plugin-worker-manager.js";
+import { remoteTerminationReceipt } from "../remote-execution-termination.js";
+import { NativeExecutionContext } from "./native-execution-context.js";
+import { buildNativeExecutionInput } from "./native-execution-input.js";
+import { currentNativeControllerIdentity } from "./native-restart-recovery.js";
+import { loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
+import { claimPluginTaskSourceRecovery } from "./plugin-task-source-recovery.js";
+import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
+import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
+
+describe.skipIf(!recoveryPostgresSupport.supported)("stopped source admission under the current native owner", () => {
+  let database: Awaited<ReturnType<typeof startRecoveryPostgres>>, db: ReturnType<typeof createDb>;
+  let home: ReturnType<typeof createIsolatedTestHome>, root: string;
+  const originalHome = process.env.PAPERCLIP_HOME;
+  const contexts: NativeExecutionContext[] = [];
+  beforeAll(async () => {
+    home = createIsolatedTestHome(); process.env.PAPERCLIP_HOME = home.home;
+    root = await mkdtemp(join(tmpdir(), "plugin-source-recovery-"));
+    database = await startRecoveryPostgres("plugin-source-recovery-"); db = createDb(database.connectionString);
+  }, 60_000);
+  afterEach(() => { for (const context of contexts.splice(0)) context.finish(); });
+  afterAll(async () => {
+    await database?.cleanup(); if (root) await rm(root, { recursive: true, force: true }); home?.cleanup();
+    if (originalHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = originalHome;
+  });
+
+  async function fixture() {
+    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), sourceRunId = randomUUID(), runId = randomUUID();
+    const sourceLeaseId = randomUUID(), leaseId = randomUUID(), environmentId = randomUUID(), pluginId = randomUUID();
+    const controller = await currentNativeControllerIdentity(), oldOwner = { bootId: controller.bootId, generation: 1, leaseOwner: randomUUID() };
+    await db.insert(companies).values({ id: companyId, name: "Source admission", issuePrefix: `S${companyId.slice(0, 6)}` });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Source worker", adapterType: "paperclip_runner" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Continue once", status: "in_progress", assigneeAgentId: agentId });
+    const [sourceRun] = await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId, nativeIssueId: issueId,
+      status: "running", runtimeMode: "native" }).returning();
+    const native = await prepareNativeHeartbeatRun({ db, run: sourceRun, issue: { id: issueId, title: "Continue once", description: null }, environmentLeaseId: sourceLeaseId });
+    const [prepared] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, sourceRunId));
+    const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.id, prepared.completionContractId!));
+    const build = (id: string) => buildNativeExecutionInput({ companyId, agentId, runId: id, normalizedSessionId: native.normalizedSessionId,
+      provider: "codex", issue: { id: issueId, identifier: "SOURCE-1", title: "Continue once", description: null, workMode: "standard" },
+      taskPrompt: "Continue once", workspace: { id, cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
+      completionContract: { id: prepared.completionContractId!, sha256: prepared.completionContractSha256!, schemaVersion: "paperclip.run-result.v1", contract: contract.contractJson as never },
+      runtimeContext: nativeRuntimeContextFixture() });
+    const sourceExecution = build(sourceRunId), execution = build(runId);
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { ...prepared.runnerProfileJson, nativeExecutionInput: sourceExecution } }).where(eq(heartbeatRuns.id, sourceRunId));
+    await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: sourceRunId, phase: "observed", controllerBootId: oldOwner.bootId,
+      controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: oldOwner.generation,
+      leaseOwner: oldOwner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) });
+    const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: sourceLeaseId, runId: sourceRunId,
+      normalizedSessionId: native.normalizedSessionId, turnId: `turn-${sourceRunId}`, itemId: `item-${sourceRunId}` };
+    const oldWriter = nativeRunnerCheckpointWriter(db, { companyId, issueId, runId: sourceRunId, authorityRunId: sourceRunId, ...oldOwner });
+    let checkpoint!: DurablePrpCheckpoint;
+    const core = new DurablePrpControlPlane({ stateDirectory: join(root, sourceRunId), identity,
+      expectedRunnerVersion: "source-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
+    core.setCheckpointPersistence(async value => { checkpoint = value; await oldWriter(value); });
+    await core.flushCheckpoint(); await core.stop();
+    const [result] = await db.insert(nativeRunResults).values({ companyId, issueId, runId: sourceRunId,
+      completionContractId: prepared.completionContractId!, serverFingerprint: "accepted-source", schemaStatus: "accepted",
+      resultJson: { summary: "Accepted once" }, canonicalSha256: "accepted-source" }).returning();
+    await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, sourceRunId));
+    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(Date.now() - 1_000), resultJson: { summary: "Accepted once" } }).where(eq(heartbeatRuns.id, sourceRunId));
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, status: "running", runtimeMode: "native",
+      nativeSessionId: native.normalizedSessionId, runnerProfileJson: { nativeExecutionInput: execution } });
+    const owner = { bootId: controller.bootId, generation: 3, leaseOwner: randomUUID() };
+    await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", controllerBootId: owner.bootId,
+      controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: owner.generation,
+      leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) });
+    await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
+    await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "source-fixture", version: "1.0.0", status: "ready",
+      manifestJson: { capabilities: ["environment.drivers.register"], environmentDrivers: [{ driverKey: "tasks", supportsTasks: true }] } as never });
+    await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "plugin", status: "active",
+      config: { pluginKey: pluginId, driverKey: "tasks" } });
+    const metadata = { driver: "plugin", pluginId, driverKey: "tasks", runnerTaskLease: true };
+    const sourceLease = { id: sourceLeaseId, companyId, heartbeatRunId: sourceRunId, provider: `plugin:${pluginId}:tasks`, providerLeaseId: "source-task" };
+    await db.insert(environmentLeases).values({ ...sourceLease, environmentId, issueId, leasePolicy: "reuse_by_environment", status: "released",
+      releasedAt: new Date(), cleanupStatus: "success", metadata: { ...metadata, runnerTaskIdentity: { runnerId: identity.runnerInstanceId,
+        leaseId: identity.environmentLeaseId, runId: sourceRunId, sessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: identity.itemId },
+        remoteExecutionTermination: remoteTerminationReceipt(sourceLease, { providerLeaseId: "source-task", state: "stopped" }) } });
+    await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, issueId, heartbeatRunId: runId,
+      provider: sourceLease.provider, providerLeaseId: "replacement-task", leasePolicy: "reuse_by_environment", status: "active", metadata });
+    const context = new NativeExecutionContext(execution, 60_000); context.bindNativeLease(owner); contexts.push(context);
+    const input = { companyId, leaseId, context, identity, resumeFrom: "source-task", selectedSourceRecoveryVersion: 1 as const,
+      accountIdentity: "current-selected-account", matchesExecution: (previous: typeof sourceExecution) => previous.provider.kind === execution.provider.kind && previous.workspace.cwd === execution.workspace.cwd };
+    const workers = { getWorker: () => ({ supportedMethods: ["environmentTask"] }), call: vi.fn(async (_id: string, _method: string, p: any) => {
+      if (p.operation.kind === "prepare") return { kind: "prepared", taskId: p.taskId,
+        launch: { configurationId: "source-configuration", requirementsDigest: `sha256:${"a".repeat(64)}`, provider: {} } };
+      if (p.operation.kind === "check_project_auth") return { kind: "project_auth", taskId: p.taskId, provider: "openai", configurationId: p.operation.configurationId, allowed: true };
+      return { kind: "accepted", taskId: p.taskId };
+    }) };
+    const operation = { kind: "prepare" as const, resumeFrom: input.resumeFrom,
+      accountIdentity: input.accountIdentity, projectIds: [], requirements: { schema: "paperclip.runner.launch-requirements.v1" as const,
+        identity, provider: "codex" as const, model: null, instructions: "Source policy", runtimeContext: null, settings: {}, approvalPolicy: "never" as const,
+        readOnly: false, preparedConversation: false, includeCollaborationModeInstructions: false } };
+    const task = (op: Parameters<typeof executeEnvironmentTask>[2]["operation"] = operation) => executeEnvironmentTask(db, workers as unknown as PluginWorkerManager,
+      { companyId, leaseId, executionContext: context, operation: op });
+    return { ...input, runId, sourceRunId, sourceLeaseId, environmentId, agentId, issueId, owner, oldOwner, oldWriter, checkpoint, execution, operation, workers, task };
+  }
+
+  it("authorizes only the claimed predecessor identity while preserving A's accepted outcome", async () => {
+    const f = await fixture();
+    const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
+    await expect(f.task()).rejects.toThrow("execution identity mismatch");
+    await claimPluginTaskSourceRecovery(db, f);
+    const attachment = { ...f.operation, resumeFrom: undefined, priorIdentity: f.identity,
+      requirements: { ...f.operation.requirements, identity: { ...f.identity, runId: f.runId, turnId: `turn-${f.runId}`, itemId: `item-${f.runId}` } } };
+    await expect(f.task(attachment)).rejects.toThrow("ownership_unverified");
+    expect(f.workers.call).not.toHaveBeenCalled();
+    await expect(f.task()).resolves.toMatchObject({ kind: "prepared" });
+    expect(f.workers.call.mock.calls[0][2]).toMatchObject({ runId: f.runId, operation: { requirements: { identity: f.identity } } });
+    await f.task({ kind: "check_project_auth", configurationId: "source-configuration", provider: "openai", projectIds: [] });
+    await expect(f.task({ kind: "submit", resumeFrom: f.resumeFrom,
+      accountIdentity: f.accountIdentity, preparedLaunchId: "source-configuration", projectIds: [], bootstrapTicket: "synthetic-test-ticket",
+      runner: { runnerId: f.identity.runnerInstanceId, leaseId: f.identity.environmentLeaseId, runId: f.sourceRunId,
+        sessionId: f.identity.normalizedSessionId, turnId: f.identity.turnId, itemId: f.identity.itemId, harness: "codex", protocolMin: 1, protocolMax: 1 },
+      launch: { runnerVersion: "source-fixture", runnerDigest: `sha256:${"a".repeat(64)}`,
+        maxOutboxBytes: 262144, p0ReserveBytes: 65536, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } },
+    })).resolves.toMatchObject({ kind: "accepted" });
+    expect(f.workers.call.mock.calls.at(-1)![2]).toMatchObject({ runId: f.runId, operation: {
+      kind: "submit", runner: { runId: f.sourceRunId },
+    } });
+    for (const key of ["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId"]) {
+      await expect(f.task({ ...attachment, requirements: { ...attachment.requirements,
+        identity: { ...attachment.requirements.identity, [key]: randomUUID() } } })).rejects.toThrow("ownership_unverified");
+    }
+    await expect(f.task({ ...attachment, accountIdentity: "different-account" })).rejects.toThrow("ownership_unverified");
+    await expect(f.task(attachment)).resolves.toMatchObject({ kind: "prepared" });
+    await expect(f.task(attachment)).resolves.toMatchObject({ kind: "prepared" });
+    expect(f.workers.call.mock.calls.at(-1)![2]).toMatchObject({ runId: f.runId, operation: {
+      kind: "prepare", priorIdentity: f.identity, requirements: { identity: { runId: f.runId } },
+    } });
+    await expect(f.task({ ...f.operation, resumeFrom: undefined })).rejects.toThrow("ownership_unverified");
+    const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
+    expect(after).toEqual(before);
+    const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId));
+    expect(JSON.stringify(lease.metadata)).not.toContain(f.owner.leaseOwner);
+  });
+
+  it("keeps naked mismatches ordinary and malformed claims fail closed", async () => {
+    const f = await fixture();
+    for (const claim of [undefined, false, {}, true, "untrusted"]) {
+      if (claim !== undefined) await db.update(environmentLeases).set({ metadata:
+        sql`${environmentLeases.metadata} || ${JSON.stringify({ runnerTaskSourceRecovery: claim })}::jsonb` }).where(eq(environmentLeases.id, f.leaseId));
+      await expect(f.task()).rejects.toThrow(claim ? "ownership_unverified" : "execution identity mismatch");
+    }
+    expect(f.workers.call).not.toHaveBeenCalled();
+  });
+
+  it("requires a complete claimed launch before any predecessor RPC", async () => {
+    const f = await fixture(); await claimPluginTaskSourceRecovery(db, f);
+    for (const operation of [
+      { ...f.operation, resumeFrom: undefined }, { ...f.operation, accountIdentity: undefined },
+      { ...f.operation, priorIdentity: f.identity },
+      { ...f.operation, requirements: { ...f.operation.requirements, provider: "acpx" as const } },
+    ]) await expect(f.task(operation)).rejects.toThrow("ownership_unverified");
+    const submit = { kind: "submit" as const, resumeFrom: f.resumeFrom, accountIdentity: f.accountIdentity,
+      preparedLaunchId: "source-configuration", projectIds: [], bootstrapTicket: "synthetic-test-ticket",
+      runner: { runnerId: f.identity.runnerInstanceId, leaseId: f.identity.environmentLeaseId, runId: f.sourceRunId,
+        sessionId: f.identity.normalizedSessionId, turnId: f.identity.turnId, itemId: f.identity.itemId, harness: "codex", protocolMin: 1, protocolMax: 1 },
+      launch: { runnerVersion: "source-fixture", runnerDigest: `sha256:${"a".repeat(64)}`,
+        maxOutboxBytes: 262144, p0ReserveBytes: 65536, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } };
+    for (const operation of [
+      { ...submit, resumeFrom: undefined }, { ...submit, accountIdentity: undefined },
+      { ...submit, preparedLaunchId: undefined }, { ...submit, launch: undefined },
+      { ...submit, runner: { ...submit.runner, harness: "claude" } },
+    ]) await expect(f.task(operation)).rejects.toThrow("ownership_unverified");
+    expect(f.workers.call).not.toHaveBeenCalled();
+  });
+
+  it.each(["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"] as const)("rejects a different source %s", async key => {
+    const f = await fixture();
+    await expect(claimPluginTaskSourceRecovery(db, { ...f, identity: { ...f.identity, [key]: randomUUID() } })).rejects.toThrow();
+    expect(f.workers.call).not.toHaveBeenCalled();
+  });
+
+  it.each(["before-claim", "after-claim"])("rejects a stale same-boot coordinator context: %s", async point => {
+    const f = await fixture(); if (point === "after-claim") await claimPluginTaskSourceRecovery(db, f);
+    await db.update(nativeRunFinalizations).set({ controllerGeneration: f.owner.generation + 1, leaseOwner: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.runId));
+    if (point === "before-claim") await expect(claimPluginTaskSourceRecovery(db, f)).rejects.toThrow("context_owner_changed");
+    else await expect(f.task()).rejects.toThrow("context_owner_changed");
+    expect(f.workers.call).not.toHaveBeenCalled();
+  });
+
+  it("rejects a worker reply after the same controller has lost its acquired generation", async () => {
+    const f = await fixture(); await claimPluginTaskSourceRecovery(db, f);
+    const perform = f.workers.call.getMockImplementation()!;
+    f.workers.call.mockImplementation(async (...args) => {
+      const result = await perform(...args);
+      await db.update(nativeRunFinalizations).set({ controllerGeneration: f.owner.generation + 1, leaseOwner: randomUUID() })
+        .where(eq(nativeRunFinalizations.runId, f.runId));
+      return result;
+    });
+    await expect(f.task()).rejects.toThrow("reconcile the same task");
+    await expect(f.task()).rejects.toThrow("context_owner_changed");
+    expect(f.workers.call).toHaveBeenCalledTimes(1);
+  });
+
+  it.each(["capability", "expired", "paused", "source-live", "driver", "header", "source-profile", "current-profile", "workspace-export", "legacy-settlement"])("holds an ineligible recovery: %s", async mode => {
+    const f = await fixture();
+    if (mode === "expired") await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId));
+    if (mode === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
+    if (mode === "source-live") await db.update(environmentLeases).set({ cleanupStatus: null }).where(eq(environmentLeases.id, f.sourceLeaseId));
+    if (mode === "driver") await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || '{"driverKey":"other"}'::jsonb` }).where(eq(environmentLeases.id, f.sourceLeaseId));
+    if (mode === "header") await db.update(nativeRunnerCheckpoints).set({ material: {} }).where(eq(nativeRunnerCheckpoints.runId, f.sourceRunId));
+    if (mode === "source-profile" || mode === "current-profile") await db.update(heartbeatRuns).set({ runnerProfileJson:
+      sql`jsonb_set(${heartbeatRuns.runnerProfileJson}, '{nativeExecutionInput,binding,issueId}', ${JSON.stringify(randomUUID())}::jsonb)` })
+      .where(eq(heartbeatRuns.id, mode === "source-profile" ? f.sourceRunId : f.runId));
+    if (mode === "workspace-export") await db.update(heartbeatRuns).set({ runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} || ${JSON.stringify({ nativeWorkspaceSync: {
+      schema: "paperclip.native-workspace-sync/v3", state: "prepared", leaseId: f.sourceLeaseId, providerLeaseId: f.resumeFrom,
+      workspaceId: f.issueId, remoteCwd: "/workspace", resourceDisposition: "keep_running", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null,
+    } })}::jsonb` }).where(eq(heartbeatRuns.id, f.sourceRunId));
+    if (mode === "legacy-settlement") await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({ legacyShutdownWorkspaceSettlement: {
+      schema: "paperclip.legacy-shutdown-workspace-settlement.v1", state: "pending", runId: f.sourceRunId, deadlineAt: new Date(Date.now() + 60_000).toISOString(),
+    } })}::jsonb` }).where(eq(environmentLeases.id, f.sourceLeaseId));
+    await expect(claimPluginTaskSourceRecovery(db, { ...f, ...(mode === "capability" ? { selectedSourceRecoveryVersion: undefined } : {}) })).rejects.toThrow();
+    const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.sourceLeaseId));
+    expect(source.metadata?.runnerTaskSourceConsumer).toBeUndefined(); expect(f.workers.call).not.toHaveBeenCalled();
+  });
+
+  it("pins the consumer and current account across retries", async () => {
+    const f = await fixture(), claim = await claimPluginTaskSourceRecovery(db, f);
+    expect(await claimPluginTaskSourceRecovery(db, f)).toEqual(claim);
+    await expect(claimPluginTaskSourceRecovery(db, { ...f, accountIdentity: "different-account" })).rejects.toThrow();
+    await expect(f.task({ ...f.operation, accountIdentity: "different-account" })).rejects.toThrow();
+    const competingLease = randomUUID();
+    const [original] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId));
+    await db.insert(environmentLeases).values({ ...original, id: competingLease, providerLeaseId: "other-replacement" });
+    await expect(claimPluginTaskSourceRecovery(db, { ...f, leaseId: competingLease })).rejects.toThrow();
+    await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || '{"runnerTaskSessionRetained":true}'::jsonb` })
+      .where(eq(environmentLeases.id, f.sourceLeaseId));
+    await expect(resolvePluginTaskResume(db, { ...f, priorIdentity: f.identity })).resolves.toBe(f.resumeFrom);
+    await expect(resolvePluginTaskResume(db, { ...f, leaseId: competingLease, priorIdentity: f.identity })).rejects.toThrow("not authorized");
+  });
+
+  it("keeps the first source claim after dispatch, even when another source is eligible", async () => {
+    const f = await fixture();
+    const first = await claimPluginTaskSourceRecovery(db, f);
+    await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || '{"runnerTaskDispatched":true,"runnerTaskResumeFrom":"source-task"}'::jsonb` })
+      .where(eq(environmentLeases.id, f.leaseId));
+    const alternateRunId = randomUUID(), alternateLeaseId = randomUUID(), alternateTaskId = "alternate-task";
+    const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
+    const original = source.runnerProfileJson!.nativeExecutionInput as NativeExecutionInput;
+    const execution = { ...original, binding: { ...original.binding, runId: alternateRunId } };
+    await db.insert(heartbeatRuns).values({ ...source, id: alternateRunId, runnerProfileJson: { ...source.runnerProfileJson, nativeExecutionInput: execution } });
+    const [result] = await db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: alternateRunId,
+      completionContractId: source.completionContractId!, serverFingerprint: "accepted-alternate", schemaStatus: "accepted",
+      resultJson: { summary: "Accepted alternate" }, canonicalSha256: "accepted-alternate" }).returning();
+    const [priorOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+    await db.insert(nativeRunFinalizations).values({ ...priorOwner, runId: alternateRunId, resultId: null,
+      phase: "observed", leaseOwner: f.oldOwner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) });
+    const identity = { ...f.identity, runId: alternateRunId, turnId: `turn-${alternateRunId}`, itemId: `item-${alternateRunId}` };
+    const core = new DurablePrpControlPlane({ stateDirectory: join(root, alternateRunId), identity,
+      expectedRunnerVersion: "source-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
+    core.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, { companyId: f.companyId, issueId: f.issueId,
+      runId: alternateRunId, authorityRunId: alternateRunId, ...f.oldOwner }));
+    try { await core.flushCheckpoint(); } finally { await core.stop(); }
+    await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null })
+      .where(eq(nativeRunFinalizations.runId, alternateRunId));
+    const [sourceLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.sourceLeaseId));
+    const alternateLease = { ...sourceLease, id: alternateLeaseId, heartbeatRunId: alternateRunId, providerLeaseId: alternateTaskId };
+    await db.insert(environmentLeases).values({ ...alternateLease, metadata: { ...sourceLease.metadata,
+      runnerTaskSourceConsumer: undefined, runnerTaskIdentity: { runnerId: identity.runnerInstanceId, leaseId: identity.environmentLeaseId,
+        runId: identity.runId, sessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: identity.itemId },
+      remoteExecutionTermination: remoteTerminationReceipt(alternateLease, { providerLeaseId: alternateTaskId, state: "stopped" }) } });
+    expect(await resolvePluginTaskResume(db, { ...f, priorIdentity: identity, inspectCheckpoint: true })).toBe(alternateTaskId);
+    await expect(claimPluginTaskSourceRecovery(db, { ...f, identity, resumeFrom: alternateTaskId })).rejects.toThrow("ownership_unverified");
+    const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId));
+    expect(lease.metadata?.runnerTaskSourceRecovery).toEqual(first);
+    const [unconsumed] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, alternateLeaseId));
+    expect(unconsumed.metadata?.runnerTaskSourceConsumer).toBeUndefined();
+    await expect(claimPluginTaskSourceRecovery(db, f)).resolves.toEqual(first);
+  });
+
+  it("refreshes only the same source under a newly acquired coordinator epoch", async () => {
+    const f = await fixture(); await claimPluginTaskSourceRecovery(db, f);
+    const owner = { ...f.owner, generation: f.owner.generation + 1, leaseOwner: randomUUID() };
+    await db.update(nativeRunFinalizations).set({ controllerGeneration: owner.generation, leaseOwner: owner.leaseOwner })
+      .where(eq(nativeRunFinalizations.runId, f.runId));
+    const context = new NativeExecutionContext(f.execution, 60_000); context.bindNativeLease(owner); contexts.push(context);
+    const refreshed = await claimPluginTaskSourceRecovery(db, { ...f, context });
+    expect(refreshed).toMatchObject({ identity: f.identity, resumeFrom: f.resumeFrom, controllerGeneration: owner.generation, contextId: context.id });
+    await expect(f.task()).rejects.toThrow("context_owner_changed");
+  });
+
+  it("allows the last A write before consumption and rejects stale A writes afterward", async () => {
+    const f = await fixture();
+    await expect(f.oldWriter(f.checkpoint)).resolves.toBeUndefined();
+    await claimPluginTaskSourceRecovery(db, f);
+    await expect(f.oldWriter(f.checkpoint)).rejects.toThrow("checkpoint_owner_lost");
+    const writer = nativeRunnerCheckpointWriter(db, { companyId: f.companyId, issueId: f.issueId, runId: f.sourceRunId,
+      authorityRunId: f.runId, ...f.owner });
+    await expect(writer(f.checkpoint)).resolves.toBeUndefined();
+    expect((await loadNativeRunnerCheckpoint(db, { companyId: f.companyId, issueId: f.issueId, runId: f.sourceRunId }))?.runId).toBe(f.sourceRunId);
+    await expect(writer({ ...f.checkpoint, records: new Map([["header", "{}"]]) })).rejects.toThrow("checkpoint_owner_lost");
+    await db.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, '{runnerTaskSourceRecovery,controllerGeneration}', '99'::jsonb)` }).where(eq(environmentLeases.id, f.leaseId));
+    await expect(writer(f.checkpoint)).rejects.toThrow("checkpoint_owner_lost");
+  });
+
+  it.each(["writer", "claim"] as const)("serializes a concurrent source writer and claim when %s commits first", async first => {
+    const f = await fixture();
+    let pending: Promise<{ error?: unknown }> | undefined;
+    try {
+      await db.transaction(async tx => {
+        if (first === "writer") {
+          await nativeRunnerCheckpointWriter(tx as unknown as typeof db, { companyId: f.companyId, issueId: f.issueId,
+            runId: f.sourceRunId, authorityRunId: f.sourceRunId, ...f.oldOwner })(f.checkpoint);
+        } else await claimPluginTaskSourceRecovery(tx as unknown as typeof db, f);
+        const [backend] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`);
+        pending = (first === "writer" ? claimPluginTaskSourceRecovery(db, f) : f.oldWriter(f.checkpoint))
+          .then(() => ({}), error => ({ error }));
+        await vi.waitFor(async () => {
+          const [waiting] = await db.execute<{ blocked: boolean }>(sql`select exists (
+            select 1 from pg_stat_activity where datname = current_database()
+              and ${backend.pid} = any(pg_blocking_pids(pid))) as blocked`);
+          expect(waiting.blocked).toBe(true);
+        });
+      });
+    } finally {
+      const result = await pending;
+      if (first === "writer") expect(result).toEqual({});
+      else expect(result?.error).toMatchObject({ message: "native_runner_checkpoint_owner_lost" });
+    }
+    const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.sourceLeaseId));
+    expect(source.metadata?.runnerTaskSourceConsumer).toMatchObject({ leaseId: f.leaseId, runId: f.runId });
+  });
+});

server/src/services/native-runtime/plugin-task-source-recovery.ts

@@ -0,0 +1,198 @@
+import { and, eq, inArray, sql } from "drizzle-orm";
+import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
+import { environmentTaskOperationSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk";
+import { parseNativeExecutionInput, inspectDurablePrpCheckpointHeader, type NativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
+import { resolvePluginTaskResume } from "../plugin-task-resume.js";
+import { legacyShutdownWorkspaceResourceProtected } from "../legacy-shutdown-workspace-settlement.js";
+import { NativeExecutionContext } from "./native-execution-context.js";
+import { isCurrentNativeControllerProcess } from "./native-restart-recovery.js";
+import { nativeWorkspaceExportPendingForResource } from "./native-workspace-finalization-state.js";
+import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js";
+import { loadNativeRunnerCheckpointHeader } from "./native-runner-checkpoint.js";
+import { nativeSha256 as fingerprint } from "./canonical.js";
+
+type Identity = Extract<PluginEnvironmentTaskOperation, { kind: "prepare" }>["requirements"]["identity"];
+type Transaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
+const keys = ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"] as const;
+const record = (value: unknown): Record<string, unknown> => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
+const deny = (): never => { throw new Error("native_source_recovery_ownership_unverified"); };
+
+export type PluginTaskSourceRecovery = {
+  schema: "paperclip.runner.source-recovery.v1";
+  resumeFrom: string;
+  sourceLeaseId: string;
+  identity: Identity;
+  consumerLeaseId: string;
+  consumerRunId: string;
+  controllerBootId: string;
+  controllerGeneration: number;
+  contextId: string;
+  accountIdentity: string;
+  sourceProfileSha256: string;
+  consumerProfileSha256: string;
+};
+
+function readRecovery(value: unknown): PluginTaskSourceRecovery | null {
+  const v = record(value);
+  const identity = environmentTaskOperationSchema.safeParse({ kind: "read_state", identity: v.identity });
+  return v.schema === "paperclip.runner.source-recovery.v1" && identity.success &&
+    ["resumeFrom", "sourceLeaseId", "consumerLeaseId", "consumerRunId", "controllerBootId", "contextId", "accountIdentity"].every(key => typeof v[key] === "string" && v[key] !== "") &&
+    Number.isSafeInteger(v.controllerGeneration) && Number(v.controllerGeneration) >= 0 &&
+    [v.sourceProfileSha256, v.consumerProfileSha256].every(hash => typeof hash === "string" && /^[a-f0-9]{64}$/.test(hash))
+    ? v as unknown as PluginTaskSourceRecovery : null;
+}
+
+function pinnedSource(recovery: PluginTaskSourceRecovery) {
+  const { controllerBootId: _boot, controllerGeneration: _generation, contextId: _context, ...source } = recovery;
+  return source;
+}
+
+async function inspect(tx: Transaction, input: {
+  companyId: string; leaseId: string; context: NativeExecutionContext; identity: Identity; resumeFrom: string; accountIdentity: string;
+}, expected?: PluginTaskSourceRecovery) {
+  input.context.assertOpen();
+  const binding = input.context.execution.binding;
+  if (binding.companyId !== input.companyId || input.identity.runId === binding.runId) return deny();
+  await tx.execute(sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`);
+  const owners = await lockSourceRecoveryCoordinators(tx, input.companyId, [input.identity.runId, binding.runId]);
+  const owner = owners.find(value => value.runId === binding.runId), prior = owners.find(value => value.runId === input.identity.runId);
+  // NOWAIT preserves the established coordinator-first order without waiting
+  // behind a workspace-maintenance path which already holds the issue.
+  const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, binding.issueId))).for("update", { noWait: true });
+  const runs = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId),
+    inArray(heartbeatRuns.id, [input.identity.runId, binding.runId])));
+  const current = runs.find(value => value.id === binding.runId), source = runs.find(value => value.id === input.identity.runId);
+  const [lease] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, input.leaseId))).for("update");
+  const [agent] = await tx.select().from(agents).where(and(eq(agents.companyId, input.companyId), eq(agents.id, binding.agentId)));
+  const [company] = await tx.select().from(companies).where(eq(companies.id, input.companyId));
+  if (!owner || !owner.leaseOwner || !owner.leaseExpiresAt || owner.leaseExpiresAt.getTime() <= Date.now() ||
+      !isCurrentNativeControllerProcess(owner) || owner.resultId || ["committed", "applied", "terminal_failure"].includes(owner.phase) ||
+      !prior || prior.phase !== "committed" || !prior.resultId || prior.leaseOwner ||
+      !current || current.status !== "running" || current.nativeIssueId !== binding.issueId || current.agentId !== binding.agentId || current.resultJson?.nativeCancellation ||
+      !source || source.status !== "succeeded" || !source.finishedAt || source.runtimeMode !== "native" || source.agentId !== binding.agentId || source.nativeIssueId !== binding.issueId ||
+      source.runnerInstanceId !== input.identity.runnerInstanceId || source.nativeSessionId !== input.identity.normalizedSessionId ||
+      !issue || issue.executionRunId !== binding.runId || issue.assigneeAgentId !== binding.agentId || issue.status !== "in_progress" ||
+      company?.status !== "active" || agent?.lifecycleState !== "ready" || ["paused", "terminated", "pending_approval"].includes(agent.status) ||
+      !lease || lease.heartbeatRunId !== binding.runId || lease.issueId !== binding.issueId || lease.status !== "active" || lease.releasedAt ||
+      lease.expiresAt && lease.expiresAt.getTime() <= Date.now() || lease.metadata?.runnerTaskRetiring === true) return deny();
+  input.context.assertNativeLease({ bootId: owner.controllerBootId!, leaseOwner: owner.leaseOwner, generation: owner.controllerGeneration });
+  const header = await loadNativeRunnerCheckpointHeader(tx, { companyId: input.companyId, issueId: binding.issueId, runId: source.id });
+  const protocolIdentity = header ? inspectDurablePrpCheckpointHeader(header) : null;
+  if (!protocolIdentity || keys.some(key => protocolIdentity[key] !== input.identity[key])) return deny();
+  const sourceExecution = parseNativeExecutionInput(source.runnerProfileJson?.nativeExecutionInput);
+  const consumerExecution = parseNativeExecutionInput(current.runnerProfileJson?.nativeExecutionInput);
+  if (sourceExecution.provider.kind !== "codex" || consumerExecution.provider.kind !== "codex" ||
+      sourceExecution.binding.companyId !== input.companyId || sourceExecution.binding.issueId !== binding.issueId ||
+      sourceExecution.binding.agentId !== binding.agentId || sourceExecution.binding.runId !== source.id ||
+      sourceExecution.session.normalizedSessionId !== input.identity.normalizedSessionId ||
+      fingerprint(consumerExecution) !== fingerprint(input.context.execution)) return deny();
+  const sourceTask = await resolvePluginTaskResume(tx as unknown as Db, { companyId: input.companyId, leaseId: input.leaseId,
+    priorIdentity: input.identity, inspectCheckpoint: true });
+  if (sourceTask !== input.resumeFrom) return deny();
+  const sources = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
+    eq(environmentLeases.heartbeatRunId, source.id), eq(environmentLeases.providerLeaseId, sourceTask))).for("update");
+  if (sources.length !== 1) return deny();
+  const sourceLease = sources[0]!;
+  const saved = record(sourceLease.metadata?.runnerTaskIdentity);
+  const aliases = { runnerInstanceId: "runnerId", environmentLeaseId: "leaseId", runId: "runId", normalizedSessionId: "sessionId", turnId: "turnId", itemId: "itemId" } as const;
+  if (keys.some(key => saved[aliases[key]] !== input.identity[key]) ||
+      await nativeWorkspaceExportPendingForResource(tx as unknown as Db, sourceLease, binding.runId) ||
+      await legacyShutdownWorkspaceResourceProtected(tx as unknown as Db, sourceLease)) return deny();
+  const allocations = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.provider, sourceLease.provider!),
+    eq(environmentLeases.providerLeaseId, sourceTask)));
+  if (allocations.some(value => value.companyId !== input.companyId ||
+      value.id !== sourceLease.id && value.id !== lease.id && ["active", "retained", "pending_cleanup"].includes(value.status))) return deny();
+  const recovery: PluginTaskSourceRecovery = { schema: "paperclip.runner.source-recovery.v1", identity: input.identity,
+    resumeFrom: sourceTask, sourceLeaseId: sourceLease.id, consumerLeaseId: lease.id, consumerRunId: binding.runId,
+    controllerBootId: owner.controllerBootId!, controllerGeneration: owner.controllerGeneration, contextId: input.context.id, accountIdentity: input.accountIdentity,
+    sourceProfileSha256: fingerprint(source.runnerProfileJson?.nativeExecutionInput), consumerProfileSha256: fingerprint(current.runnerProfileJson?.nativeExecutionInput) };
+  if (expected && fingerprint(recovery) !== fingerprint(expected)) return deny();
+  input.context.assertOpen();
+  if (owner.leaseExpiresAt.getTime() <= Date.now() || lease.expiresAt && lease.expiresAt.getTime() <= Date.now()) return deny();
+  return { recovery, sourceLease, lease, sourceExecution, consumerExecution, owner };
+}
+
+/** Only current B's native authority can consume succeeded A for maintenance. */
+export async function claimPluginTaskSourceRecovery(db: Db, input: {
+  companyId: string; leaseId: string; context: NativeExecutionContext; identity: Identity; resumeFrom: string; accountIdentity: string;
+  selectedSourceRecoveryVersion?: 1;
+  matchesExecution(source: NativeExecutionInput): boolean;
+}): Promise<PluginTaskSourceRecovery> {
+  if (input.selectedSourceRecoveryVersion !== 1 || !input.accountIdentity) return deny();
+  return db.transaction(async tx => {
+    const inspected = await inspect(tx, input);
+    if (!input.matchesExecution(inspected.sourceExecution)) return deny();
+    const { recovery, sourceLease, lease } = inspected;
+    if (Object.hasOwn(lease.metadata ?? {}, "runnerTaskSourceRecovery")) {
+      const saved = readRecovery(lease.metadata?.runnerTaskSourceRecovery);
+      if (!saved || fingerprint(pinnedSource(saved)) !== fingerprint(pinnedSource(recovery))) return deny();
+    }
+    const consumer = { leaseId: lease.id, runId: recovery.consumerRunId, resumeFrom: recovery.resumeFrom,
+      identity: recovery.identity, accountIdentity: recovery.accountIdentity };
+    const existing = sourceLease.metadata?.runnerTaskSourceConsumer;
+    if (existing && fingerprint(existing) !== fingerprint(consumer)) return deny();
+    const others = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
+      eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.provider, sourceLease.provider!),
+      sql`${environmentLeases.id} <> ${lease.id}`,
+      sql`${environmentLeases.metadata}->>'runnerTaskResumeFrom' = ${input.resumeFrom}`));
+    if (others.length) return deny();
+    input.context.assertOpen();
+    if (inspected.owner.leaseExpiresAt!.getTime() <= Date.now()) return deny();
+    await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ runnerTaskSourceConsumer: consumer })}::jsonb` })
+      .where(eq(environmentLeases.id, sourceLease.id));
+    await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ runnerTaskSourceRecovery: recovery })}::jsonb` })
+      .where(eq(environmentLeases.id, lease.id));
+    return recovery;
+  });
+}
+
+async function assertSourceIdentity(db: Db, input: {
+  companyId: string; leaseId: string; context: NativeExecutionContext | undefined;
+  identity: Identity; accountIdentity?: string; resumeFrom?: string; requireDispatched?: boolean;
+}): Promise<PluginTaskSourceRecovery> {
+  if (!input.context) return deny();
+  const { identity } = input;
+  return db.transaction(async tx => {
+    const [lease] = await tx.select({ metadata: environmentLeases.metadata }).from(environmentLeases).where(and(
+      eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, input.leaseId)));
+    const expected = readRecovery(lease?.metadata?.runnerTaskSourceRecovery);
+    if (!expected || keys.some(key => expected.identity[key] !== identity[key]) ||
+        input.resumeFrom !== undefined && expected.resumeFrom !== input.resumeFrom || expected.accountIdentity !== input.accountIdentity) return deny();
+    const inspected = await inspect(tx, { ...input, context: input.context!, resumeFrom: expected.resumeFrom, accountIdentity: expected.accountIdentity }, expected);
+    if (input.requireDispatched && (inspected.lease.metadata?.runnerTaskDispatched !== true ||
+        inspected.lease.metadata?.runnerTaskExecutionStopped === true)) return deny();
+    const consumer = record(inspected.sourceLease.metadata?.runnerTaskSourceConsumer);
+    if (consumer.leaseId !== input.leaseId || consumer.runId !== expected.consumerRunId || consumer.resumeFrom !== expected.resumeFrom || consumer.accountIdentity !== expected.accountIdentity ||
+        fingerprint(consumer.identity) !== fingerprint(identity)) return deny();
+    return expected;
+  });
+}
+
+/** The worker never derives this exception from an arbitrary mismatched run ID. */
+export async function assertPluginTaskSourceRecovery(db: Db, input: {
+  companyId: string; leaseId: string; context: NativeExecutionContext | undefined;
+  operation: Extract<PluginEnvironmentTaskOperation, { kind: "prepare" | "submit" }>;
+}): Promise<PluginTaskSourceRecovery> {
+  const operation = input.operation;
+  if (!operation.resumeFrom || !operation.accountIdentity ||
+      (operation.kind === "prepare" ? operation.requirements.provider !== "codex" || !!operation.priorIdentity
+        : operation.runner.harness !== "codex" || !operation.preparedLaunchId || !operation.launch)) return deny();
+  const identity = operation.kind === "prepare" ? operation.requirements.identity : {
+    runnerInstanceId: operation.runner.runnerId, environmentLeaseId: operation.runner.leaseId, runId: operation.runner.runId,
+    normalizedSessionId: operation.runner.sessionId, turnId: operation.runner.turnId, itemId: operation.runner.itemId,
+  };
+  return assertSourceIdentity(db, { ...input, identity, resumeFrom: operation.resumeFrom, accountIdentity: operation.accountIdentity });
+}
+
+/** A fresh source incarnation can attach B without pretending it was warm reuse. */
+export async function assertPluginTaskSourceAttachment(db: Db, input: {
+  companyId: string; leaseId: string; context: NativeExecutionContext | undefined;
+  operation: Extract<PluginEnvironmentTaskOperation, { kind: "prepare" }>;
+}): Promise<PluginTaskSourceRecovery> {
+  const { operation } = input, target = operation.requirements.identity;
+  if (!operation.priorIdentity || operation.resumeFrom || operation.requirements.provider !== "codex") return deny();
+  const recovery = await assertSourceIdentity(db, { ...input, identity: operation.priorIdentity, accountIdentity: operation.accountIdentity, requireDispatched: true });
+  if (target.runId !== recovery.consumerRunId || ["runnerInstanceId", "environmentLeaseId", "normalizedSessionId"].some(key =>
+    target[key as keyof Identity] !== recovery.identity[key as keyof Identity])) return deny();
+  return recovery;
+}

server/src/services/plugin-task-resume.ts

@@ -0,0 +1,51 @@
+import { and, eq, or } from "drizzle-orm";
+import { environmentLeases, heartbeatRuns, type Db } from "@paperclipai/db";
+import { hasRemoteTerminationReceipt, remoteExecutionHasStopped } from "./remote-execution-termination.js";
+
+/** A saved provider task can resume only within the same authorized agent workspace. */
+export async function resolvePluginTaskResume(db: Db, input: {
+  companyId: string; leaseId: string;
+  priorIdentity: { runnerInstanceId: string; environmentLeaseId: string; runId: string; normalizedSessionId: string };
+  /** Read-only discovery; launching still requires an attested retained checkpoint. */
+  inspectCheckpoint?: boolean;
+}): Promise<string> {
+  const rows = await db.select({ lease: environmentLeases, run: heartbeatRuns }).from(environmentLeases)
+    .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), eq(heartbeatRuns.companyId, input.companyId)))
+    .where(and(eq(environmentLeases.companyId, input.companyId),
+      or(eq(environmentLeases.id, input.leaseId), eq(heartbeatRuns.id, input.priorIdentity.runId))));
+  const next = rows.find(row => row.lease.id === input.leaseId);
+  // A retained PRP lease is logical; later provider tasks have different physical lease IDs.
+  const candidates = rows.filter(row => {
+    const identity = row.lease.metadata?.runnerTaskIdentity as Record<string, unknown> | undefined;
+    const consumer = row.lease.metadata?.runnerTaskSourceConsumer as { leaseId?: unknown; runId?: unknown; resumeFrom?: unknown } | null | undefined;
+    return row.lease.id !== input.leaseId && row.run.id === input.priorIdentity.runId &&
+      (!Object.hasOwn(row.lease.metadata ?? {}, "runnerTaskSourceConsumer") ||
+        consumer?.leaseId === input.leaseId && consumer.runId === next?.run.id && consumer.resumeFrom === row.lease.providerLeaseId) &&
+      (input.inspectCheckpoint || row.lease.metadata?.runnerTaskSessionRetained === true) &&
+      identity?.runnerId === input.priorIdentity.runnerInstanceId && identity.leaseId === input.priorIdentity.environmentLeaseId &&
+      identity.sessionId === input.priorIdentity.normalizedSessionId && identity.runId === input.priorIdentity.runId &&
+      !rows.some(successor => successor.lease.id !== input.leaseId &&
+        successor.lease.metadata?.runnerTaskResumeFrom === row.lease.providerLeaseId);
+  });
+  const old = candidates.length === 1 ? candidates[0] : undefined;
+  const identity = old?.lease.metadata?.runnerTaskIdentity as Record<string, unknown> | undefined;
+  if (!old || !next || !old.lease.providerLeaseId || !identity ||
+      identity.runnerId !== input.priorIdentity.runnerInstanceId || identity.leaseId !== input.priorIdentity.environmentLeaseId ||
+      identity.sessionId !== input.priorIdentity.normalizedSessionId || identity.runId !== input.priorIdentity.runId ||
+      old.lease.metadata?.driver !== "plugin" || next.lease.metadata?.driver !== "plugin" ||
+      !next.lease.providerLeaseId || next.lease.providerLeaseId === old.lease.providerLeaseId ||
+      (old.run.id !== next.run.id && !["succeeded", "interrupted", "failed", "cancelled", "timed_out"].includes(old.run.status)) ||
+      old.run.agentId !== next.run.agentId || old.lease.issueId !== next.lease.issueId ||
+      old.lease.executionWorkspaceId !== next.lease.executionWorkspaceId ||
+      !old.lease.environmentId || old.lease.environmentId !== next.lease.environmentId ||
+      !next.lease.provider?.startsWith("plugin:") || old.lease.provider !== next.lease.provider ||
+      old.lease.metadata?.pluginId !== next.lease.metadata?.pluginId || old.lease.metadata?.driverKey !== next.lease.metadata?.driverKey ||
+      next.lease.status !== "active" || next.lease.releasedAt || next.run.status !== "running" ||
+      (next.lease.expiresAt && next.lease.expiresAt.getTime() <= Date.now()) ||
+      !(old.run.id === next.run.id
+        ? rows.filter(row => row.lease.id !== next.lease.id).every(row => hasRemoteTerminationReceipt(row.lease))
+        : await remoteExecutionHasStopped(db, input.companyId, old.run.id))) {
+    throw new Error("Plugin Runner retained session is not authorized");
+  }
+  return old.lease.providerLeaseId;
+}