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"); } }