From 31bf6e44828cd3bd1d8b811ffa2f4cddb583ca57 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 5 Nov 2026 12:14:09 -0800 Subject: [PATCH] Pin plugin Runner cleanup to its physical task (#19300) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Plugin Runner owners manage a physical task through an environment lease. > - The existing replacement path can assign a new physical task to that same lease. > - A delayed old owner can then stop the replacement or mark it stopped. > - Cleanup must keep the original physical task identity across RPC and database writes. > - This pull request pins that identity and keeps cleanup available to the new owner. ## Linked Issues or Issue Description Refs #19134 and #19157. This fix is based on master and does not enable repeated stopped-source recovery. A search found no duplicate fix. **What happened?** The maintained `createOwnedSandbox` replacement path changes `providerLeaseId` on one lease. An old task owner uses that mutable lease for a later Stop. A stopped-state response already in flight can also update the new task's flags. **Expected behavior** The old owner must address its original task. Its late cleanup writes must not alter the replacement. The replacement's own owner must still stop and complete it. **Steps to reproduce** 1. Create a plugin task owner for a stopped task. 2. Replace that task through the existing describe/startup path on the same lease. 3. Resume an old checkpoint or a held old status response. 4. Observe a Stop for the replacement or an incorrect stopped flag before this fix. ## What Changed - Require each task owner to keep the physical ID returned by its description. Reject a rebind to a different task. - Carry the ID through the existing internal task context after startup ends. Reject a changed physical task before RPC dispatch. - Match the physical task in stopped-state, completion and credential-handoff writes. - Hold the exact lease row through the retained-session database update and local checkpoint save. - Cover the maintained replacement path with real PostgreSQL tests. Preserve positive cleanup for the replacement owner. Document this boundary. ## Verification - Both replacement regressions failed on unmodified master `10a7d039c8`: delayed old-owner Stop targeted the replacement, and a held old status response marked it stopped. - Six focused files passed 160 tests: task authority, task runtime, warm tasks, execution target, credential writeback and task owner. - Four selected executor factory cases passed. They cover task-only providers and the physical task binding after the startup context ends. - Runner TypeScript, plugin SDK and adapter-utils builds passed. Server TypeScript check passed. Diff and added-line secret scans passed. - All 15 setup-probe cases passed with the actual Runner and fake provider binaries. The first local attempt lacked cargo on PATH. A sandboxed native attempt timed out in seven process cases; the unchanged run with loopback permission passed all 15. A separate initial DB integration attempt was denied a loopback listener; its permission-correct run passed. - Review found three missed constructors in the existing SDK/Core lifecycle fixture. They now use the actual describe result. All four cases in that real PostgreSQL and worker RPC file passed. - The first GitHub attempt failed in the unchanged GitHub recovery pagination case in `chat-channels.integration.test.ts`: its zero-request assertion observed a configuration GET. That exact case passed locally without code or assertion changes. The one failed-job-only retry passed; the initial failure log is retained. - [GitHub workflow 42082587576](https://github.com/paperclipai/paperclip/actions/runs/42082587576) passed on its second attempt. All CI checks completed, and Snyk passed. Standard Greptile scored exact head `fb914d98b99e` 5/5 with no unresolved threads. The branch is mergeable and ready for review. ## Risks The task context change is internal. It adds no plugin wire field, schema or Runner mode. A stale owner now gets an error and cannot clean up a replacement. A new owner can still clean up that task. The retained checkpoint write holds one existing lease row during a bounded local file write. Repeated stopped-source incarnation admission, runtime pins and deployments are outside this change. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning. Used repository tools, TypeScript and real PostgreSQL tests. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used with version and capability details - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have described the issue in-PR using the bug template - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change - [x] I have run the focused tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation - [x] I have considered and documented risks - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no unresolved findings - [x] I will address reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md | 5 ++ .../adapter-utils/src/execution-target.ts | 4 +- .../environment-execution-target.test.ts | 11 +++ ...ronment-task-authority.integration.test.ts | 36 ++++++++-- .../plugin-runner-task.integration.test.ts | 12 ++-- .../plugin-task-credential-writeback.test.ts | 10 ++- .../services/environment-execution-target.ts | 3 +- server/src/services/environment-runtime.ts | 2 +- .../src/services/environment-task-runtime.ts | 14 ++-- .../native-session-executor.test.ts | 27 +++++++ .../native-runtime/native-session-executor.ts | 28 +++++--- .../plugin-runner-probe.test.ts | 4 +- .../native-runtime/plugin-runner-probe.ts | 2 +- .../native-runtime/plugin-runner-task.test.ts | 70 ++++++++++--------- .../native-runtime/plugin-runner-task.ts | 9 ++- .../plugin-task-credential-writeback.ts | 8 ++- 16 files changed, 173 insertions(+), 72 deletions(-) diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index 55e3d57b15c..90591811789 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -157,6 +157,11 @@ An accepted stop does not prove termination. Set `executionStopped: true` only after the Runner and all descendants have stopped. A terminal phase alone does not permit Core to release execution authority. +Core binds each Runner owner to the physical task ID returned by its description. +An old owner cannot follow a changed lease to stop or complete its replacement. +Late stopped-state, retained-session and credential-handoff writes must still +match that same physical task. A new owner can clean up the replacement. + Core fences new submissions when cleanup starts. It stops the saved task, checks termination, saves credential results, and completes the task before it releases or destroys the environment lease. An uncertain result leaves a saved diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 3abae87bf63..08a942542a0 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -185,8 +185,8 @@ export interface AdapterSandboxExecutionTarget extends AdapterExecutionTargetWor kind: "remote"; transport: "sandbox"; providerKey?: string | null; - /** Host-bound plugin task RPC. Both sides validate the wire contract. */ - runnerTask?: (operation: unknown, context?: { executionContext?: unknown }) => Promise; + /** Host-bound plugin task RPC. expectedTaskId pins an owner to its physical task after startup. */ + runnerTask?: (operation: unknown, context?: { executionContext?: unknown; expectedTaskId?: string }) => Promise; runnerTaskProjectIds?: readonly string[]; /** Private Core-issued identity for a bounded, tool-free setup probe. */ runnerTaskProbe?: { runId: string; sessionId: string }; diff --git a/server/src/__tests__/environment-execution-target.test.ts b/server/src/__tests__/environment-execution-target.test.ts index 636a29ed302..fd2edf8107d 100644 --- a/server/src/__tests__/environment-execution-target.test.ts +++ b/server/src/__tests__/environment-execution-target.test.ts @@ -104,6 +104,17 @@ describe("resolveEnvironmentExecutionTarget", () => { delete process.env.PAPERCLIP_RUNTIME_API_URL; }); + it("forwards the physical task binding through the plugin task target", async () => { + const task = vi.fn(async () => ({ kind: "accepted" as const, taskId: "physical-task" })); + const target = await resolveEnvironmentExecutionTarget({ db: {} as never, companyId: "company", adapterType: "paperclip_runner", + environment: { id: "environment", driver: "plugin", config: {} }, leaseMetadata: {}, + lease: { id: "lease", companyId: "company", environmentId: "environment", providerLeaseId: "physical-task", metadata: {} } as never, + environmentRuntime: { supportsTasks: async () => true, task } as never }); + if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runnerTask) throw new Error("Plugin task target unavailable"); + await target.runnerTask({ kind: "stop" }, { expectedTaskId: "physical-task" }); + expect(task).toHaveBeenCalledWith({ companyId: "company", leaseId: "lease", operation: { kind: "stop" }, expectedTaskId: "physical-task" }); + }); + it("uses a bounded default cwd for sandbox targets when lease metadata omits remoteCwd", async () => { mockResolveEnvironmentDriverConfigForRuntime.mockResolvedValue({ driver: "sandbox", diff --git a/server/src/__tests__/environment-task-authority.integration.test.ts b/server/src/__tests__/environment-task-authority.integration.test.ts index 98964501240..8a09b003fe9 100644 --- a/server/src/__tests__/environment-task-authority.integration.test.ts +++ b/server/src/__tests__/environment-task-authority.integration.test.ts @@ -6,6 +6,7 @@ import { resolvePluginTaskResume } from "../services/environment-task-runtime.js import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { environmentRuntimeService } from "../services/environment-runtime.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { createPluginRunnerTaskOwner } from "../services/native-runtime/plugin-runner-task.js"; import { NativeExecutionContext } from "../services/native-runtime/native-execution-context.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -21,7 +22,7 @@ databaseSuite("environment task lease authority", () => { }); afterAll(async () => { await stopDb?.(); }); - it("replaces a stopped warm candidate within its tracked startup context", async () => { + it.each(["dispatch", "writeback"])("fences stale owner %s through maintained physical task replacement", async mode => { const companyId = randomUUID(), agentId = randomUUID(), environmentId = randomUUID(), pluginId = randomUUID(), runId = randomUUID(), leaseId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Warm candidate", issuePrefix: `T${companyId.slice(0, 7).replace(/-/g, "").toUpperCase()}` }); await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "paperclip_runner" }); @@ -33,10 +34,17 @@ databaseSuite("environment task lease authority", () => { await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, heartbeatRunId: runId, provider: "plugin:fixture:tasks", providerLeaseId: "stopped-task", status: "active", leasePolicy: "reuse_by_environment", metadata: { driver: "plugin", pluginId, pluginKey: pluginId, driverKey: "tasks", runnerTaskLease: true, runnerTaskReused: true } }); - const calls: string[] = []; + const calls: string[] = [], stoppedTasks: string[] = []; + let holdStatus = false, releaseStatus!: () => void, enterStatus!: () => void; + const statusEntered = new Promise(resolve => { enterStatus = resolve; }); + const statusReleased = new Promise(resolve => { releaseStatus = resolve; }); const workers = { getWorker: () => ({ supportedMethods: ["environmentTask", "environmentAcquireLease"] }), call: async (_plugin: string, method: string, params: any, _timeout?: number, _sink?: unknown, settled?: () => void) => { calls.push(method === "environmentTask" ? params.operation.kind : method); + if (method === "environmentTask" && params.operation.kind === "stop") stoppedTasks.push(params.taskId); + if (method === "environmentTask" && params.operation.kind === "status" && holdStatus) { + holdStatus = false; enterStatus(); await statusReleased; + } try { if (method === "environmentAcquireLease") return { providerLeaseId: "replacement-task" }; if (params.operation.kind === "status") return { kind: "status", taskId: params.taskId, phase: "completed", executionStopped: true }; @@ -47,16 +55,36 @@ databaseSuite("environment task lease authority", () => { }, } as unknown as PluginWorkerManager; const context = new NativeExecutionContext({ binding: { companyId, agentId, runId } } as never, 1_000); + const runtime = environmentRuntimeService(db, { pluginWorkerManager: workers }); + const task = (operation: unknown, options?: { expectedTaskId?: string }) => runtime.task({ companyId, leaseId, + operation: operation as never, ...options }); + const identity = { runnerInstanceId: "runner", environmentLeaseId: leaseId, runId, normalizedSessionId: "session", turnId: "turn", itemId: "item" }; + const oldOwner = createPluginRunnerTaskOwner({ taskId: "stopped-task", task, projectIds: [], harness: "codex", adoptIdentity: identity }); + const oldRegistration = await oldOwner.registration({} as never, runId); + holdStatus = mode === "writeback"; + const lateStatus = holdStatus ? task({ kind: "status" }, { expectedTaskId: "stopped-task" }) : Promise.resolve(); + if (holdStatus) await statusEntered; try { await expect(environmentRuntimeService(db, { pluginWorkerManager: workers }).task({ companyId, leaseId, operation: { kind: "describe" }, executionContext: context })).resolves.toMatchObject({ kind: "description", taskId: "replacement-task" }); - expect(calls).toEqual(["status", "status", "complete", "environmentAcquireLease", "describe"]); + expect(calls).toEqual([...(mode === "writeback" ? ["status"] : []), "status", "status", "complete", "environmentAcquireLease", "describe"]); expect(context.pending).toBe(false); const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); expect(lease.providerLeaseId).toBe("replacement-task"); expect(lease.metadata?.runnerTaskReused).toBe(false); expect(JSON.stringify(lease.metadata)).not.toContain(context.id); - } finally { context.finish(); } + releaseStatus(); await lateStatus; + const [afterLateStatus] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + expect(afterLateStatus.metadata?.runnerTaskExecutionStopped).toBe(false); + const failedCleanup = await oldRegistration.checkpoint("unsettled").then(() => undefined, error => error); + expect(stoppedTasks).not.toContain("replacement-task"); + expect(failedCleanup).toBeInstanceOf(Error); + const nextOwner = createPluginRunnerTaskOwner({ taskId: "replacement-task", task, projectIds: [], harness: "codex", adoptIdentity: identity }); + const nextRegistration = await nextOwner.registration({} as never, runId); + try { await expect(nextRegistration.checkpoint("unsettled")).resolves.toBeUndefined(); } + finally { await nextRegistration.release("detach"); } + expect(stoppedTasks).toEqual(["replacement-task"]); + } finally { releaseStatus(); await lateStatus.catch(() => undefined); context.finish(); await oldRegistration.release("detach"); } }); it.each(["late", "post-result", "retiring"])("queues a replacement instead of attaching after %s startup cancellation", async mode => { diff --git a/server/src/__tests__/plugin-runner-task.integration.test.ts b/server/src/__tests__/plugin-runner-task.integration.test.ts index c122fe3b370..a9196cab70a 100644 --- a/server/src/__tests__/plugin-runner-task.integration.test.ts +++ b/server/src/__tests__/plugin-runner-task.integration.test.ts @@ -8,7 +8,7 @@ import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environ import { definePlugin } from "../../../packages/plugins/sdk/src/define-plugin.js"; import { startWorkerRpcHost } from "../../../packages/plugins/sdk/src/worker-rpc-host.js"; import { createRequest, parseMessage, serializeMessage, isJsonRpcResponse, isJsonRpcSuccessResponse } from "../../../packages/plugins/sdk/src/protocol.js"; -import { environmentTaskOperationSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; +import { environmentTaskOperationSchema, environmentTaskResultSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import { agentEnvironmentTestService } from "../services/agent-environment-test.js"; import { heartbeatService } from "../services/heartbeat.js"; @@ -191,11 +191,11 @@ databaseSuite("Core plugin Runner task lifecycle", () => { expect(target).toMatchObject({ kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: acquired.lease.id }); if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runnerTask) throw new Error("Missing plugin target"); expect(target.runner).toBeUndefined(); - await expect(target.runnerTask({ kind: "describe" })) - .resolves.toMatchObject({ kind: "description", runner: { version: "fixture" } }); + const description = environmentTaskResultSchema.parse(await target.runnerTask({ kind: "describe" })); + expect(description).toMatchObject({ kind: "description", runner: { version: "fixture" } }); const onDispatched = vi.fn(), checkpoint = vi.fn(async () => {}), close = vi.fn(async () => {}); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close }); - const owner = createPluginRunnerTaskOwner({ task: target.runnerTask, projectAuthCheckVersion: 1, + const owner = createPluginRunnerTaskOwner({ taskId: description.taskId, task: target.runnerTask, projectAuthCheckVersion: 1, credentialIdentity: "revision", accountIdentity: "account", credential: { type: "codex-auth-json", authJson: '{"OPENAI_API_KEY":"synthetic-account-key"}' }, projectIds: target.runnerTaskProjectIds ?? [], harness: "codex", onSessionRetained: checkpoint, onDispatched, pollMs: 1 }); const identity = { runnerInstanceId: randomUUID(), environmentLeaseId: acquired.lease.id, runId, @@ -244,7 +244,7 @@ databaseSuite("Core plugin Runner task lifecycle", () => { reattachRemoteLease: { kind: "plugin_task", leaseId: saved.id }, }); expect(recovered.lease.id).toBe(saved.id); - const adopted = createPluginRunnerTaskOwner({ task: target.runnerTask, projectIds: [projectId], harness: "codex", + const adopted = createPluginRunnerTaskOwner({ taskId: description.taskId, task: target.runnerTask, projectIds: [projectId], harness: "codex", projectAuthCheckVersion: 1, adoptIdentity: identity }); await adopted.driver.authorizeRunnerAdoption!(prepared.configurationId, identity); const adoptedRegistration = await adopted.registration({} as never, runId); @@ -254,7 +254,7 @@ databaseSuite("Core plugin Runner task lifecycle", () => { expect(calls.slice(callsBefore).some(op => ["prepare", "submit", "stop", "complete"].includes(op.kind))).toBe(false); expect(executionStopped).toBe(false); projectAuthAllowed = false; - const revoked = createPluginRunnerTaskOwner({ task: target.runnerTask, projectIds: [projectId], harness: "codex", + const revoked = createPluginRunnerTaskOwner({ taskId: description.taskId, task: target.runnerTask, projectIds: [projectId], harness: "codex", projectAuthCheckVersion: 1, adoptIdentity: identity }); const connectionsBeforeRevocation = calls.filter(op => op.kind === "connection").length; await expect(revoked.driver.authorizeRunnerAdoption!(prepared.configurationId, identity)).rejects.toThrow("project authentication conflicts"); diff --git a/server/src/__tests__/plugin-task-credential-writeback.test.ts b/server/src/__tests__/plugin-task-credential-writeback.test.ts index 0603d09f59a..203e2b35e83 100644 --- a/server/src/__tests__/plugin-task-credential-writeback.test.ts +++ b/server/src/__tests__/plugin-task-credential-writeback.test.ts @@ -8,7 +8,7 @@ import { agents, companies, companyMemberships, createDb, environmentLeases, env import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { aiConnectionService } from "../services/ai-connections.js"; import { prepareManagedAiRuntime } from "../services/ai-connection-runtime.js"; -import { registerPluginTaskCredentialHandoff } from "../services/plugin-task-credential-writeback.js"; +import { registerPluginTaskCredentialHandoff, settlePluginTaskCredentialHandoff } from "../services/plugin-task-credential-writeback.js"; import { executeEnvironmentTask, settleDispatchedEnvironmentTask } from "../services/environment-task-runtime.js"; import { secretService } from "../services/secrets.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; @@ -123,6 +123,14 @@ suite("durable plugin task credential handoff", () => { expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete" }); }); + it("does not mark a replacement task's identical credential handoff complete", async () => { + const f = await fixture(false), prior = await f.lease(); + await db.update(environmentLeases).set({ providerLeaseId: "replacement-task" }).where(eq(environmentLeases.id, f.leaseId)); + await expect(settlePluginTaskCredentialHandoff(db, { companyId: f.companyId, leaseId: f.leaseId, expectedTaskId: "fixture-task" }, + prior.metadata!, async () => { throw new Error("Undispatched task must not export credentials"); })).rejects.toThrow("handoff changed"); + expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" }); + }); + it("retires an undispatched task without exporting credentials and blocks a late submit", async () => { const f = await fixture(false); expect(await f.settle()).toBe(false); diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index e72af716e18..a4d3a8c16da 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -277,7 +277,8 @@ export async function resolveEnvironmentExecutionTarget(input: { runnerTask: async (operation, context) => { const result = await runtime.task({ companyId: input.companyId, leaseId: lease.id, operation: environmentTaskOperationSchema.parse(operation), - ...(context?.executionContext ? { executionContext: context.executionContext } : {}) }); + ...(context?.executionContext ? { executionContext: context.executionContext } : {}), + ...(context?.expectedTaskId !== undefined ? { expectedTaskId: context.expectedTaskId } : {}) }); if (result.kind === "description" && result.taskId !== acquisition.providerLeaseId) { acquisition.outcome = "replacement"; acquisition.providerLeaseId = result.taskId; diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index b4fffc7ada9..50a779cef76 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -3787,7 +3787,7 @@ export function environmentRuntimeService( }, /** Dispatch plugin-provided Runner execution, including resource preparation and task lifecycle. */ - async task(input: { companyId: string; leaseId: string; operation: PluginEnvironmentTaskOperation; executionContext?: unknown }) { + async task(input: { companyId: string; leaseId: string; operation: PluginEnvironmentTaskOperation; executionContext?: unknown; expectedTaskId?: string }) { if (!options.pluginWorkerManager) throw new Error("Plugin-provided Runner execution worker manager unavailable"); return executeEnvironmentTask(db, options.pluginWorkerManager, input); }, diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index f7b89c69628..1bc5ed54b1e 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -35,6 +35,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage leaseId: string; operation: PluginEnvironmentTaskOperation; executionContext?: unknown; + /** Physical task captured by an existing owner, independent of the mutable lease. */ + expectedTaskId?: string; }) { const operation = environmentTaskOperationSchema.parse(input.operation); const [row] = await db.select({ lease: environmentLeases, environment: environments, run: heartbeatRuns }) @@ -52,6 +54,7 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage context.assertScope({ companyId: input.companyId, agentId: run.agentId, runId: run.id }); } const taskId = row.lease.providerLeaseId; + if (input.expectedTaskId !== undefined && input.expectedTaskId !== taskId) throw new Error("Plugin Runner physical task changed"); const metadata = lease.metadata ?? {}; const probe = metadata.runnerTaskProbe as { runId?: unknown; sessionId?: unknown; expiresAtMs?: unknown } | undefined; const activeProbe = lease.heartbeatRunId === null && lease.issueId === null && probe @@ -338,8 +341,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage } } if (operation.kind === "complete") { - await settlePluginTaskCredentialHandoff(db, { companyId: input.companyId, leaseId: lease.id }, metadata, async accountIdentity => { - const snapshot = await executeEnvironmentTask(db, workers, { ...input, operation: { kind: "credentials", accountIdentity } }); + await settlePluginTaskCredentialHandoff(db, { companyId: input.companyId, leaseId: lease.id, expectedTaskId: taskId }, metadata, async accountIdentity => { + const snapshot = await executeEnvironmentTask(db, workers, { ...input, expectedTaskId: taskId, operation: { kind: "credentials", accountIdentity } }); if (snapshot.kind !== "credentials") throw new Error("Plugin Runner credential snapshot unavailable"); return snapshot.credential; }); @@ -405,7 +408,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ runnerTaskExecutionStopped: true, runnerTaskSessionRetained: parsed.kind === "status" && parsed.sessionStateAvailable === true, ...(parsed.kind === "status" && parsed.preparationStopped === true ? { runnerTaskPreparationStopped: true } : {}) })}::jsonb`, - }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId))); + }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId), + eq(environmentLeases.providerLeaseId, taskId))); } if (operation.kind === "complete") { if (freshWorkspace && metadata.runnerTaskExecutionStopped === true) @@ -420,13 +424,13 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ ...(owner.status === "retained" ? {status:"released" as const,releasedAt:new Date(),cleanupStatus:"success" as const}:{}), metadata:sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({remoteExecutionTermination:receipt,runnerTaskExecutionStopped:true})}::jsonb`, - }).where(eq(environmentLeases.id,owner.id)), { companyId: owner.companyId, runId: owner.heartbeatRunId ?? undefined }); + }).where(and(eq(environmentLeases.id,owner.id),eq(environmentLeases.providerLeaseId,taskId))), { companyId: owner.companyId, runId: owner.heartbeatRunId ?? undefined }); } } await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskDispatched":false}'::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId), - sql`${environmentLeases.metadata}->>'runnerTaskExecutionStopped' = 'true'`)); + eq(environmentLeases.providerLeaseId, taskId), sql`${environmentLeases.metadata}->>'runnerTaskExecutionStopped' = 'true'`)); } return parsed; } catch { diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 8090b3cab6f..15021181cfa 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -13430,6 +13430,33 @@ describe("runnerd provider runtime wiring", () => { return state.createTransport.mock.calls[0]![0].stateDirectory!; } + it("keeps physical task cleanup binding after the startup context finishes", async () => { + const current = scopedNativeExecution("physical-task-cleanup"); + const context = new NativeExecutionContext(current, 1_000); + const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner"); + const db = leaseDb(current), update = db.update.bind(db); + db.update = ((table: unknown) => table === environmentLeases + ? { set: () => ({ where: () => ({ returning: async () => [] }) }) } + : update(table as never)) as typeof db.update; + db.transaction = (async callback => callback(db as never)) as typeof db.transaction; + const task = vi.fn(async (_operation: unknown, _options?: { executionContext?: unknown; expectedTaskId?: string }) => ({ + kind: "description", taskId: "physical-task", launchPreparationVersion: 1, projectAuthCheckVersion: 1, + runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` }, + })); + try { + await createRunnerdBackend({ db, execution: current, runnerInstanceId: "physical-task-runner", + executionContext: context, runnerIngressAuthorized: true, runnerExecutionTarget: { + kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: "lease", runnerTask: task, + } }); + const input = owner.mock.calls.at(-1)![0]; + expect(input.taskId).toBe("physical-task"); + context.finish(); + await input.task({ kind: "stop" }, { expectedTaskId: "physical-task" }); + expect(task).toHaveBeenLastCalledWith({ kind: "stop" }, { expectedTaskId: "physical-task" }); + await expect(input.onSessionRetained!({})).rejects.toThrow("physical task changed"); + } finally { context.finish(); owner.mockRestore(); } + }); + it("rejects a late backend factory after its startup context has finished", async () => { const current = scopedNativeExecution("late-context-factory"); state.createBackend.mockClear(); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 5d183df60ae..d0de738f785 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -13257,10 +13257,11 @@ async function createRunnerdBackendWithinSessionClaim( const cloudArtifactRoot = input.execution.provider.kind === "cursor_cloud" ? join(runnerdStateRoot(input.execution), "cursor-cloud", "artifacts") : undefined; if (input.execution.provider.kind === "cursor_cloud" && remoteTarget) throw new Error("cursor_cloud_local_controller_required"); const rawPluginTask = remoteTarget?.transport === "sandbox" ? remoteTarget.runnerTask : undefined; - const pluginTask = rawPluginTask && input.executionContext ? (operation: unknown) => { + const pluginTask = rawPluginTask && input.executionContext ? (operation: unknown, options?: { expectedTaskId?: string }) => { const startupOperation = ["describe", "prepare", "check_project_auth", "connection"].includes(String(record(operation).kind)); if (startupOperation) input.executionContext?.signal.throwIfAborted(); - return rawPluginTask(operation, input.executionContext?.active && startupOperation ? { executionContext: input.executionContext } : undefined); + return rawPluginTask(operation, { ...options, + ...(input.executionContext?.active && startupOperation ? { executionContext: input.executionContext } : {}) }); } : rawPluginTask; let runnerControlPlane: DurablePrpControlPlane | undefined; if (pluginTask && input.runnerIngressAuthorized !== true) throw new Error("runner_ingress_unavailable"); @@ -13493,7 +13494,7 @@ async function createRunnerdBackendWithinSessionClaim( await registerPluginTaskCredentialHandoff(input.db, { companyId: input.execution.binding.companyId, leaseId: remoteTarget.leaseId }, credentials.accountIdentity, input.managedAiCredentialWriteback); } - const ownerInput: Parameters[0] = { ...credentials, + const ownerInput: Parameters[0] = { ...credentials, taskId: description.taskId, executionContext: input.executionContext?.pluginSupported ? input.executionContext : undefined, // Warm retirement can outlive the run's temporary credential home; complete uses the durable handoff. onCredentials: input.execution.session.lifecyclePolicy.mode === "warm" ? undefined : credentials.onCredentials, @@ -13511,14 +13512,19 @@ async function createRunnerdBackendWithinSessionClaim( } : undefined, credentialIdentity: input.managedAiCredentialIdentity, onSessionRetained: async state => { - savePluginRunnerCheckpoint(root, state, state); - await input.db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ - runnerTaskSessionRetained: true, runnerTaskIdentity: { - runnerId: state.runnerInstanceId, leaseId: state.environmentLeaseId, - runId: state.runId, sessionId: state.normalizedSessionId, turnId: state.turnId, itemId: state.itemId, - }, - })}::jsonb` }) - .where(and(eq(environmentLeases.id, remoteTarget.leaseId!), eq(environmentLeases.companyId, input.execution.binding.companyId))); + await input.db.transaction(async tx => { + const saved = await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ + runnerTaskSessionRetained: true, runnerTaskIdentity: { + runnerId: state.runnerInstanceId, leaseId: state.environmentLeaseId, + runId: state.runId, sessionId: state.normalizedSessionId, turnId: state.turnId, itemId: state.itemId, + }, + })}::jsonb` }) + .where(and(eq(environmentLeases.id, remoteTarget.leaseId!), eq(environmentLeases.companyId, input.execution.binding.companyId), + eq(environmentLeases.providerLeaseId, description.taskId))).returning({ id: environmentLeases.id }); + if (saved.length !== 1) throw new Error("Plugin Runner physical task changed"); + // Keep the lease row locked until its local checkpoint has been saved. + savePluginRunnerCheckpoint(root, state, state); + }); }, onDispatched: input.onTaskDispatched, }; if (input.execution.session.lifecyclePolicy.mode === "warm" && (!credentials.accountIdentity || description.warmTaskVersion !== 1)) throw new Error("Plugin warm task capability or account unavailable"); diff --git a/server/src/services/native-runtime/plugin-runner-probe.test.ts b/server/src/services/native-runtime/plugin-runner-probe.test.ts index fdfeb6e69b9..89551fa83bf 100644 --- a/server/src/services/native-runtime/plugin-runner-probe.test.ts +++ b/server/src/services/native-runtime/plugin-runner-probe.test.ts @@ -229,7 +229,7 @@ describe("plugin Runner setup", () => { await mkdir(resources.provider_state!, { recursive: true }); await writeFile(join(resources.provider_state!, "AGENTS.md"), "Installed persistent instructions"); const retained = vi.fn(async () => {}); - const owner = createPluginRunnerTaskOwner({ task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", onSessionRetained: retained, pollMs: 1 }); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", onSessionRetained: retained, pollMs: 1 }); const first = createRunnerdCodexTransport({ ...owner.driver, stateDirectory, runnerArtifact: artifact, runtimeContext: context("first"), prpIdentity: identity, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }); let thread: Record; @@ -245,7 +245,7 @@ describe("plugin Runner setup", () => { for (const run of ["same", "next"]) { if (run !== "same") await writeFile(join(resources.provider_state!, "AGENTS.md"), `Installed persistent instructions ${run}`); const priorRun = "run-first"; - const nextOwner = createPluginRunnerTaskOwner({ task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", resumeFrom: "task", pollMs: 1, + const nextOwner = createPluginRunnerTaskOwner({ taskId: "task", task, projectAuthCheckVersion: 1, projectIds: [], harness: "codex", resumeFrom: "task", pollMs: 1, priorIdentity: { ...identity, runId: priorRun, turnId: identity.turnId, itemId: identity.itemId } }); const second = createRunnerdCodexTransport({ ...nextOwner.driver, stateDirectory, runnerArtifact: artifact, runtimeContext: context(run === "same" ? "first" : run), diff --git a/server/src/services/native-runtime/plugin-runner-probe.ts b/server/src/services/native-runtime/plugin-runner-probe.ts index 41a776bf1f4..4bf9f3cc022 100644 --- a/server/src/services/native-runtime/plugin-runner-probe.ts +++ b/server/src/services/native-runtime/plugin-runner-probe.ts @@ -43,7 +43,7 @@ export async function probePluginRunnerEnvironment(context: AdapterEnvironmentTe identity: managed?.identity, accountIdentity: managed?.accountIdentity, home: managed?.home, environment, }); - const owner = createPluginRunnerTaskOwner({ ...credentials, projectAuthCheckVersion: description.projectAuthCheckVersion, task, projectIds: [], + const owner = createPluginRunnerTaskOwner({ ...credentials, taskId: description.taskId, projectAuthCheckVersion: description.projectAuthCheckVersion, task, projectIds: [], harness: profile.provider === "acpx" ? profile.acpxAgent! : profile.provider, credentialIdentity: managed?.identity }); probeStarted = true; diff --git a/server/src/services/native-runtime/plugin-runner-task.test.ts b/server/src/services/native-runtime/plugin-runner-task.test.ts index b1525d74cca..2ce25e84837 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -54,12 +54,17 @@ function provider() { } describe("plugin native Runner ownership", () => { + it("requires an explicit physical task binding", () => { + const { task } = provider(); + expect(() => createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" } as never)).toThrow("physical task unavailable"); + }); + it.each(["finish", "rebind"])("does not send startup Stop after the context %s", mode => { const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); const next = new NativeExecutionContext(context.execution, 1_000); const remove = vi.spyOn(context.signal, "removeEventListener"); const { task } = provider(); - const input = { task, projectIds: [], harness: "codex", executionContext: context, adoptIdentity: requirements.identity }; + const input = { taskId: "task", task, projectIds: [], harness: "codex", executionContext: context, adoptIdentity: requirements.identity }; const owner = createPluginRunnerTaskOwner(input); try { if (mode === "finish") context.finish(); @@ -74,7 +79,7 @@ describe("plugin native Runner ownership", () => { let allowed = true; task.mockImplementation(async op => op.kind === "check_project_auth" ? { ...await perform(op) as object, allowed } : perform(op)); - const input = { task, projectIds: ["project-retained"], harness: "codex", projectAuthCheckVersion: 1 as const, + const input = { taskId: "task", task, projectIds: ["project-retained"], harness: "codex", projectAuthCheckVersion: 1 as const, adoptIdentity: requirements.identity }; const owner = createPluginRunnerTaskOwner(input); await owner.driver.authorizeRunnerAdoption!("retained-configuration", requirements.identity); @@ -90,7 +95,7 @@ describe("plugin native Runner ownership", () => { let finish!: () => void; const gate = new Promise(resolve => { finish = resolve; }); task.mockImplementation(async op => { if (mode === "late-release" && op.kind === "check_project_auth") await gate; return perform(op); }); - const owner = createPluginRunnerTaskOwner({ task, projectIds: ["project-current"], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: ["project-current"], harness: "codex", projectAuthCheckVersion: mode === "unsupported" ? undefined : 1, adoptIdentity: requirements.identity }); const registration = await owner.registration({} as never, "run"); if (mode === "released") await registration.release("detach"); @@ -107,7 +112,7 @@ describe("plugin native Runner ownership", () => { const { task } = provider(); const close = vi.fn(async () => {}); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close }); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", adoptIdentity: requirements.identity }); const registration = await owner.registration({} as never, "run"); registration.activate(); await registration.ready(); @@ -131,7 +136,7 @@ describe("plugin native Runner ownership", () => { return result; }); ingress.connect.mockClear(); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", adoptIdentity: requirements.identity }); const registration = await owner.registration({} as never, "run", abort.signal); registration.activate(); @@ -144,7 +149,7 @@ describe("plugin native Runner ownership", () => { const { task } = provider(); const retained = vi.fn(async () => {}); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} }); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", adoptIdentity: requirements.identity, onSessionRetained: retained }); const registration = await owner.registration({} as never, "run"); registration.activate(); await registration.ready(); @@ -155,7 +160,7 @@ describe("plugin native Runner ownership", () => { it.each(["settled", "unsettled"] as const)("proves retirement independently of %s checkpoint reuse", async settlement => { const { task } = provider(); const retained = vi.fn(async () => {}); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", adoptIdentity: requirements.identity, onSessionRetained: retained }); const registration = await owner.registration({} as never, "run"); expect(owner.isRetired()).toBe(false); @@ -176,7 +181,7 @@ describe("plugin native Runner ownership", () => { if (operation.kind === "credentials") return { kind: "credentials", taskId: "task", credential: null }; return perform(operation); }); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", accountIdentity: "account", + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", accountIdentity: "account", adoptIdentity: requirements.identity, onCredentials: async () => {}, onSessionRetained: async () => {} }); const registration = await owner.registration({} as never, "run"); await registration.release(); @@ -187,7 +192,7 @@ describe("plugin native Runner ownership", () => { it("submits only a prepared identity and configuration ID", async () => { const { task } = provider(); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); expect(() => owner.launcher(launchSpec())).toThrow("requires preparation"); await prepare(owner); expect(task.mock.calls[0][0]).toEqual({ kind: "prepare", requirements, projectIds: [] }); @@ -206,7 +211,7 @@ describe("plugin native Runner ownership", () => { const perform = task.getMockImplementation()!; task.mockImplementation(async op => op.kind === "prepare" ? { ...await perform(op) as object, projectAuthCheckAtSubmit: true } : perform(op)); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); await prepare(owner); expect(task.mock.calls.map(([op]) => op.kind)).toEqual(["prepare"]); @@ -216,7 +221,7 @@ describe("plugin native Runner ownership", () => { registration.activate(); await registration.ready(); await registration.release(); await registration.checkpoint("unsettled"); await handle.completion; expect(task.mock.calls.filter(([op]) => op.kind === "submit")).toHaveLength(1); - const unsupported = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" }); + const unsupported = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex" }); await expect(prepare(unsupported)).rejects.toThrow("validation unavailable"); }); @@ -225,7 +230,7 @@ describe("plugin native Runner ownership", () => { const gate = new Promise(resolve => { finish = resolve; }); const { task } = provider(); const perform = task.getMockImplementation()!; task.mockImplementation(async op => { if (op.kind === "prepare") await gate; return perform(op); }); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex" }); + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex" }); const prepared = prepare(owner); const registration = await owner.registration({} as never, "run"); await registration.release(); finish(); @@ -238,11 +243,11 @@ describe("plugin native Runner ownership", () => { let release!: () => void; const gate = new Promise(resolve => { release = resolve; }); task.mockImplementation(async operation => { if (operation.kind === "prepare") await gate; return perform(operation); }); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, executionContext: context }); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, executionContext: context }); const preparing = prepare(owner); - await vi.waitFor(() => expect(task).toHaveBeenCalledWith(expect.objectContaining({ kind: "prepare" }))); + await vi.waitFor(() => expect(task).toHaveBeenCalledWith(expect.objectContaining({ kind: "prepare" }), { expectedTaskId: "task" })); context.abort(new Error("operator Stop")); - expect(task).toHaveBeenCalledWith({ kind: "stop" }); + expect(task).toHaveBeenCalledWith({ kind: "stop" }, { expectedTaskId: "task" }); release(); await expect(preparing).rejects.toThrow("preparation cancelled"); expect(task.mock.calls.some(([operation]) => ["check_project_auth", "connection", "submit"].includes(operation.kind))).toBe(false); @@ -254,7 +259,7 @@ describe("plugin native Runner ownership", () => { ? { kind: "project_auth", taskId: "task", allowed: mode !== "denied", configurationId: mode === "wrong-configuration" ? "other" : op.configurationId, provider: mode === "wrong-provider" ? "anthropic" : op.provider } : perform(op)); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: mode === "unsupported" ? undefined : 1, credentialIdentity: "revision", accountIdentity: "account", credential: { type: "codex-auth-json", authJson: '{"OPENAI_API_KEY":"never-transfer"}' } }); await expect(prepare(owner)).rejects.toThrow(mode === "unsupported" ? "validation unavailable" : "conflicts"); @@ -264,7 +269,7 @@ describe("plugin native Runner ownership", () => { }); it("rejects an installed account when the plugin lacks project validation", async () => { const { task } = provider(); - const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" }); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex" }); await expect(prepare(owner)).rejects.toThrow("validation unavailable"); expect(task.mock.calls.some(([op]) => op.kind === "submit")).toBe(false); }); @@ -282,7 +287,7 @@ describe("plugin native Runner ownership", () => { return perform(operation); }); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} }); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", credential, + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", credential, credentialIdentity: "revision", accountIdentity: "stable-account", pollMs: 1, onCredentials: async value => { expect(value).toEqual(credential); copied = true; } }); await prepare(owner); @@ -297,7 +302,7 @@ describe("plugin native Runner ownership", () => { const perform = task.getMockImplementation()!; task.mockImplementation(async op => op.kind === "credentials" ? { kind: "credentials", taskId: "task", credential: { type: "codex-auth-json", authJson: "{}" } } : perform(op)); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} }); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", accountIdentity: "account", pollMs: 1, + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", accountIdentity: "account", pollMs: 1, onCredentials: async () => { throw new Error("credential handoff failed"); } }); await prepare(owner); const registration = await owner.registration({} as never, "run"); @@ -324,7 +329,7 @@ describe("plugin native Runner ownership", () => { }); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} }); const checkpoint = vi.fn(async () => {}); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: checkpoint, pollMs: 1, + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: checkpoint, pollMs: 1, }); await prepare(owner); const registration = await owner.registration({} as never, "run"); @@ -349,7 +354,7 @@ describe("plugin native Runner ownership", () => { let fail!: (error: Error) => void; const failure = new Promise((_resolve, reject) => { fail = reject; }); ingress.connect.mockReturnValue({ ready: Promise.resolve(), failure, close: async () => {} }); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); await prepare(owner); const registration = await owner.registration({} as never, "run"); const observed = registration.failure.catch(error => error); @@ -370,7 +375,7 @@ describe("plugin native Runner ownership", () => { return perform(op); }); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} }); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); await prepare(owner); const registration = await owner.registration({} as never, "run"); @@ -388,7 +393,7 @@ describe("plugin native Runner ownership", () => { it("retains task grants when retention recording fails after route release", async () => { const { task } = provider(); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close: async () => {} }); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: async () => { throw new Error("retention write failed"); }, pollMs: 1 }); + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onSessionRetained: async () => { throw new Error("retention write failed"); }, pollMs: 1 }); await prepare(owner); const registration = await owner.registration({} as never, "run"); const process = owner.launcher(launchSpec()); @@ -403,7 +408,7 @@ describe("plugin native Runner ownership", () => { const { task } = provider(); const close = vi.fn(async () => {}); ingress.connect.mockReturnValue({ ready: Promise.resolve(), close }); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); await prepare(owner); const registration = await owner.registration({} as never, "run"); @@ -417,7 +422,7 @@ describe("plugin native Runner ownership", () => { let finish!: () => void; const prepared = new Promise(resolve => { finish = resolve; }); const { task } = provider(); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", onDispatched: () => prepared, pollMs: 1 }); await prepare(owner); const registration = await owner.registration({} as never, "run"); @@ -433,7 +438,7 @@ describe("plugin native Runner ownership", () => { const { task } = provider(); ingress.connect.mockClear(); const controller = new AbortController(); controller.abort(); - const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task", projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); await prepare(owner); const registration = await owner.registration({} as never, "run", controller.signal); @@ -456,7 +461,7 @@ describe("plugin task workspace file transfer", () => { let file={kind:"file",taskId:"task",base64:bytes.toString("base64"),byteSize:bytes.length,sha256:createHash("sha256").update(bytes).digest("hex")}; let afterRead=async()=>{}; task.mockImplementation(async op=>{if(op.kind==="read_file"){await afterRead();return file;}return perform(op);}); - const owner=createPluginRunnerTaskOwner({projectAuthCheckVersion:1,task,projectIds:[],harness:"codex",pollMs:1}); + const owner=createPluginRunnerTaskOwner({ taskId: "task",projectAuthCheckVersion:1,task,projectIds:[],harness:"codex",pollMs:1}); await expect(owner.readWorkspaceFile("proof.txt",1024)).rejects.toThrow("unavailable"); await prepare(owner); ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}}); @@ -480,7 +485,7 @@ it.each(["codex", "claude"])("reuses a warm %s process across run authorities wi const {task} = provider(); ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}}); const retained = vi.fn(async()=>{}); - const input = {task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const, + const input = {taskId:"task",task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const, accountIdentity:"account",credentialIdentity:"revision",pollMs:1,onSessionRetained:retained, githubBroker:{url:"https://core.example.test",token:"first-run-capability"}}; const owner = createPluginRunnerTaskOwner(input); await prepare(owner, launchRequirements(harness)); @@ -512,7 +517,7 @@ it.each(["codex", "claude"])("reuses a warm %s process across run authorities wi it.each(["codex", "claude"])("adopts a verified retained %s task without submitting a replacement", async harness => { const {task}=provider(); ingress.connect.mockReturnValue({ready:Promise.resolve(),close:async()=>{}}); - const owner=createPluginRunnerTaskOwner({task,projectIds:[],harness,projectAuthCheckVersion:1, + const owner=createPluginRunnerTaskOwner({ taskId: "task",task,projectIds:[],harness,projectAuthCheckVersion:1, warmTaskVersion:1,accountIdentity:"account",adoptIdentity:requirements.identity,pollMs:1}); const recovered = {configurationId:"saved",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}}; await expect(owner.driver.prepareLaunch!(launchRequirements(harness),recovered)).resolves.toEqual(recovered); @@ -531,7 +536,7 @@ it.each(["codex", "claude"])("denies retained %s process adoption when current p task.mockImplementation(async op=>op.kind === "check_project_auth" ? {kind:"project_auth",taskId:"task",configurationId:op.configurationId,provider:op.provider,allowed:false} as never : {kind:"accepted",taskId:"task"} as never); - const owner=createPluginRunnerTaskOwner({task,projectIds:["project"],harness,projectAuthCheckVersion:1, + const owner=createPluginRunnerTaskOwner({ taskId: "task",task,projectIds:["project"],harness,projectAuthCheckVersion:1, warmTaskVersion:1,accountIdentity:"account",adoptIdentity:requirements.identity}); await expect(owner.driver.prepareLaunch!(launchRequirements(harness),{configurationId:"saved",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}})) .rejects.toThrow("project authentication conflicts"); @@ -540,12 +545,13 @@ it.each(["codex", "claude"])("denies retained %s process adoption when current p it.each(["codex", "claude"])("refuses warm %s account, harness, project, or credential generation changes", async harness=>{ const {task}=provider(); - const input={task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const, + const input={taskId:"task",task,projectIds:[],harness,projectAuthCheckVersion:1 as const,warmTaskVersion:1 as const, accountIdentity:"account",credentialIdentity:"revision",pollMs:1}; const owner=createPluginRunnerTaskOwner(input); await prepare(owner, launchRequirements(harness)); const spec=launchSpec();spec.nativeRunner!.lifecyclePolicy={mode:"warm",idleTimeoutMs:300000}; const process=owner.launcher(spec); + expect(()=>owner.rebind({...input,taskId:"replacement"})).toThrow("cannot be rebound"); expect(()=>owner.rebind({...input,accountIdentity:"other"})).toThrow("cannot be rebound"); expect(()=>owner.rebind({...input,credentialIdentity:"rotated"})).toThrow("cannot be rebound"); expect(()=>owner.rebind({...input,harness:harness === "codex" ? "claude" : "codex"})).toThrow("cannot be rebound"); @@ -560,7 +566,7 @@ it.each(["codex", "claude"])("refuses warm %s account, harness, project, or cred task.mockImplementation(async operation => operation.kind === "describe" ? {kind:"description",taskId:"task",runner:{version:"fixture",digest:`sha256:${"a".repeat(64)}`}} : implementation(operation)); - const owner = createPluginRunnerTaskOwner({task,projectIds:[],harness:"codex", + const owner = createPluginRunnerTaskOwner({ taskId: "task",task,projectIds:[],harness:"codex", githubBroker:{url:"https://core.example.test",token:"runtime-capability"}}); await expect(prepare(owner)).rejects.toThrow("credential broker v1"); expect(task.mock.calls.map(([op])=>op.kind)).toEqual(["describe"]); diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index 92e0de33507..3857dcfa431 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -29,7 +29,8 @@ export async function verifyPluginRunnerTaskReattachment( */ export function createPluginRunnerTaskOwner(input: { executionContext?: NativeExecutionContext; - task: (operation: unknown) => Promise; + taskId: string; + task: (operation: unknown, options?: { expectedTaskId?: string }) => Promise; projectIds: readonly string[]; harness: string; projectAuthCheckVersion?: 1; @@ -46,6 +47,8 @@ export function createPluginRunnerTaskOwner(input: { githubBroker?: { url: string; token: string }; adoptIdentity?: DurableRecoveryIdentity; }) { + const taskId = input.taskId; + if (!taskId) throw new Error("Plugin Runner physical task unavailable"); const pollMs = input.pollMs ?? 500; let registrationGeneration = 0; let activeOutbound: ReturnType | null = null; @@ -72,7 +75,7 @@ export function createPluginRunnerTaskOwner(input: { const phase = ({ prepare: "execution.plugin_prepare", check_project_auth: "execution.plugin_project_auth", submit: "execution.plugin_submit", connection: "execution.plugin_connection" } as const)[operation.kind as "prepare" | "check_project_auth" | "submit" | "connection"]; const result = environmentTaskResultSchema.parse(await (phase - ? measureRunPhase(phase, () => input.task(operation)) : input.task(operation))); + ? measureRunPhase(phase, () => input.task(operation, { expectedTaskId: taskId })) : input.task(operation, { expectedTaskId: taskId }))); if (result.kind !== (operation.kind === "read_file" ? "file" : operation.kind === "describe" ? "description" : operation.kind === "prepare" ? "prepared" : operation.kind === "check_project_auth" ? "project_auth" : ["read_state", "archive_state"].includes(operation.kind) ? "state" : ["status", "connection", "credentials"].includes(operation.kind) ? operation.kind : "accepted")) { throw new Error("Plugin Runner returned an invalid operation result"); } @@ -353,7 +356,7 @@ export function createPluginRunnerTaskOwner(input: { // stop request. Completion also proves credential/writeback obligations ran. isRetired: () => stoppedState?.executionStopped === true && completed && !detached, rebind(next: typeof input) { - if (!launched || stopRequested || stoppedState || detached || next.harness !== input.harness || + if (!launched || stopRequested || stoppedState || detached || next.taskId !== taskId || next.harness !== input.harness || input.accountIdentity !== next.accountIdentity || input.credentialIdentity !== next.credentialIdentity || JSON.stringify(input.projectIds) !== JSON.stringify(next.projectIds)) throw new Error("Plugin warm task cannot be rebound"); input = next; diff --git a/server/src/services/plugin-task-credential-writeback.ts b/server/src/services/plugin-task-credential-writeback.ts index e13630b5f58..bd963b3e0b2 100644 --- a/server/src/services/plugin-task-credential-writeback.ts +++ b/server/src/services/plugin-task-credential-writeback.ts @@ -24,7 +24,7 @@ const handoffSchema = z.object({ state: z.enum(["pending", "complete"]), }).strict(); const key = "runnerTaskCredentialHandoff"; -type Scope = { companyId: string; leaseId: string }; +type Scope = { companyId: string; leaseId: string; expectedTaskId?: string }; /** Save authority, never credential bytes or a controller-local temporary path. */ export async function registerPluginTaskCredentialHandoff( @@ -126,13 +126,15 @@ export async function settlePluginTaskCredentialHandoff( metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ [key]: { ...handoff, state: "complete" } })}::jsonb`, }).where(and( eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId), + scope.expectedTaskId === undefined ? undefined : eq(environmentLeases.providerLeaseId, scope.expectedTaskId), sql`${environmentLeases.metadata}->${key} = ${JSON.stringify(handoff)}::jsonb`, )).returning({ id: environmentLeases.id }); if (completed.length !== 1) { - const [current] = await db.select({ metadata: environmentLeases.metadata }).from(environmentLeases).where(and( + const [current] = await db.select({ metadata: environmentLeases.metadata, taskId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and( eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId), )); - if (!isDeepStrictEqual(handoffSchema.safeParse(current?.metadata?.[key]).data, { ...handoff, state: "complete" })) { + if (scope.expectedTaskId !== undefined && current?.taskId !== scope.expectedTaskId || + !isDeepStrictEqual(handoffSchema.safeParse(current?.metadata?.[key]).data, { ...handoff, state: "complete" })) { throw new Error("Plugin task credential handoff changed"); } }