diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index 90591811789..62d97ad739c 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -84,6 +84,39 @@ authority as submission. Core requires preparation version 1 for native plugin execution. It validates and saves the result before it submits its configuration ID. +### Stopped source reconciliation + +A plugin’s `describe` response advertises `codexSourceRecoveryVersion: 1` for +Codex App Server and `acpxSourceRecoveryVersion: 1` for ACPX. It can advertise +either, both, or neither. Each field means the host can verify and restore a +stopped source for that provider without starting another turn. Core requires +the field for the selected provider and verifies the run’s eligibility. + +The authenticated request belongs to +the current run while `prepare` and `submit` carry the predecessor Runner identity. +Core derives this relationship from its saved source-consumer claim; a mismatched +run ID alone never grants recovery. Both operations must name the same `resumeFrom`, exact source Runner identity, and current +`accountIdentity`. The request and lease still belong to the current run. +Preparation must save this source binding with its new configuration. Submission +must verify the same binding before it starts the source process. + +Core requires the source run to have a committed successful result, an encrypted +PRP journal, confirmed physical termination, and no other source consumer. Core +claims the source under the current native controller lease. This does not change +the source run's status or result. A stale source controller cannot write its +journal after this claim. Workspace export and settlement holds still apply. +After source submission, the same current owner can prepare its attachment with +the exact source as `priorIdentity`. This does not grant ordinary warm reuse. + +The provider must verify the original account, workspace, runtime, and source +ownership before launch. The provider must reject saved active goals, pending tools, ambiguous starts, and +uncertain work before launch. Runner verifies that a saved idle provider remains +idle when restored. +Until the existing authenticated `run.attach` operation changes authority, the +Core retains semantic inputs and sends no fresh predecessor business commands +while the restored source reconciles its journal and known cleanup. Ordinary suspended recovery and live attachment keep their existing +checks. Providers that omit the capability do not support predecessor recovery. + ## Launch authentication validation Declare `projectAuthCheckVersion: 1` in `describe` for each native task launch. diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index c2d35cd32f8..35c816f46c2 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -76,7 +76,7 @@ the conversation. Core uses ordinary recovery for A, then attaches B with authenticated `run.attach`. If the original process is alive, its owner can verify and adopt it instead. -**Is A idle?** Before launch, the host driver (`sourceRecoveryVersion: 1`) +**Is A idle?** Before launch, the host driver (`codexSourceRecoveryVersion: 1`) checks that A’s process stopped and its saved session was idle. After launch, the Runner (`recoveryIntegrityVersion: 1`) verifies that Codex restored an idle session when no command effect is uncertain. It reads the goal, then the diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 523df43429d..61c4dc2f7c3 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -675,6 +675,7 @@ mod tests { let metadata = build_metadata(); assert_eq!(metadata["recoveryIntegrityVersion"], 1); assert!(metadata.get("sourceRecoveryVersion").is_none()); + assert!(metadata.get("codexSourceRecoveryVersion").is_none()); assert_eq!(metadata["acpxRecoveryIntegrityVersion"], 1); assert!(metadata.get("acpxSourceRecoveryVersion").is_none()); assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA); diff --git a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts index 482cd81261d..abb1e85b2a5 100644 --- a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts +++ b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts @@ -9,9 +9,9 @@ import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js"; * specifications stay private to the driver; plugins receive versioned requirements. */ export interface RunnerExecutionDriver { - /** Attests stopped ownership and dormant prelaunch admission for the selected provider. */ - sourceRecoveryVersion?: 1; - /** Same trusted prelaunch attestation for a qualified ACPX provider. */ + /** The driver verifies a stopped, idle Codex App Server source before launch. */ + codexSourceRecoveryVersion?: 1; + /** The driver verifies a stopped, idle source for the selected ACPX provider before launch. */ acpxSourceRecoveryVersion?: 1; /** Current authenticated lease/checkpoint-writer owner; may change during warm reuse. */ authorityRunId?: string; diff --git a/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts b/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts index 10dfe108514..90481f91449 100644 --- a/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts @@ -56,7 +56,7 @@ async function fixture(agent: "codex" | "claude" = "codex", nativeSource = false let recovering = false; const options: CapabilityRunnerdCodexTransportOptions = { provider: "acpx", acpxAgent: agent, runnerBinary: binary, stateDirectory: root, prpIdentity: identity, - runnerArtifact: readRunnerdArtifactBinding(binary), sourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1, + runnerArtifact: readRunnerdArtifactBinding(binary), codexSourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, runnerProcessReconnectGraceMs: null, runnerReconnectGraceMs: 3_000, controlPlaneRegistration: async next => { const generation = ++registrationGeneration; authority = next; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 8b81b27ad8f..67f21ab4fd3 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -5613,7 +5613,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { throw new Error("native_runner_source_reconciliation_requires_exact_authority"); } if (lostPredecessor && (!["codex", "acpx"].includes(this.options.provider ?? "codex") || this.options.acpxCandidateAdmission)) throw new Error("native_runner_recovery_integrity_unsupported"); - const recoveryCapability = this.options.provider === "acpx" ? this.options.acpxSourceRecoveryVersion : this.options.sourceRecoveryVersion; + const recoveryCapability = this.options.provider === "acpx" ? this.options.acpxSourceRecoveryVersion : this.options.codexSourceRecoveryVersion; if (lostPredecessor && recoveryCapability !== 1) { throw new Error("native_runner_recovery_integrity_unsupported"); } diff --git a/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts b/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts index 603f9123d63..0728c1b8832 100644 --- a/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts @@ -90,7 +90,7 @@ async function sourceFixture(registration?: CapabilityRunnerdCodexTransportOptio await vi.waitFor(() => { expect(absent(evidence.runnerPid!)).toBe(true); expect(absent(evidence.codexPid!)).toBe(true); }); expect((await read(runnerFile)).lifecycle).toBe("ready"); const recover = (overrides: Partial = {}) => { - const bundle = createCapabilityRunnerdCodexTransport({ ...options, sourceRecoveryVersion: 1, resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return Object.hasOwn(overrides, "authorityRunId") ? overrides.authorityRunId : "run-target"; } }); + const bundle = createCapabilityRunnerdCodexTransport({ ...options, codexSourceRecoveryVersion: 1, resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return Object.hasOwn(overrides, "authorityRunId") ? overrides.authorityRunId : "run-target"; } }); bundles.push(bundle); return bundle; }; return { root, options, identity, thread, runnerFile, providerFile, coreFile, calls, complete, settled, recover, dispose, nativeIdentity, execution, nativeSnapshot }; @@ -101,6 +101,8 @@ it.each(["isolated", "preconfigured"] as const)("recovers the real native backen const f = await sourceFixture(undefined, true, homeMode); try { expect(f.nativeSnapshot!.sessionId).toBe(f.thread.id); + await expect(f.recover({ codexSourceRecoveryVersion: undefined, acpxSourceRecoveryVersion: 1 }) + .transport.request("thread/read", {})).rejects.toThrow("recovery_integrity_unsupported"); let owner = "run-target"; const source = f.recover({ ...(homeMode === "isolated" ? { environment: { ...process.env, PAPERCLIP_RUNNER_HOME_MODE: "preconfigured" } } : {}), get authorityRunId() { return owner; } }); await expect(source.transport.request("thread/read", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch"); @@ -198,7 +200,7 @@ it.each([["launch", "default"], ["adopt", "default"], ["adopt-denied", "default" expect((await prepare(requirements)).provider).toEqual(provider); prepare.mockClear(); const binary = fixture.options.runnerBinary; let owner: string | undefined = "prepared-target"; - const sourceOptions: Partial = { sourceRecoveryVersion: 1, get authorityRunId() { return owner; }, + const sourceOptions: Partial = { codexSourceRecoveryVersion: 1, get authorityRunId() { return owner; }, recoveryLaunch: { requirements }, prepareLaunch: prepare, disableCodexWebSearch: policy === "conflict" ? false : undefined, runnerArtifact: { version: "0.3.0", digest: `sha256:${createHash("sha256").update(await readFile(binary)).digest("hex")}` }, diff --git a/packages/plugins/sdk/src/environment-tasks.ts b/packages/plugins/sdk/src/environment-tasks.ts index f2ded292d19..017a0d0fa92 100644 --- a/packages/plugins/sdk/src/environment-tasks.ts +++ b/packages/plugins/sdk/src/environment-tasks.ts @@ -137,6 +137,9 @@ export const environmentTaskResultSchema = z.discriminatedUnion("kind", [ warmTaskVersion: z.literal(1).optional(), launchPreparationVersion: z.literal(1).optional(), executionContextVersion: z.literal(1).optional(), + /** Each version attests stopped-source recovery for its provider independently. */ + codexSourceRecoveryVersion: z.literal(1).optional(), + acpxSourceRecoveryVersion: z.literal(1).optional(), freshWorkspaceVersion: z.literal(1).optional(), projectAuthCheckVersion: z.literal(1).optional(), workspaceFileReadVersion: z.literal(1).optional(), diff --git a/packages/plugins/sdk/tests/environment-tasks.test.ts b/packages/plugins/sdk/tests/environment-tasks.test.ts index 47f368f909d..43521e36ae7 100644 --- a/packages/plugins/sdk/tests/environment-tasks.test.ts +++ b/packages/plugins/sdk/tests/environment-tasks.test.ts @@ -1,5 +1,5 @@ import {describe,expect,it} from "vitest"; -import {environmentTaskOperationSchema} from "../src/environment-tasks.js"; +import {environmentTaskOperationSchema,environmentTaskResultSchema} from "../src/environment-tasks.js"; const submit={kind:"submit",projectIds:[],bootstrapTicket:"synthetic",runner:{harness:"codex",protocolMin:1,protocolMax:3,runnerId:"runner",leaseId:"lease",runId:"run",sessionId:"session",turnId:"turn",itemId:"item"}, launch:{runnerVersion:"fixture",runnerDigest:`sha256:${"a".repeat(64)}`,maxOutboxBytes:262144,p0ReserveBytes:65536,lifecyclePolicy:{mode:"warm",idleTimeoutMs:300000}}}; describe("task lifecycle policy",()=>{ @@ -24,6 +24,31 @@ const prepare = { }, }; +describe("retained source launch fields", () => { + it("uses the existing source identity and resume reference without a recovery mode", () => { + for (const operation of [prepare, submit]) { + const recovery = { ...operation, accountIdentity: "selected-account", resumeFrom: "source-task" }; + expect(environmentTaskOperationSchema.parse(recovery)).toEqual(recovery); + expect(environmentTaskOperationSchema.safeParse({ ...recovery, recoveryPurpose: "source_reconcile" }).success).toBe(false); + } + }); +}); + +describe("provider-specific source recovery capability", () => { + const description = { kind: "description", taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; + it.each(["codexSourceRecoveryVersion", "acpxSourceRecoveryVersion"] as const)("accepts %s independently", key => { + const advertised = { ...description, [key]: 1 }; + expect(environmentTaskResultSchema.parse(advertised)).toEqual(advertised); + expect(environmentTaskResultSchema.safeParse({ ...description, [key]: 2 }).success).toBe(false); + }); + it("accepts both or neither and rejects an unqualified recovery flag", () => { + const both = { ...description, codexSourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1 }; + expect(environmentTaskResultSchema.parse(both)).toEqual(both); + expect(environmentTaskResultSchema.parse(description)).toEqual(description); + expect(environmentTaskResultSchema.safeParse({ ...description, sourceRecoveryVersion: 1 }).success).toBe(false); + }); +}); + describe("task hosted search restriction", () => { it("preserves Core's explicit restriction through plugin launch preparation", () => { const operation = { ...prepare, requirements: { ...prepare.requirements, diff --git a/scripts/__tests__/run-vitest-stable-shard.test.mjs b/scripts/__tests__/run-vitest-stable-shard.test.mjs index 1494fb0762c..9804de802a9 100644 --- a/scripts/__tests__/run-vitest-stable-shard.test.mjs +++ b/scripts/__tests__/run-vitest-stable-shard.test.mjs @@ -323,6 +323,7 @@ const nativeRunnerSuitePath = const ingressRetrySuitePath = "server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts"; const usageDeliverySuitePath = "server/src/services/native-runtime/native-usage-delivery.integration.test.ts"; const warmCheckpointOwnershipSuitePath = "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts"; +const pluginWarmOwnerSuitePath = "server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts"; const dotRunnerSuitePath = "server/src/__tests__/dot-runner.test.ts"; const restartRecoverySuitePath = "server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts"; @@ -371,7 +372,8 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes assert.ok(!files.includes(restartRecoverySuitePath)); assert.ok(!files.includes(usageDeliverySuitePath)); assert.ok(!files.includes(warmCheckpointOwnershipSuitePath)); - assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort()); + assert.ok(!files.includes(pluginWarmOwnerSuitePath)); + assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort()); assert.equal(new Set(files).size, files.length); const defaultRun = dryRunJson([], prEnv); assert.ok(defaultRun.generalServerSuiteCount === full.generalServerSuiteCount); @@ -392,6 +394,7 @@ for (const [caller, envOverrides] of [["Release", { GITHUB_WORKFLOW: "Release" } assert.ok(files.includes(dotRunnerSuitePath)); assert.ok(files.includes(restartRecoverySuitePath)); assert.ok(files.includes(warmCheckpointOwnershipSuitePath)); + assert.ok(files.includes(pluginWarmOwnerSuitePath)); assert.deepEqual([...files, chatSuitePath].sort(), full.selectedGeneralServerSuites.sort()); assert.equal(new Set(files).size, files.length); }); @@ -425,6 +428,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => { ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, + pluginWarmOwnerSuitePath, ]); }); diff --git a/scripts/run-vitest-stable.mjs b/scripts/run-vitest-stable.mjs index fa74e5eef99..f5dde1a91eb 100644 --- a/scripts/run-vitest-stable.mjs +++ b/scripts/run-vitest-stable.mjs @@ -104,6 +104,7 @@ const nativeRunnerSuites = [ "server/src/services/native-runtime/native-runner-restart-diskless.integration.test.ts", "server/src/services/native-runtime/native-usage-delivery.integration.test.ts", "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts", + "server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts", ]; // In the PR workflow (pr.yml, the caller of pr-trusted.yml — reusable // workflows inherit the caller's GITHUB_WORKFLOW), the last Verify Paperclip diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index 1bc5ed54b1e..4b6489243cc 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -1,17 +1,20 @@ +import { resolvePluginTaskResume } from "./plugin-task-resume.js"; +export { resolvePluginTaskResume } from "./plugin-task-resume.js"; import { attachSandboxCreation, environmentService } from "./environments.js"; import { createOwnedSandbox, queueSandboxCreationCleanup, recordSandboxCreationResource } from "./sandbox-creation-operations.js"; import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; import { readWorkspaceProviderBinding } from "@paperclipai/shared"; import { settlePluginTaskCredentialHandoff } from "./plugin-task-credential-writeback.js"; -import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { and, eq, inArray, sql } from "drizzle-orm"; import { environmentLeases, executionWorkspaces, environments, heartbeatRuns, issues, projects, type Db } from "@paperclipai/db"; import { freshWorkspaceSchema, environmentTaskOperationSchema, parseEnvironmentTaskResult, type HostToWorkerMethodName, type HostToWorkerMethods, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import { pluginRegistryService } from "./plugin-registry.js"; -import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt } from "./remote-execution-termination.js"; +import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import { NativeExecutionContext } from "./native-runtime/native-execution-context.js"; +import { assertPluginTaskSourceAttachment, assertPluginTaskSourceRecovery } from "./native-runtime/plugin-task-source-recovery.js"; import { pluginAgentRuntimeIdentity } from "./plugin-agent-runtime-state.js"; export async function pluginEnvironmentSupportsTasks(db: Db, workers: PluginWorkerManager, lease: { companyId: string; metadata: Record | null }) { @@ -56,6 +59,11 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage const taskId = row.lease.providerLeaseId; if (input.expectedTaskId !== undefined && input.expectedTaskId !== taskId) throw new Error("Plugin Runner physical task changed"); const metadata = lease.metadata ?? {}; + // A different launch identity is authorized by the saved consumer claim, never by the mismatch alone. + const sourceRecovery = run && ((operation.kind === "prepare" && operation.requirements.identity.runId !== run.id && metadata.runnerTaskSourceRecovery) || + (operation.kind === "submit" && operation.runner.runId !== run.id && metadata.runnerTaskSourceRecovery)) + ? await assertPluginTaskSourceRecovery(db, { companyId: input.companyId, leaseId: lease.id, + context: context instanceof NativeExecutionContext ? context : undefined, operation }) : undefined; const probe = metadata.runnerTaskProbe as { runId?: unknown; sessionId?: unknown; expiresAtMs?: unknown } | undefined; const activeProbe = lease.heartbeatRunId === null && lease.issueId === null && probe && typeof probe.runId === "string" && typeof probe.sessionId === "string" @@ -94,15 +102,19 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage if ((operation.kind === "read_file" || operation.kind === "describe" || operation.kind === "prepare" || operation.kind === "submit" || (operation.kind === "connection" && !retainedConnection) || (operation.kind === "read_state" && metadata.runnerTaskExecutionStopped !== true && !retainedCheckpointRead) || operation.kind === "archive_state" || operation.kind === "check_project_auth" || operation.kind === "copilot_metadata") && (metadata.runnerTaskRetiring === true || !environment || (!activeProbe && run?.status !== "running") || lease.status !== "active" || (lease.expiresAt && lease.expiresAt.getTime() <= Date.now()))) throw new Error("Plugin-provided Runner execution lease is not active"); if (operation.kind === "copilot_metadata" && !activeProbe) throw new Error("Copilot metadata requires a private setup lease"); - const warmAttachment = !!prior && metadata.runnerTaskReused === true && + const sourceReference = metadata.runnerTaskSourceRecovery as { identity?: { runId?: unknown } } | undefined; + const sourceAttachment = operation.kind === "prepare" && prior && sourceReference?.identity?.runId === prior.runId + ? await assertPluginTaskSourceAttachment(db, { companyId: input.companyId, leaseId: lease.id, + context: context instanceof NativeExecutionContext ? context : undefined, operation }) : undefined; + const warmAttachment = !!prior && (metadata.runnerTaskReused === true || !!sourceAttachment) && (matches(prior, savedIdentity) || matches(prior, priorSaved) && operation.kind === "prepare" && matches(operation.requirements.identity, savedIdentity)) && runner?.runnerId === prior.runnerInstanceId && runner?.leaseId === prior.environmentLeaseId && runner?.sessionId === prior.normalizedSessionId; if (prior && !warmAttachment) throw new Error("Plugin warm task prior identity mismatch"); - if (runner && ((!checkpointIdentity || !resumeFrom) && !(ownCheckpoint && operation.kind !== "read_file") && runner.runId !== (activeProbe ? probe.runId : run?.id) || (!resumeFrom && runner.leaseId !== lease.id && !ownCheckpoint && !warmAttachment) + if (runner && !sourceRecovery && ((!checkpointIdentity || !resumeFrom) && !(ownCheckpoint && operation.kind !== "read_file") && runner.runId !== (activeProbe ? probe.runId : run?.id) || (!resumeFrom && runner.leaseId !== lease.id && !ownCheckpoint && !warmAttachment) || (activeProbe && (runner.sessionId !== probe.sessionId || (launches && operation.projectIds.length > 0))))) { throw new Error("Plugin-provided Runner execution identity mismatch"); } - if (runner && resumeFrom) { + if (runner && resumeFrom && !sourceRecovery) { const inspectCheckpoint = operation.kind === "read_state"; const sources = await db.select({ id: environmentLeases.id, runId: environmentLeases.heartbeatRunId, metadata: environmentLeases.metadata }).from(environmentLeases).where(and( eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.provider, lease.provider!), @@ -373,6 +385,14 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage const result = await callWorker("environmentTask", params, timeout); if (executionContext) context!.assertOpen(); const parsed = parseEnvironmentTaskResult(operation, taskId, result); + if (sourceRecovery && (operation.kind === "prepare" || operation.kind === "submit")) { + await assertPluginTaskSourceRecovery(db, { companyId: input.companyId, leaseId: lease.id, + context: context instanceof NativeExecutionContext ? context : undefined, operation }); + } + if (sourceAttachment && operation.kind === "prepare") { + await assertPluginTaskSourceAttachment(db, { companyId: input.companyId, leaseId: lease.id, + context: context instanceof NativeExecutionContext ? context : undefined, operation }); + } if (operation.kind === "read_file") { const [current] = await db.select({ lease: environmentLeases, run: heartbeatRuns }).from(environmentLeases) .leftJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), eq(heartbeatRuns.companyId, environmentLeases.companyId))) @@ -470,48 +490,3 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor throw new Error("Plugin Runner termination is unconfirmed; the saved lease requires cleanup"); } } - -/** A saved provider task can resume only within the same authorized agent workspace. */ -export async function resolvePluginTaskResume(db: Db, input: { - companyId: string; leaseId: string; - priorIdentity: { runnerInstanceId: string; environmentLeaseId: string; runId: string; normalizedSessionId: string }; - /** Read-only discovery; launching still requires an attested retained checkpoint. */ - inspectCheckpoint?: boolean; -}): Promise { - const rows = await db.select({ lease: environmentLeases, run: heartbeatRuns }).from(environmentLeases) - .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), eq(heartbeatRuns.companyId, input.companyId))) - .where(and(eq(environmentLeases.companyId, input.companyId), - or(eq(environmentLeases.id, input.leaseId), eq(heartbeatRuns.id, input.priorIdentity.runId)))); - const next = rows.find(row => row.lease.id === input.leaseId); - // A retained PRP lease is logical; later provider tasks have different physical lease IDs. - const candidates = rows.filter(row => { - const identity = row.lease.metadata?.runnerTaskIdentity as Record | undefined; - return row.lease.id !== input.leaseId && row.run.id === input.priorIdentity.runId && - (input.inspectCheckpoint || row.lease.metadata?.runnerTaskSessionRetained === true) && - identity?.runnerId === input.priorIdentity.runnerInstanceId && identity.leaseId === input.priorIdentity.environmentLeaseId && - identity.sessionId === input.priorIdentity.normalizedSessionId && identity.runId === input.priorIdentity.runId && - !rows.some(successor => successor.lease.id !== input.leaseId && - successor.lease.metadata?.runnerTaskResumeFrom === row.lease.providerLeaseId); - }); - const old = candidates.length === 1 ? candidates[0] : undefined; - const identity = old?.lease.metadata?.runnerTaskIdentity as Record | undefined; - if (!old || !next || !old.lease.providerLeaseId || !identity || - identity.runnerId !== input.priorIdentity.runnerInstanceId || identity.leaseId !== input.priorIdentity.environmentLeaseId || - identity.sessionId !== input.priorIdentity.normalizedSessionId || identity.runId !== input.priorIdentity.runId || - old.lease.metadata?.driver !== "plugin" || next.lease.metadata?.driver !== "plugin" || - !next.lease.providerLeaseId || next.lease.providerLeaseId === old.lease.providerLeaseId || - (old.run.id !== next.run.id && !["succeeded", "interrupted", "failed", "cancelled", "timed_out"].includes(old.run.status)) || - old.run.agentId !== next.run.agentId || old.lease.issueId !== next.lease.issueId || - old.lease.executionWorkspaceId !== next.lease.executionWorkspaceId || - !old.lease.environmentId || old.lease.environmentId !== next.lease.environmentId || - !next.lease.provider?.startsWith("plugin:") || old.lease.provider !== next.lease.provider || - old.lease.metadata?.pluginId !== next.lease.metadata?.pluginId || old.lease.metadata?.driverKey !== next.lease.metadata?.driverKey || - next.lease.status !== "active" || next.lease.releasedAt || next.run.status !== "running" || - (next.lease.expiresAt && next.lease.expiresAt.getTime() <= Date.now()) || - !(old.run.id === next.run.id - ? rows.filter(row => row.lease.id !== next.lease.id).every(row => hasRemoteTerminationReceipt(row.lease)) - : await remoteExecutionHasStopped(db, input.companyId, old.run.id))) { - throw new Error("Plugin Runner retained session is not authorized"); - } - return old.lease.providerLeaseId; -} diff --git a/server/src/services/native-runtime/native-execution-context.ts b/server/src/services/native-runtime/native-execution-context.ts index e635061d2b6..41614d52f62 100644 --- a/server/src/services/native-runtime/native-execution-context.ts +++ b/server/src/services/native-runtime/native-execution-context.ts @@ -15,6 +15,7 @@ export class NativeExecutionContext { #transport?: Promise; #target?: string; #finished = false; + #nativeLease?: { bootId: string; leaseOwner: string; generation: number }; pluginSupported = false; constructor(execution: NativeExecutionInput, timeoutMs: number) { @@ -42,6 +43,19 @@ export class NativeExecutionContext { this.assertOpen(); } + /** Bind only the coordinator lease actually acquired by this execution. */ + bindNativeLease(owner: { bootId: string; leaseOwner: string; generation: number }): void { + this.assertOpen(); + if (this.#nativeLease) { this.assertNativeLease(owner); return; } + this.#nativeLease = { ...owner }; + } + + assertNativeLease(owner: { bootId: string; leaseOwner: string; generation: number }): void { + this.assertOpen(); + if (!this.#nativeLease || this.#nativeLease.bootId !== owner.bootId || this.#nativeLease.leaseOwner !== owner.leaseOwner || + this.#nativeLease.generation !== owner.generation) throw new Error("native_execution_context_owner_changed"); + } + async wait(operation: () => Promise): Promise { this.assertOpen(); let remove = () => {}; diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index e23de011265..9e7340553d5 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -4,6 +4,7 @@ import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunn import { inspectDurablePrpCheckpointHeader, inspectUnclaimedBootstrapCheckpoint, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import { legacyControllerBootId } from "../legacy-controller-lease.js"; +import { assertNativeSourceCheckpointConsumer, lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js"; type Binding = { companyId: string; issueId: string; runId: string }; export const NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES = 1024 * 1024; @@ -85,6 +86,7 @@ export async function validateNativeRunnerCheckpointOwner( const activeOwner = owner.leaseOwner !== null && current.leaseOwner === owner.leaseOwner && current.leaseExpiresAt !== null && current.leaseExpiresAt > new Date(owned.now); if (!activeOwner) { + await assertNativeSourceCheckpointConsumer(tx, owner); const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))); const [successor] = task?.executionRunId && task.executionRunId !== owner.runId @@ -119,6 +121,10 @@ export function nativeRunnerCheckpointWriter(db: Db, owner: NativeRunnerCheckpoi if (checkpoint.runId !== owner.runId) throw new Error("native_runner_checkpoint_binding_mismatch"); const changed = encryptParts(owner, checkpoint.changedRecords); await persistenceDb.transaction(async tx => { + if (owner.runId !== owner.authorityRunId) { + await lockSourceRecoveryCoordinators(tx, owner.companyId, [owner.runId, owner.authorityRunId]); + await assertNativeSourceCheckpointConsumer(tx, owner, checkpoint); + } await assertNativeRunnerCheckpointOwner(tx, owner, checkpoint.workCommandIds.some(id => !previousWork.has(id))); if (checkpoint.predecessor) { const [prior] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( 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 15021181cfa..e7c63c6d419 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -164,6 +164,7 @@ type BackendFactoryOptions = { }; type RunnerTransportOptions = { + authorityRunId?: string; onAccountingEvent?: (event: PrpEvent) => Promise; onRecoveryAccountingEvent?: unknown; environment?: NodeJS.ProcessEnv; @@ -7924,7 +7925,8 @@ describe("native terminal-turn accounting", () => { binding: { ...execution.binding, runId: "plugin-usage-run", executionWorkspaceId: "plugin-usage-workspace" }, session: { ...execution.session, normalizedSessionId: "plugin-usage-session" } } as NativeExecutionInput; const launcher = vi.fn(), registration = vi.fn(); - const driver = { prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration }; + let authorityRunId = current.binding.runId; + const driver = { get authorityRunId() { return authorityRunId; }, prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration }; const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue( { launcher, registration, driver, readWorkspaceFile: vi.fn(), isRetired: () => false, rebind: vi.fn() } as never); const task = vi.fn(async (operation: unknown) => { @@ -7944,6 +7946,9 @@ describe("native terminal-turn accounting", () => { await options.onAccountingEvent!(report); expect(capture).toHaveBeenCalledExactlyOnceWith(report); expect(options.onRecoveryAccountingEvent).toEqual(expect.any(Function)); + expect(options.authorityRunId).toBe(current.binding.runId); + authorityRunId = "next-plugin-owner"; + expect(options.authorityRunId).toBe("next-plugin-owner"); } finally { owner.mockRestore(); if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 69ff4f36ba5..6f7d5deced1 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -13595,7 +13595,7 @@ async function createRunnerdBackendWithinSessionClaim( ...(input.recoveryPurpose === "stop_cleanup" ? { recoveryPurpose: "stop_cleanup" as const } : {}), measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined, recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined, - ...owner.driver, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? description.runner, + ...owner.driver, get authorityRunId() { return owner.driver.authorityRunId; }, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? description.runner, controlPlaneRegistration: input.controllerPersistence ? async (authority, identity, signal) => { await nativeRunnerCheckpointRegistration({ ...nativeRunnerCheckpointPersistence(input.db, input.controllerPersistence!.bootId, diff --git a/server/src/services/native-runtime/native-source-recovery-ownership.ts b/server/src/services/native-runtime/native-source-recovery-ownership.ts new file mode 100644 index 00000000000..eeec0f1196f --- /dev/null +++ b/server/src/services/native-runtime/native-source-recovery-ownership.ts @@ -0,0 +1,44 @@ +import { and, asc, eq, inArray, sql } from "drizzle-orm"; +import { environmentLeases, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { inspectDurablePrpCheckpointHeader, type DurablePrpCheckpoint } from "../../vendor/paperclip-runner/index.js"; + +type Transaction = Parameters[0]>[0]; +const record = (value: unknown): Record => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +/** Claims and recovered-source writers lock both coordinators in this order. */ +export async function lockSourceRecoveryCoordinators(tx: Transaction, companyId: string, runIds: readonly string[]) { + return tx.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId), + inArray(nativeRunFinalizations.runId, [...new Set(runIds)]))).orderBy(asc(nativeRunFinalizations.runId)).for("update"); +} + +export async function readNativeSourceConsumers(tx: Transaction, companyId: string, runId: string) { + return tx.select({ id: environmentLeases.id, metadata: environmentLeases.metadata }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, runId), + sql`${environmentLeases.metadata}->'runnerTaskSourceConsumer' is not null`)); +} + +/** Physical source consumption revokes the completed controller's late writer. */ +export async function assertNativeSourceCheckpointConsumer(tx: Transaction, owner: { + companyId: string; runId: string; authorityRunId: string; bootId: string; generation: number; +}, checkpoint?: DurablePrpCheckpoint) { + const rows = await readNativeSourceConsumers(tx, owner.companyId, owner.runId); + if (!rows.length) return; + const deny = (): never => { throw new Error("native_runner_checkpoint_owner_lost"); }; + if (owner.runId === owner.authorityRunId || rows.length !== 1 || !checkpoint) return deny(); + const source = rows[0]!, consumer = record(source.metadata?.runnerTaskSourceConsumer); + if (typeof consumer.leaseId !== "string" || consumer.runId !== owner.authorityRunId) return deny(); + const [lease] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, owner.companyId), + eq(environmentLeases.id, consumer.leaseId))); + const recovery = record(lease?.metadata?.runnerTaskSourceRecovery); + const header = checkpoint.records.get("header"); + const identity = header ? inspectDurablePrpCheckpointHeader(header) : null; + const expected = record(recovery.identity); + if (!lease || lease.heartbeatRunId !== owner.authorityRunId || lease.status !== "active" || lease.releasedAt || + lease.expiresAt && lease.expiresAt.getTime() <= Date.now() || + recovery.schema !== "paperclip.runner.source-recovery.v1" || recovery.sourceLeaseId !== source.id || + recovery.consumerLeaseId !== lease.id || recovery.consumerRunId !== owner.authorityRunId || + typeof recovery.accountIdentity !== "string" || recovery.accountIdentity !== consumer.accountIdentity || + recovery.controllerBootId !== owner.bootId || recovery.controllerGeneration !== owner.generation || + ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].some(key => + typeof expected[key] !== "string" || expected[key] !== record(identity)[key] || expected[key] !== record(consumer.identity)[key])) return deny(); +} diff --git a/server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts b/server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts new file mode 100644 index 00000000000..7e30a6d889d --- /dev/null +++ b/server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts @@ -0,0 +1,83 @@ +import { existsSync } from "node:fs"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { spawn } from "node:child_process"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it } from "vitest"; +import { createRunnerdCodexTransport, createRunnerProviderLaunch, defaultCapabilityRunnerdBinary, + runnerLaunchRequirementsDigest, readRunnerdArtifactBinding, type RunnerProcessLaunchSpec, type RunnerProviderLaunchRequirements } from "@paperclipai/paperclip-runner"; +import { createPluginRunnerTaskOwner } from "./plugin-runner-task.js"; + +const fakeCodex = resolve(import.meta.dirname, "../../../../packages/paperclip-runner/runner/target/debug/fake-codex-app-server"); + +it.skipIf(!existsSync(defaultCapabilityRunnerdBinary()) || !existsSync(fakeCodex))( + "keeps the current plugin owner through a real ordinary warm attachment", async () => { + const root = await mkdtemp(join(tmpdir(), "plugin-warm-owner-")), calls = join(root, "calls"); + const source = { runnerInstanceId: "warm-runner", environmentLeaseId: "warm-lease", runId: "first-run", + normalizedSessionId: "warm-session", turnId: "first-turn", itemId: "first-item" }; + const target = { ...source, runId: "second-run", turnId: "second-turn", itemId: "second-item" }; + let spec: RunnerProcessLaunchSpec | undefined, child: ReturnType | undefined; + let submits = 0, stopped = false; + const task = async (operation: unknown): Promise => { + const op = operation as { kind: string; requirements: RunnerProviderLaunchRequirements; configurationId: string; provider: string }; + if (op.kind === "prepare") return { kind: "prepared", taskId: "warm-task", launch: { + configurationId: `configuration-${op.requirements.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(op.requirements), + provider: createRunnerProviderLaunch(op.requirements, { options: { codexArgs: ["--state-file", join(root, "fake.json"), "--call-log", calls, "--record-process-start"] }, + root, cwd: root, codexHome: join(root, "codex-home"), codexCommand: fakeCodex, providerNodeCommand: process.execPath, + opencodeProxyPath: "", acpxSidecarPath: "" }), + } }; + if (op.kind === "check_project_auth") return { kind: "project_auth", taskId: "warm-task", configurationId: op.configurationId, provider: op.provider, allowed: true }; + if (op.kind === "submit") { + submits++; + child = spawn(spec!.command, [...spec!.args], { cwd: spec!.cwd, env: spec!.environment, stdio: "ignore", detached: true }); + } + if (op.kind === "stop") { stopped = true; if (child?.pid) process.kill(-child.pid, "SIGTERM"); } + if (op.kind === "status") return { kind: "status", taskId: "warm-task", phase: stopped ? "cancelled" : "running", + executionStopped: stopped && child?.exitCode != null, sessionStateAvailable: false }; + if (op.kind === "describe") return { kind: "description", taskId: "warm-task", warmTaskVersion: 1, + runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; + return { kind: "accepted", taskId: "warm-task" }; + }; + const input = { taskId: "warm-task", task, harness: "codex", projectIds: [], projectAuthCheckVersion: 1 as const, + authorityRunId: source.runId, accountIdentity: "synthetic-account", warmTaskVersion: 1 as const, pollMs: 5 }; + const owner = createPluginRunnerTaskOwner(input); + const launcher = owner.driver.runnerProcessLauncher!; + const cores = new Set(); + const bundle = createRunnerdCodexTransport({ ...owner.driver, + get authorityRunId() { return owner.driver.authorityRunId; }, + runnerProcessLauncher: next => { spec = next; return launcher(next); }, + stateDirectory: root, runnerBinary: defaultCapabilityRunnerdBinary(), runnerArtifact: readRunnerdArtifactBinding(defaultCapabilityRunnerdBinary()), prpIdentity: source, + codexCommand: fakeCodex, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, + // This fixture isolates ordinary owner rotation from hosted ingress timing. + controlPlaneRegistration: async core => { + if (!cores.has(core)) { cores.add(core); await core.start(); } + return { connectUrl: core.connectUrl, release: () => {} }; + }, + }); + try { + const opened = await bundle.transport.request("thread/start", { cwd: root, dynamicTools: [] }); + const turn = async () => { + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Complete once." }] }); + for await (const event of bundle.transport.notifications()) if (event.method === "turn/completed") return; + throw new Error("Warm fixture ended without completion"); + }; + await turn(); + const pid = child!.pid; + owner.rebind({ ...input, authorityRunId: target.runId }); + await bundle.transport.attachRun!(target); + await turn(); + const retained = await bundle.transport.request("thread/read", {}); + expect(retained.thread).toMatchObject({ id: (opened.thread as { id: string }).id }); + expect(child!.pid).toBe(pid); expect(submits).toBe(1); + const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/); + expect(methods.filter(method => method === "process-start")).toHaveLength(1); + expect(methods.filter(method => method === "turn/start")).toHaveLength(2); + } finally { + const providerPid = bundle.evidence().providerPid; + await bundle.detachControllerForRestart().catch(() => undefined); + for (const pid of [providerPid, child?.pid]) if (pid) try { process.kill(-pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } + for (const core of cores) await core.stop(); + await rm(root, { recursive: true, force: true }); + } + }, 30_000, +); 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 2ce25e84837..b9ac804b10d 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -59,6 +59,81 @@ describe("plugin native Runner ownership", () => { expect(() => createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex" } as never)).toThrow("physical task unavailable"); }); + it("publishes current trusted recovery authority through warm owner rebind", () => { + const { task } = provider(); + const input = { taskId: "task", task, projectIds: [], harness: "codex", authorityRunId: "first-owner", codexSourceRecoveryVersion: 1 as const, adoptIdentity: requirements.identity }; + const owner = createPluginRunnerTaskOwner(input); + expect(owner.driver.authorityRunId).toBe("first-owner"); + expect(owner.driver.codexSourceRecoveryVersion).toBe(1); + owner.rebind({ ...input, authorityRunId: "next-owner", codexSourceRecoveryVersion: undefined }); + expect(owner.driver.authorityRunId).toBe("next-owner"); + expect(owner.driver.codexSourceRecoveryVersion).toBeUndefined(); + }); + it.each(["capability", "source", "identity", "account", "provider", "adoption"])("refuses an unqualified predecessor preparation: %s", async mode => { + const { task } = provider(); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: mode === "provider" ? "claude" : "codex", + authorityRunId: "next", codexSourceRecoveryVersion: mode === "capability" ? undefined : 1, + resumeFrom: mode === "source" ? undefined : "source-task", priorIdentity: mode === "identity" ? undefined : requirements.identity, + accountIdentity: mode === "account" ? undefined : "selected-account", ...(mode === "adoption" ? { adoptIdentity: requirements.identity } : {}), + }); + await expect(prepare(owner)).rejects.toThrow("source reconciliation unavailable"); + expect(task).not.toHaveBeenCalled(); + }); + it("reuses exact adopted A preparation under current B without a source launch", async () => { + const { task } = provider(); + const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "next" } } as never, 1_000); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, + executionContext: context, adoptIdentity: requirements.identity }); + const recovery = { configurationId: "retained-configuration", requirementsDigest: `sha256:${"a".repeat(64)}` as const, provider: { kind: "codex" } }; + try { + await expect(owner.driver.prepareLaunch!(requirements, recovery)).resolves.toEqual(recovery); + expect(task.mock.calls.map(([operation]) => operation.kind)).toEqual(["check_project_auth"]); + } finally { context.finish(); } + }); + it.each(["same-run", "suspended-predecessor"])("keeps ordinary %s preparation independent of predecessor capability", async mode => { + const { task } = provider(); + const identity = mode === "same-run" ? requirements.identity : { ...requirements.identity, runId: "next" }; + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, + authorityRunId: identity.runId, priorIdentity: requirements.identity, resumeFrom: "source-task" }); + await prepare(owner, { ...requirements, identity }); + expect(task.mock.calls[0][0]).toMatchObject({ kind: "prepare", resumeFrom: "source-task", requirements: { identity } }); + expect(task.mock.calls[0][0]).not.toHaveProperty("recoveryPurpose"); + }); + it("keeps the resume parent on ordinary repeated preparation after launch", async () => { + const { task } = provider(); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, pollMs: 1, + authorityRunId: "run", resumeFrom: "source-task", priorIdentity: requirements.identity }); + await prepare(owner); + const registration = await owner.registration({} as never, "run"); + try { + owner.launcher(launchSpec()); + await vi.waitFor(() => expect(task.mock.calls.some(([op]) => op.kind === "submit")).toBe(true)); + await prepare(owner); + expect(task.mock.calls.map(([op]) => op).filter(op => op.kind === "prepare").at(-1)).toMatchObject({ resumeFrom: "source-task" }); + } finally { await registration.release("detach"); } + }); + it("prepares and submits exact A, then prepares B without copying or reconciling A again", async () => { + const { task } = provider(); + const owner = createPluginRunnerTaskOwner({ taskId: "task", task, projectIds: [], harness: "codex", projectAuthCheckVersion: 1, pollMs: 1, + authorityRunId: "next", codexSourceRecoveryVersion: 1, resumeFrom: "source-task", priorIdentity: requirements.identity, + accountIdentity: "selected-account" }); + await expect(prepare(owner, { ...requirements, identity: { ...requirements.identity, runId: "foreign" } })).rejects.toThrow("source reconciliation unavailable"); + await prepare(owner); + const registration = await owner.registration({} as never, "run"); + try { + owner.launcher(launchSpec()); + await vi.waitFor(() => expect(task.mock.calls.some(([op]) => op.kind === "submit")).toBe(true)); + const initial = task.mock.calls.map(([op]) => op).filter(op => ["prepare", "submit"].includes(op.kind)); + expect(initial).toEqual([expect.objectContaining({ kind: "prepare", resumeFrom: "source-task", requirements: expect.objectContaining({ identity: requirements.identity }) }), + expect.objectContaining({ kind: "submit", resumeFrom: "source-task", runner: expect.objectContaining({ runId: "run", turnId: "turn", itemId: "item" }) })]); + await prepare(owner); + expect(task.mock.calls.map(([op]) => op).filter(op => op.kind === "prepare").at(-1)).toMatchObject({ resumeFrom: "source-task" }); + await prepare(owner, { ...requirements, identity: { ...requirements.identity, runId: "next", turnId: "next-turn", itemId: "next-item" } }); + const attached = task.mock.calls.map(([op]) => op).filter(op => op.kind === "prepare").at(-1); + expect(attached).toMatchObject({ priorIdentity: requirements.identity, requirements: { identity: { runId: "next" } } }); + expect(attached).not.toHaveProperty("recoveryPurpose"); expect(attached).not.toHaveProperty("resumeFrom"); + } finally { await registration.release("detach"); } + }); it.each(["finish", "rebind"])("does not send startup Stop after the context %s", mode => { const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); const next = new NativeExecutionContext(context.execution, 1_000); diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index 3857dcfa431..fa14bd0ea55 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -36,6 +36,9 @@ export function createPluginRunnerTaskOwner(input: { projectAuthCheckVersion?: 1; resumeFrom?: string; priorIdentity?: DurableRecoveryIdentity; + /** Current authenticated lease owner; the restored PRP identity may belong to its predecessor. */ + authorityRunId?: string; + codexSourceRecoveryVersion?: 1; credentialIdentity?: string; accountIdentity?: string; credential?: PluginEnvironmentTaskCredential; @@ -55,6 +58,7 @@ export function createPluginRunnerTaskOwner(input: { let preparedLaunchId: string | undefined; let preparedIdentity: DurableRecoveryIdentity | undefined = input.adoptIdentity; let launched = !!input.adoptIdentity; + let launchIdentity = input.adoptIdentity; let stopRequested = false; let released = false; let submitted = !!input.adoptIdentity; @@ -119,6 +123,13 @@ export function createPluginRunnerTaskOwner(input: { await authorizeRunnerAdoption(recovery.configurationId, requirements.identity); return recovery; } + const authorityRunId = input.authorityRunId ?? input.executionContext?.execution.binding.runId; + const sourceRecovery = !!authorityRunId && requirements.identity.runId !== authorityRunId; + if (sourceRecovery && (input.codexSourceRecoveryVersion !== 1 || !input.resumeFrom || !input.priorIdentity || !input.accountIdentity || + input.harness !== "codex" || input.adoptIdentity || Object.entries(input.priorIdentity).some(([key, value]) => + requirements.identity[key as keyof DurableRecoveryIdentity] !== value))) { + throw new Error("Plugin Runner source reconciliation unavailable"); + } if (input.githubBroker) { const description = await call({ kind: "describe" }); if (description.kind !== "description" || description.githubCredentialBrokerVersion !== 1) { @@ -128,7 +139,8 @@ export function createPluginRunnerTaskOwner(input: { const result = await call(environmentTaskOperationSchema.parse({ kind: "prepare", requirements, ...(input.githubBroker ? { githubBroker: input.githubBroker } : {}), projectIds: [...input.projectIds], ...(recovery ? { recovery } : {}), - ...(input.resumeFrom ? { resumeFrom: input.resumeFrom } : {}), + ...(input.resumeFrom && (!authorityRunId || !launchIdentity || launchIdentity.runId === authorityRunId || sourceRecovery) + ? { resumeFrom: input.resumeFrom } : {}), ...(input.accountIdentity ? { accountIdentity: input.accountIdentity } : {}), ...(launched && preparedIdentity && requirements.identity.runId !== preparedIdentity.runId ? { priorIdentity: preparedIdentity } : {}), @@ -202,6 +214,7 @@ export function createPluginRunnerTaskOwner(input: { reconnectGraceMs: native.reconnectGraceMs, lifecyclePolicy: native.lifecyclePolicy, }, }); + launchIdentity = native.identity; launched = true; const child: RunnerProcessHandle["child"] = { exitCode: null, signalCode: null, kill: () => { @@ -320,6 +333,8 @@ export function createPluginRunnerTaskOwner(input: { }; }; const driver: RunnerExecutionDriver = { + get authorityRunId() { return input.authorityRunId ?? input.executionContext?.execution.binding.runId; }, + get codexSourceRecoveryVersion() { return input.codexSourceRecoveryVersion; }, prepareLaunch, authorizeRunnerAdoption, readRunnerState, runnerProcessLauncher: launcher, // The provider resolves the authorized retained task before each state operation. prepareExternalRunnerState: async () => {}, diff --git a/server/src/services/native-runtime/plugin-task-source-recovery.integration.test.ts b/server/src/services/native-runtime/plugin-task-source-recovery.integration.test.ts new file mode 100644 index 00000000000..1d56b4ca5af --- /dev/null +++ b/server/src/services/native-runtime/plugin-task-source-recovery.integration.test.ts @@ -0,0 +1,337 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, companies, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations, + nativeRunResults, nativeRunnerCheckpoints, plugins } from "@paperclipai/db"; +import { DurablePrpControlPlane, type DurablePrpCheckpoint, type NativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; +import { recoveryPostgresSupport, startRecoveryPostgres } from "../../__tests__/helpers/recovery-postgres.js"; +import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; +import { executeEnvironmentTask, resolvePluginTaskResume } from "../environment-task-runtime.js"; +import type { PluginWorkerManager } from "../plugin-worker-manager.js"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { NativeExecutionContext } from "./native-execution-context.js"; +import { buildNativeExecutionInput } from "./native-execution-input.js"; +import { currentNativeControllerIdentity } from "./native-restart-recovery.js"; +import { loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; +import { claimPluginTaskSourceRecovery } from "./plugin-task-source-recovery.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; + +describe.skipIf(!recoveryPostgresSupport.supported)("stopped source admission under the current native owner", () => { + let database: Awaited>, db: ReturnType; + let home: ReturnType, root: string; + const originalHome = process.env.PAPERCLIP_HOME; + const contexts: NativeExecutionContext[] = []; + beforeAll(async () => { + home = createIsolatedTestHome(); process.env.PAPERCLIP_HOME = home.home; + root = await mkdtemp(join(tmpdir(), "plugin-source-recovery-")); + database = await startRecoveryPostgres("plugin-source-recovery-"); db = createDb(database.connectionString); + }, 60_000); + afterEach(() => { for (const context of contexts.splice(0)) context.finish(); }); + afterAll(async () => { + await database?.cleanup(); if (root) await rm(root, { recursive: true, force: true }); home?.cleanup(); + if (originalHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = originalHome; + }); + + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), sourceRunId = randomUUID(), runId = randomUUID(); + const sourceLeaseId = randomUUID(), leaseId = randomUUID(), environmentId = randomUUID(), pluginId = randomUUID(); + const controller = await currentNativeControllerIdentity(), oldOwner = { bootId: controller.bootId, generation: 1, leaseOwner: randomUUID() }; + await db.insert(companies).values({ id: companyId, name: "Source admission", issuePrefix: `S${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Source worker", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue once", status: "in_progress", assigneeAgentId: agentId }); + const [sourceRun] = await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId, nativeIssueId: issueId, + status: "running", runtimeMode: "native" }).returning(); + const native = await prepareNativeHeartbeatRun({ db, run: sourceRun, issue: { id: issueId, title: "Continue once", description: null }, environmentLeaseId: sourceLeaseId }); + const [prepared] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, sourceRunId)); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.id, prepared.completionContractId!)); + const build = (id: string) => buildNativeExecutionInput({ companyId, agentId, runId: id, normalizedSessionId: native.normalizedSessionId, + provider: "codex", issue: { id: issueId, identifier: "SOURCE-1", title: "Continue once", description: null, workMode: "standard" }, + taskPrompt: "Continue once", workspace: { id, cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + completionContract: { id: prepared.completionContractId!, sha256: prepared.completionContractSha256!, schemaVersion: "paperclip.run-result.v1", contract: contract.contractJson as never }, + runtimeContext: nativeRuntimeContextFixture() }); + const sourceExecution = build(sourceRunId), execution = build(runId); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...prepared.runnerProfileJson, nativeExecutionInput: sourceExecution } }).where(eq(heartbeatRuns.id, sourceRunId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: sourceRunId, phase: "observed", controllerBootId: oldOwner.bootId, + controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: oldOwner.generation, + leaseOwner: oldOwner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }); + const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: sourceLeaseId, runId: sourceRunId, + normalizedSessionId: native.normalizedSessionId, turnId: `turn-${sourceRunId}`, itemId: `item-${sourceRunId}` }; + const oldWriter = nativeRunnerCheckpointWriter(db, { companyId, issueId, runId: sourceRunId, authorityRunId: sourceRunId, ...oldOwner }); + let checkpoint!: DurablePrpCheckpoint; + const core = new DurablePrpControlPlane({ stateDirectory: join(root, sourceRunId), identity, + expectedRunnerVersion: "source-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + core.setCheckpointPersistence(async value => { checkpoint = value; await oldWriter(value); }); + await core.flushCheckpoint(); await core.stop(); + const [result] = await db.insert(nativeRunResults).values({ companyId, issueId, runId: sourceRunId, + completionContractId: prepared.completionContractId!, serverFingerprint: "accepted-source", schemaStatus: "accepted", + resultJson: { summary: "Accepted once" }, canonicalSha256: "accepted-source" }).returning(); + await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, sourceRunId)); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(Date.now() - 1_000), resultJson: { summary: "Accepted once" } }).where(eq(heartbeatRuns.id, sourceRunId)); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, status: "running", runtimeMode: "native", + nativeSessionId: native.normalizedSessionId, runnerProfileJson: { nativeExecutionInput: execution } }); + const owner = { bootId: controller.bootId, generation: 3, leaseOwner: randomUUID() }; + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", controllerBootId: owner.bootId, + controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: owner.generation, + leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "source-fixture", version: "1.0.0", status: "ready", + manifestJson: { capabilities: ["environment.drivers.register"], environmentDrivers: [{ driverKey: "tasks", supportsTasks: true }] } as never }); + await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "plugin", status: "active", + config: { pluginKey: pluginId, driverKey: "tasks" } }); + const metadata = { driver: "plugin", pluginId, driverKey: "tasks", runnerTaskLease: true }; + const sourceLease = { id: sourceLeaseId, companyId, heartbeatRunId: sourceRunId, provider: `plugin:${pluginId}:tasks`, providerLeaseId: "source-task" }; + await db.insert(environmentLeases).values({ ...sourceLease, environmentId, issueId, leasePolicy: "reuse_by_environment", status: "released", + releasedAt: new Date(), cleanupStatus: "success", metadata: { ...metadata, runnerTaskIdentity: { runnerId: identity.runnerInstanceId, + leaseId: identity.environmentLeaseId, runId: sourceRunId, sessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: identity.itemId }, + remoteExecutionTermination: remoteTerminationReceipt(sourceLease, { providerLeaseId: "source-task", state: "stopped" }) } }); + await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, issueId, heartbeatRunId: runId, + provider: sourceLease.provider, providerLeaseId: "replacement-task", leasePolicy: "reuse_by_environment", status: "active", metadata }); + const context = new NativeExecutionContext(execution, 60_000); context.bindNativeLease(owner); contexts.push(context); + const input = { companyId, leaseId, context, identity, resumeFrom: "source-task", selectedSourceRecoveryVersion: 1 as const, + accountIdentity: "current-selected-account", matchesExecution: (previous: typeof sourceExecution) => previous.provider.kind === execution.provider.kind && previous.workspace.cwd === execution.workspace.cwd }; + const workers = { getWorker: () => ({ supportedMethods: ["environmentTask"] }), call: vi.fn(async (_id: string, _method: string, p: any) => { + if (p.operation.kind === "prepare") return { kind: "prepared", taskId: p.taskId, + launch: { configurationId: "source-configuration", requirementsDigest: `sha256:${"a".repeat(64)}`, provider: {} } }; + if (p.operation.kind === "check_project_auth") return { kind: "project_auth", taskId: p.taskId, provider: "openai", configurationId: p.operation.configurationId, allowed: true }; + return { kind: "accepted", taskId: p.taskId }; + }) }; + const operation = { kind: "prepare" as const, resumeFrom: input.resumeFrom, + accountIdentity: input.accountIdentity, projectIds: [], requirements: { schema: "paperclip.runner.launch-requirements.v1" as const, + identity, provider: "codex" as const, model: null, instructions: "Source policy", runtimeContext: null, settings: {}, approvalPolicy: "never" as const, + readOnly: false, preparedConversation: false, includeCollaborationModeInstructions: false } }; + const task = (op: Parameters[2]["operation"] = operation) => executeEnvironmentTask(db, workers as unknown as PluginWorkerManager, + { companyId, leaseId, executionContext: context, operation: op }); + return { ...input, runId, sourceRunId, sourceLeaseId, environmentId, agentId, issueId, owner, oldOwner, oldWriter, checkpoint, execution, operation, workers, task }; + } + + it("authorizes only the claimed predecessor identity while preserving A's accepted outcome", async () => { + const f = await fixture(); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + await expect(f.task()).rejects.toThrow("execution identity mismatch"); + await claimPluginTaskSourceRecovery(db, f); + const attachment = { ...f.operation, resumeFrom: undefined, priorIdentity: f.identity, + requirements: { ...f.operation.requirements, identity: { ...f.identity, runId: f.runId, turnId: `turn-${f.runId}`, itemId: `item-${f.runId}` } } }; + await expect(f.task(attachment)).rejects.toThrow("ownership_unverified"); + expect(f.workers.call).not.toHaveBeenCalled(); + await expect(f.task()).resolves.toMatchObject({ kind: "prepared" }); + expect(f.workers.call.mock.calls[0][2]).toMatchObject({ runId: f.runId, operation: { requirements: { identity: f.identity } } }); + await f.task({ kind: "check_project_auth", configurationId: "source-configuration", provider: "openai", projectIds: [] }); + await expect(f.task({ kind: "submit", resumeFrom: f.resumeFrom, + accountIdentity: f.accountIdentity, preparedLaunchId: "source-configuration", projectIds: [], bootstrapTicket: "synthetic-test-ticket", + runner: { runnerId: f.identity.runnerInstanceId, leaseId: f.identity.environmentLeaseId, runId: f.sourceRunId, + sessionId: f.identity.normalizedSessionId, turnId: f.identity.turnId, itemId: f.identity.itemId, harness: "codex", protocolMin: 1, protocolMax: 1 }, + launch: { runnerVersion: "source-fixture", runnerDigest: `sha256:${"a".repeat(64)}`, + maxOutboxBytes: 262144, p0ReserveBytes: 65536, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } }, + })).resolves.toMatchObject({ kind: "accepted" }); + expect(f.workers.call.mock.calls.at(-1)![2]).toMatchObject({ runId: f.runId, operation: { + kind: "submit", runner: { runId: f.sourceRunId }, + } }); + for (const key of ["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId"]) { + await expect(f.task({ ...attachment, requirements: { ...attachment.requirements, + identity: { ...attachment.requirements.identity, [key]: randomUUID() } } })).rejects.toThrow("ownership_unverified"); + } + await expect(f.task({ ...attachment, accountIdentity: "different-account" })).rejects.toThrow("ownership_unverified"); + await expect(f.task(attachment)).resolves.toMatchObject({ kind: "prepared" }); + await expect(f.task(attachment)).resolves.toMatchObject({ kind: "prepared" }); + expect(f.workers.call.mock.calls.at(-1)![2]).toMatchObject({ runId: f.runId, operation: { + kind: "prepare", priorIdentity: f.identity, requirements: { identity: { runId: f.runId } }, + } }); + await expect(f.task({ ...f.operation, resumeFrom: undefined })).rejects.toThrow("ownership_unverified"); + const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(after).toEqual(before); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + expect(JSON.stringify(lease.metadata)).not.toContain(f.owner.leaseOwner); + }); + + it("keeps naked mismatches ordinary and malformed claims fail closed", async () => { + const f = await fixture(); + for (const claim of [undefined, false, {}, true, "untrusted"]) { + if (claim !== undefined) await db.update(environmentLeases).set({ metadata: + sql`${environmentLeases.metadata} || ${JSON.stringify({ runnerTaskSourceRecovery: claim })}::jsonb` }).where(eq(environmentLeases.id, f.leaseId)); + await expect(f.task()).rejects.toThrow(claim ? "ownership_unverified" : "execution identity mismatch"); + } + expect(f.workers.call).not.toHaveBeenCalled(); + }); + + it("requires a complete claimed launch before any predecessor RPC", async () => { + const f = await fixture(); await claimPluginTaskSourceRecovery(db, f); + for (const operation of [ + { ...f.operation, resumeFrom: undefined }, { ...f.operation, accountIdentity: undefined }, + { ...f.operation, priorIdentity: f.identity }, + { ...f.operation, requirements: { ...f.operation.requirements, provider: "acpx" as const } }, + ]) await expect(f.task(operation)).rejects.toThrow("ownership_unverified"); + const submit = { kind: "submit" as const, resumeFrom: f.resumeFrom, accountIdentity: f.accountIdentity, + preparedLaunchId: "source-configuration", projectIds: [], bootstrapTicket: "synthetic-test-ticket", + runner: { runnerId: f.identity.runnerInstanceId, leaseId: f.identity.environmentLeaseId, runId: f.sourceRunId, + sessionId: f.identity.normalizedSessionId, turnId: f.identity.turnId, itemId: f.identity.itemId, harness: "codex", protocolMin: 1, protocolMax: 1 }, + launch: { runnerVersion: "source-fixture", runnerDigest: `sha256:${"a".repeat(64)}`, + maxOutboxBytes: 262144, p0ReserveBytes: 65536, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } }; + for (const operation of [ + { ...submit, resumeFrom: undefined }, { ...submit, accountIdentity: undefined }, + { ...submit, preparedLaunchId: undefined }, { ...submit, launch: undefined }, + { ...submit, runner: { ...submit.runner, harness: "claude" } }, + ]) await expect(f.task(operation)).rejects.toThrow("ownership_unverified"); + expect(f.workers.call).not.toHaveBeenCalled(); + }); + + it.each(["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"] as const)("rejects a different source %s", async key => { + const f = await fixture(); + await expect(claimPluginTaskSourceRecovery(db, { ...f, identity: { ...f.identity, [key]: randomUUID() } })).rejects.toThrow(); + expect(f.workers.call).not.toHaveBeenCalled(); + }); + + it.each(["before-claim", "after-claim"])("rejects a stale same-boot coordinator context: %s", async point => { + const f = await fixture(); if (point === "after-claim") await claimPluginTaskSourceRecovery(db, f); + await db.update(nativeRunFinalizations).set({ controllerGeneration: f.owner.generation + 1, leaseOwner: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (point === "before-claim") await expect(claimPluginTaskSourceRecovery(db, f)).rejects.toThrow("context_owner_changed"); + else await expect(f.task()).rejects.toThrow("context_owner_changed"); + expect(f.workers.call).not.toHaveBeenCalled(); + }); + + it("rejects a worker reply after the same controller has lost its acquired generation", async () => { + const f = await fixture(); await claimPluginTaskSourceRecovery(db, f); + const perform = f.workers.call.getMockImplementation()!; + f.workers.call.mockImplementation(async (...args) => { + const result = await perform(...args); + await db.update(nativeRunFinalizations).set({ controllerGeneration: f.owner.generation + 1, leaseOwner: randomUUID() }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + return result; + }); + await expect(f.task()).rejects.toThrow("reconcile the same task"); + await expect(f.task()).rejects.toThrow("context_owner_changed"); + expect(f.workers.call).toHaveBeenCalledTimes(1); + }); + + it.each(["capability", "expired", "paused", "source-live", "driver", "header", "source-profile", "current-profile", "workspace-export", "legacy-settlement"])("holds an ineligible recovery: %s", async mode => { + const f = await fixture(); + if (mode === "expired") await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + if (mode === "source-live") await db.update(environmentLeases).set({ cleanupStatus: null }).where(eq(environmentLeases.id, f.sourceLeaseId)); + if (mode === "driver") await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || '{"driverKey":"other"}'::jsonb` }).where(eq(environmentLeases.id, f.sourceLeaseId)); + if (mode === "header") await db.update(nativeRunnerCheckpoints).set({ material: {} }).where(eq(nativeRunnerCheckpoints.runId, f.sourceRunId)); + if (mode === "source-profile" || mode === "current-profile") await db.update(heartbeatRuns).set({ runnerProfileJson: + sql`jsonb_set(${heartbeatRuns.runnerProfileJson}, '{nativeExecutionInput,binding,issueId}', ${JSON.stringify(randomUUID())}::jsonb)` }) + .where(eq(heartbeatRuns.id, mode === "source-profile" ? f.sourceRunId : f.runId)); + if (mode === "workspace-export") await db.update(heartbeatRuns).set({ runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} || ${JSON.stringify({ nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v3", state: "prepared", leaseId: f.sourceLeaseId, providerLeaseId: f.resumeFrom, + workspaceId: f.issueId, remoteCwd: "/workspace", resourceDisposition: "keep_running", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + } })}::jsonb` }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (mode === "legacy-settlement") await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({ legacyShutdownWorkspaceSettlement: { + schema: "paperclip.legacy-shutdown-workspace-settlement.v1", state: "pending", runId: f.sourceRunId, deadlineAt: new Date(Date.now() + 60_000).toISOString(), + } })}::jsonb` }).where(eq(environmentLeases.id, f.sourceLeaseId)); + await expect(claimPluginTaskSourceRecovery(db, { ...f, ...(mode === "capability" ? { selectedSourceRecoveryVersion: undefined } : {}) })).rejects.toThrow(); + const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.sourceLeaseId)); + expect(source.metadata?.runnerTaskSourceConsumer).toBeUndefined(); expect(f.workers.call).not.toHaveBeenCalled(); + }); + + it("pins the consumer and current account across retries", async () => { + const f = await fixture(), claim = await claimPluginTaskSourceRecovery(db, f); + expect(await claimPluginTaskSourceRecovery(db, f)).toEqual(claim); + await expect(claimPluginTaskSourceRecovery(db, { ...f, accountIdentity: "different-account" })).rejects.toThrow(); + await expect(f.task({ ...f.operation, accountIdentity: "different-account" })).rejects.toThrow(); + const competingLease = randomUUID(); + const [original] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + await db.insert(environmentLeases).values({ ...original, id: competingLease, providerLeaseId: "other-replacement" }); + await expect(claimPluginTaskSourceRecovery(db, { ...f, leaseId: competingLease })).rejects.toThrow(); + await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || '{"runnerTaskSessionRetained":true}'::jsonb` }) + .where(eq(environmentLeases.id, f.sourceLeaseId)); + await expect(resolvePluginTaskResume(db, { ...f, priorIdentity: f.identity })).resolves.toBe(f.resumeFrom); + await expect(resolvePluginTaskResume(db, { ...f, leaseId: competingLease, priorIdentity: f.identity })).rejects.toThrow("not authorized"); + }); + + it("keeps the first source claim after dispatch, even when another source is eligible", async () => { + const f = await fixture(); + const first = await claimPluginTaskSourceRecovery(db, f); + await db.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || '{"runnerTaskDispatched":true,"runnerTaskResumeFrom":"source-task"}'::jsonb` }) + .where(eq(environmentLeases.id, f.leaseId)); + const alternateRunId = randomUUID(), alternateLeaseId = randomUUID(), alternateTaskId = "alternate-task"; + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + const original = source.runnerProfileJson!.nativeExecutionInput as NativeExecutionInput; + const execution = { ...original, binding: { ...original.binding, runId: alternateRunId } }; + await db.insert(heartbeatRuns).values({ ...source, id: alternateRunId, runnerProfileJson: { ...source.runnerProfileJson, nativeExecutionInput: execution } }); + const [result] = await db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: alternateRunId, + completionContractId: source.completionContractId!, serverFingerprint: "accepted-alternate", schemaStatus: "accepted", + resultJson: { summary: "Accepted alternate" }, canonicalSha256: "accepted-alternate" }).returning(); + const [priorOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.insert(nativeRunFinalizations).values({ ...priorOwner, runId: alternateRunId, resultId: null, + phase: "observed", leaseOwner: f.oldOwner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }); + const identity = { ...f.identity, runId: alternateRunId, turnId: `turn-${alternateRunId}`, itemId: `item-${alternateRunId}` }; + const core = new DurablePrpControlPlane({ stateDirectory: join(root, alternateRunId), identity, + expectedRunnerVersion: "source-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, { companyId: f.companyId, issueId: f.issueId, + runId: alternateRunId, authorityRunId: alternateRunId, ...f.oldOwner })); + try { await core.flushCheckpoint(); } finally { await core.stop(); } + await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, alternateRunId)); + const [sourceLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.sourceLeaseId)); + const alternateLease = { ...sourceLease, id: alternateLeaseId, heartbeatRunId: alternateRunId, providerLeaseId: alternateTaskId }; + await db.insert(environmentLeases).values({ ...alternateLease, metadata: { ...sourceLease.metadata, + runnerTaskSourceConsumer: undefined, runnerTaskIdentity: { runnerId: identity.runnerInstanceId, leaseId: identity.environmentLeaseId, + runId: identity.runId, sessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: identity.itemId }, + remoteExecutionTermination: remoteTerminationReceipt(alternateLease, { providerLeaseId: alternateTaskId, state: "stopped" }) } }); + expect(await resolvePluginTaskResume(db, { ...f, priorIdentity: identity, inspectCheckpoint: true })).toBe(alternateTaskId); + await expect(claimPluginTaskSourceRecovery(db, { ...f, identity, resumeFrom: alternateTaskId })).rejects.toThrow("ownership_unverified"); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + expect(lease.metadata?.runnerTaskSourceRecovery).toEqual(first); + const [unconsumed] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, alternateLeaseId)); + expect(unconsumed.metadata?.runnerTaskSourceConsumer).toBeUndefined(); + await expect(claimPluginTaskSourceRecovery(db, f)).resolves.toEqual(first); + }); + + it("refreshes only the same source under a newly acquired coordinator epoch", async () => { + const f = await fixture(); await claimPluginTaskSourceRecovery(db, f); + const owner = { ...f.owner, generation: f.owner.generation + 1, leaseOwner: randomUUID() }; + await db.update(nativeRunFinalizations).set({ controllerGeneration: owner.generation, leaseOwner: owner.leaseOwner }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const context = new NativeExecutionContext(f.execution, 60_000); context.bindNativeLease(owner); contexts.push(context); + const refreshed = await claimPluginTaskSourceRecovery(db, { ...f, context }); + expect(refreshed).toMatchObject({ identity: f.identity, resumeFrom: f.resumeFrom, controllerGeneration: owner.generation, contextId: context.id }); + await expect(f.task()).rejects.toThrow("context_owner_changed"); + }); + + it("allows the last A write before consumption and rejects stale A writes afterward", async () => { + const f = await fixture(); + await expect(f.oldWriter(f.checkpoint)).resolves.toBeUndefined(); + await claimPluginTaskSourceRecovery(db, f); + await expect(f.oldWriter(f.checkpoint)).rejects.toThrow("checkpoint_owner_lost"); + const writer = nativeRunnerCheckpointWriter(db, { companyId: f.companyId, issueId: f.issueId, runId: f.sourceRunId, + authorityRunId: f.runId, ...f.owner }); + await expect(writer(f.checkpoint)).resolves.toBeUndefined(); + expect((await loadNativeRunnerCheckpoint(db, { companyId: f.companyId, issueId: f.issueId, runId: f.sourceRunId }))?.runId).toBe(f.sourceRunId); + await expect(writer({ ...f.checkpoint, records: new Map([["header", "{}"]]) })).rejects.toThrow("checkpoint_owner_lost"); + await db.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, '{runnerTaskSourceRecovery,controllerGeneration}', '99'::jsonb)` }).where(eq(environmentLeases.id, f.leaseId)); + await expect(writer(f.checkpoint)).rejects.toThrow("checkpoint_owner_lost"); + }); + + it.each(["writer", "claim"] as const)("serializes a concurrent source writer and claim when %s commits first", async first => { + const f = await fixture(); + let pending: Promise<{ error?: unknown }> | undefined; + try { + await db.transaction(async tx => { + if (first === "writer") { + await nativeRunnerCheckpointWriter(tx as unknown as typeof db, { companyId: f.companyId, issueId: f.issueId, + runId: f.sourceRunId, authorityRunId: f.sourceRunId, ...f.oldOwner })(f.checkpoint); + } else await claimPluginTaskSourceRecovery(tx as unknown as typeof db, f); + const [backend] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`); + pending = (first === "writer" ? claimPluginTaskSourceRecovery(db, f) : f.oldWriter(f.checkpoint)) + .then(() => ({}), error => ({ error })); + await vi.waitFor(async () => { + const [waiting] = await db.execute<{ blocked: boolean }>(sql`select exists ( + select 1 from pg_stat_activity where datname = current_database() + and ${backend.pid} = any(pg_blocking_pids(pid))) as blocked`); + expect(waiting.blocked).toBe(true); + }); + }); + } finally { + const result = await pending; + if (first === "writer") expect(result).toEqual({}); + else expect(result?.error).toMatchObject({ message: "native_runner_checkpoint_owner_lost" }); + } + const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.sourceLeaseId)); + expect(source.metadata?.runnerTaskSourceConsumer).toMatchObject({ leaseId: f.leaseId, runId: f.runId }); + }); +}); diff --git a/server/src/services/native-runtime/plugin-task-source-recovery.ts b/server/src/services/native-runtime/plugin-task-source-recovery.ts new file mode 100644 index 00000000000..2b8f45b83f6 --- /dev/null +++ b/server/src/services/native-runtime/plugin-task-source-recovery.ts @@ -0,0 +1,198 @@ +import { and, eq, inArray, sql } from "drizzle-orm"; +import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentTaskOperationSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; +import { parseNativeExecutionInput, inspectDurablePrpCheckpointHeader, type NativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; +import { resolvePluginTaskResume } from "../plugin-task-resume.js"; +import { legacyShutdownWorkspaceResourceProtected } from "../legacy-shutdown-workspace-settlement.js"; +import { NativeExecutionContext } from "./native-execution-context.js"; +import { isCurrentNativeControllerProcess } from "./native-restart-recovery.js"; +import { nativeWorkspaceExportPendingForResource } from "./native-workspace-finalization-state.js"; +import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js"; +import { loadNativeRunnerCheckpointHeader } from "./native-runner-checkpoint.js"; +import { nativeSha256 as fingerprint } from "./canonical.js"; + +type Identity = Extract["requirements"]["identity"]; +type Transaction = Parameters[0]>[0]; +const keys = ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"] as const; +const record = (value: unknown): Record => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +const deny = (): never => { throw new Error("native_source_recovery_ownership_unverified"); }; + +export type PluginTaskSourceRecovery = { + schema: "paperclip.runner.source-recovery.v1"; + resumeFrom: string; + sourceLeaseId: string; + identity: Identity; + consumerLeaseId: string; + consumerRunId: string; + controllerBootId: string; + controllerGeneration: number; + contextId: string; + accountIdentity: string; + sourceProfileSha256: string; + consumerProfileSha256: string; +}; + +function readRecovery(value: unknown): PluginTaskSourceRecovery | null { + const v = record(value); + const identity = environmentTaskOperationSchema.safeParse({ kind: "read_state", identity: v.identity }); + return v.schema === "paperclip.runner.source-recovery.v1" && identity.success && + ["resumeFrom", "sourceLeaseId", "consumerLeaseId", "consumerRunId", "controllerBootId", "contextId", "accountIdentity"].every(key => typeof v[key] === "string" && v[key] !== "") && + Number.isSafeInteger(v.controllerGeneration) && Number(v.controllerGeneration) >= 0 && + [v.sourceProfileSha256, v.consumerProfileSha256].every(hash => typeof hash === "string" && /^[a-f0-9]{64}$/.test(hash)) + ? v as unknown as PluginTaskSourceRecovery : null; +} + +function pinnedSource(recovery: PluginTaskSourceRecovery) { + const { controllerBootId: _boot, controllerGeneration: _generation, contextId: _context, ...source } = recovery; + return source; +} + +async function inspect(tx: Transaction, input: { + companyId: string; leaseId: string; context: NativeExecutionContext; identity: Identity; resumeFrom: string; accountIdentity: string; +}, expected?: PluginTaskSourceRecovery) { + input.context.assertOpen(); + const binding = input.context.execution.binding; + if (binding.companyId !== input.companyId || input.identity.runId === binding.runId) return deny(); + await tx.execute(sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`); + const owners = await lockSourceRecoveryCoordinators(tx, input.companyId, [input.identity.runId, binding.runId]); + const owner = owners.find(value => value.runId === binding.runId), prior = owners.find(value => value.runId === input.identity.runId); + // NOWAIT preserves the established coordinator-first order without waiting + // behind a workspace-maintenance path which already holds the issue. + const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, binding.issueId))).for("update", { noWait: true }); + const runs = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), + inArray(heartbeatRuns.id, [input.identity.runId, binding.runId]))); + const current = runs.find(value => value.id === binding.runId), source = runs.find(value => value.id === input.identity.runId); + const [lease] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, input.leaseId))).for("update"); + const [agent] = await tx.select().from(agents).where(and(eq(agents.companyId, input.companyId), eq(agents.id, binding.agentId))); + const [company] = await tx.select().from(companies).where(eq(companies.id, input.companyId)); + if (!owner || !owner.leaseOwner || !owner.leaseExpiresAt || owner.leaseExpiresAt.getTime() <= Date.now() || + !isCurrentNativeControllerProcess(owner) || owner.resultId || ["committed", "applied", "terminal_failure"].includes(owner.phase) || + !prior || prior.phase !== "committed" || !prior.resultId || prior.leaseOwner || + !current || current.status !== "running" || current.nativeIssueId !== binding.issueId || current.agentId !== binding.agentId || current.resultJson?.nativeCancellation || + !source || source.status !== "succeeded" || !source.finishedAt || source.runtimeMode !== "native" || source.agentId !== binding.agentId || source.nativeIssueId !== binding.issueId || + source.runnerInstanceId !== input.identity.runnerInstanceId || source.nativeSessionId !== input.identity.normalizedSessionId || + !issue || issue.executionRunId !== binding.runId || issue.assigneeAgentId !== binding.agentId || issue.status !== "in_progress" || + company?.status !== "active" || agent?.lifecycleState !== "ready" || ["paused", "terminated", "pending_approval"].includes(agent.status) || + !lease || lease.heartbeatRunId !== binding.runId || lease.issueId !== binding.issueId || lease.status !== "active" || lease.releasedAt || + lease.expiresAt && lease.expiresAt.getTime() <= Date.now() || lease.metadata?.runnerTaskRetiring === true) return deny(); + input.context.assertNativeLease({ bootId: owner.controllerBootId!, leaseOwner: owner.leaseOwner, generation: owner.controllerGeneration }); + const header = await loadNativeRunnerCheckpointHeader(tx, { companyId: input.companyId, issueId: binding.issueId, runId: source.id }); + const protocolIdentity = header ? inspectDurablePrpCheckpointHeader(header) : null; + if (!protocolIdentity || keys.some(key => protocolIdentity[key] !== input.identity[key])) return deny(); + const sourceExecution = parseNativeExecutionInput(source.runnerProfileJson?.nativeExecutionInput); + const consumerExecution = parseNativeExecutionInput(current.runnerProfileJson?.nativeExecutionInput); + if (sourceExecution.provider.kind !== "codex" || consumerExecution.provider.kind !== "codex" || + sourceExecution.binding.companyId !== input.companyId || sourceExecution.binding.issueId !== binding.issueId || + sourceExecution.binding.agentId !== binding.agentId || sourceExecution.binding.runId !== source.id || + sourceExecution.session.normalizedSessionId !== input.identity.normalizedSessionId || + fingerprint(consumerExecution) !== fingerprint(input.context.execution)) return deny(); + const sourceTask = await resolvePluginTaskResume(tx as unknown as Db, { companyId: input.companyId, leaseId: input.leaseId, + priorIdentity: input.identity, inspectCheckpoint: true }); + if (sourceTask !== input.resumeFrom) return deny(); + const sources = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), + eq(environmentLeases.heartbeatRunId, source.id), eq(environmentLeases.providerLeaseId, sourceTask))).for("update"); + if (sources.length !== 1) return deny(); + const sourceLease = sources[0]!; + const saved = record(sourceLease.metadata?.runnerTaskIdentity); + const aliases = { runnerInstanceId: "runnerId", environmentLeaseId: "leaseId", runId: "runId", normalizedSessionId: "sessionId", turnId: "turnId", itemId: "itemId" } as const; + if (keys.some(key => saved[aliases[key]] !== input.identity[key]) || + await nativeWorkspaceExportPendingForResource(tx as unknown as Db, sourceLease, binding.runId) || + await legacyShutdownWorkspaceResourceProtected(tx as unknown as Db, sourceLease)) return deny(); + const allocations = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.provider, sourceLease.provider!), + eq(environmentLeases.providerLeaseId, sourceTask))); + if (allocations.some(value => value.companyId !== input.companyId || + value.id !== sourceLease.id && value.id !== lease.id && ["active", "retained", "pending_cleanup"].includes(value.status))) return deny(); + const recovery: PluginTaskSourceRecovery = { schema: "paperclip.runner.source-recovery.v1", identity: input.identity, + resumeFrom: sourceTask, sourceLeaseId: sourceLease.id, consumerLeaseId: lease.id, consumerRunId: binding.runId, + controllerBootId: owner.controllerBootId!, controllerGeneration: owner.controllerGeneration, contextId: input.context.id, accountIdentity: input.accountIdentity, + sourceProfileSha256: fingerprint(source.runnerProfileJson?.nativeExecutionInput), consumerProfileSha256: fingerprint(current.runnerProfileJson?.nativeExecutionInput) }; + if (expected && fingerprint(recovery) !== fingerprint(expected)) return deny(); + input.context.assertOpen(); + if (owner.leaseExpiresAt.getTime() <= Date.now() || lease.expiresAt && lease.expiresAt.getTime() <= Date.now()) return deny(); + return { recovery, sourceLease, lease, sourceExecution, consumerExecution, owner }; +} + +/** Only current B's native authority can consume succeeded A for maintenance. */ +export async function claimPluginTaskSourceRecovery(db: Db, input: { + companyId: string; leaseId: string; context: NativeExecutionContext; identity: Identity; resumeFrom: string; accountIdentity: string; + selectedSourceRecoveryVersion?: 1; + matchesExecution(source: NativeExecutionInput): boolean; +}): Promise { + if (input.selectedSourceRecoveryVersion !== 1 || !input.accountIdentity) return deny(); + return db.transaction(async tx => { + const inspected = await inspect(tx, input); + if (!input.matchesExecution(inspected.sourceExecution)) return deny(); + const { recovery, sourceLease, lease } = inspected; + if (Object.hasOwn(lease.metadata ?? {}, "runnerTaskSourceRecovery")) { + const saved = readRecovery(lease.metadata?.runnerTaskSourceRecovery); + if (!saved || fingerprint(pinnedSource(saved)) !== fingerprint(pinnedSource(recovery))) return deny(); + } + const consumer = { leaseId: lease.id, runId: recovery.consumerRunId, resumeFrom: recovery.resumeFrom, + identity: recovery.identity, accountIdentity: recovery.accountIdentity }; + const existing = sourceLease.metadata?.runnerTaskSourceConsumer; + if (existing && fingerprint(existing) !== fingerprint(consumer)) return deny(); + const others = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.provider, sourceLease.provider!), + sql`${environmentLeases.id} <> ${lease.id}`, + sql`${environmentLeases.metadata}->>'runnerTaskResumeFrom' = ${input.resumeFrom}`)); + if (others.length) return deny(); + input.context.assertOpen(); + if (inspected.owner.leaseExpiresAt!.getTime() <= Date.now()) return deny(); + await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ runnerTaskSourceConsumer: consumer })}::jsonb` }) + .where(eq(environmentLeases.id, sourceLease.id)); + await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ runnerTaskSourceRecovery: recovery })}::jsonb` }) + .where(eq(environmentLeases.id, lease.id)); + return recovery; + }); +} + +async function assertSourceIdentity(db: Db, input: { + companyId: string; leaseId: string; context: NativeExecutionContext | undefined; + identity: Identity; accountIdentity?: string; resumeFrom?: string; requireDispatched?: boolean; +}): Promise { + if (!input.context) return deny(); + const { identity } = input; + return db.transaction(async tx => { + const [lease] = await tx.select({ metadata: environmentLeases.metadata }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, input.leaseId))); + const expected = readRecovery(lease?.metadata?.runnerTaskSourceRecovery); + if (!expected || keys.some(key => expected.identity[key] !== identity[key]) || + input.resumeFrom !== undefined && expected.resumeFrom !== input.resumeFrom || expected.accountIdentity !== input.accountIdentity) return deny(); + const inspected = await inspect(tx, { ...input, context: input.context!, resumeFrom: expected.resumeFrom, accountIdentity: expected.accountIdentity }, expected); + if (input.requireDispatched && (inspected.lease.metadata?.runnerTaskDispatched !== true || + inspected.lease.metadata?.runnerTaskExecutionStopped === true)) return deny(); + const consumer = record(inspected.sourceLease.metadata?.runnerTaskSourceConsumer); + if (consumer.leaseId !== input.leaseId || consumer.runId !== expected.consumerRunId || consumer.resumeFrom !== expected.resumeFrom || consumer.accountIdentity !== expected.accountIdentity || + fingerprint(consumer.identity) !== fingerprint(identity)) return deny(); + return expected; + }); +} + +/** The worker never derives this exception from an arbitrary mismatched run ID. */ +export async function assertPluginTaskSourceRecovery(db: Db, input: { + companyId: string; leaseId: string; context: NativeExecutionContext | undefined; + operation: Extract; +}): Promise { + const operation = input.operation; + if (!operation.resumeFrom || !operation.accountIdentity || + (operation.kind === "prepare" ? operation.requirements.provider !== "codex" || !!operation.priorIdentity + : operation.runner.harness !== "codex" || !operation.preparedLaunchId || !operation.launch)) return deny(); + const identity = operation.kind === "prepare" ? operation.requirements.identity : { + runnerInstanceId: operation.runner.runnerId, environmentLeaseId: operation.runner.leaseId, runId: operation.runner.runId, + normalizedSessionId: operation.runner.sessionId, turnId: operation.runner.turnId, itemId: operation.runner.itemId, + }; + return assertSourceIdentity(db, { ...input, identity, resumeFrom: operation.resumeFrom, accountIdentity: operation.accountIdentity }); +} + +/** A fresh source incarnation can attach B without pretending it was warm reuse. */ +export async function assertPluginTaskSourceAttachment(db: Db, input: { + companyId: string; leaseId: string; context: NativeExecutionContext | undefined; + operation: Extract; +}): Promise { + const { operation } = input, target = operation.requirements.identity; + if (!operation.priorIdentity || operation.resumeFrom || operation.requirements.provider !== "codex") return deny(); + const recovery = await assertSourceIdentity(db, { ...input, identity: operation.priorIdentity, accountIdentity: operation.accountIdentity, requireDispatched: true }); + if (target.runId !== recovery.consumerRunId || ["runnerInstanceId", "environmentLeaseId", "normalizedSessionId"].some(key => + target[key as keyof Identity] !== recovery.identity[key as keyof Identity])) return deny(); + return recovery; +} diff --git a/server/src/services/plugin-task-resume.ts b/server/src/services/plugin-task-resume.ts new file mode 100644 index 00000000000..d5326f3d0dd --- /dev/null +++ b/server/src/services/plugin-task-resume.ts @@ -0,0 +1,51 @@ +import { and, eq, or } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, type Db } from "@paperclipai/db"; +import { hasRemoteTerminationReceipt, remoteExecutionHasStopped } from "./remote-execution-termination.js"; + +/** A saved provider task can resume only within the same authorized agent workspace. */ +export async function resolvePluginTaskResume(db: Db, input: { + companyId: string; leaseId: string; + priorIdentity: { runnerInstanceId: string; environmentLeaseId: string; runId: string; normalizedSessionId: string }; + /** Read-only discovery; launching still requires an attested retained checkpoint. */ + inspectCheckpoint?: boolean; +}): Promise { + const rows = await db.select({ lease: environmentLeases, run: heartbeatRuns }).from(environmentLeases) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), eq(heartbeatRuns.companyId, input.companyId))) + .where(and(eq(environmentLeases.companyId, input.companyId), + or(eq(environmentLeases.id, input.leaseId), eq(heartbeatRuns.id, input.priorIdentity.runId)))); + const next = rows.find(row => row.lease.id === input.leaseId); + // A retained PRP lease is logical; later provider tasks have different physical lease IDs. + const candidates = rows.filter(row => { + const identity = row.lease.metadata?.runnerTaskIdentity as Record | undefined; + const consumer = row.lease.metadata?.runnerTaskSourceConsumer as { leaseId?: unknown; runId?: unknown; resumeFrom?: unknown } | null | undefined; + return row.lease.id !== input.leaseId && row.run.id === input.priorIdentity.runId && + (!Object.hasOwn(row.lease.metadata ?? {}, "runnerTaskSourceConsumer") || + consumer?.leaseId === input.leaseId && consumer.runId === next?.run.id && consumer.resumeFrom === row.lease.providerLeaseId) && + (input.inspectCheckpoint || row.lease.metadata?.runnerTaskSessionRetained === true) && + identity?.runnerId === input.priorIdentity.runnerInstanceId && identity.leaseId === input.priorIdentity.environmentLeaseId && + identity.sessionId === input.priorIdentity.normalizedSessionId && identity.runId === input.priorIdentity.runId && + !rows.some(successor => successor.lease.id !== input.leaseId && + successor.lease.metadata?.runnerTaskResumeFrom === row.lease.providerLeaseId); + }); + const old = candidates.length === 1 ? candidates[0] : undefined; + const identity = old?.lease.metadata?.runnerTaskIdentity as Record | undefined; + if (!old || !next || !old.lease.providerLeaseId || !identity || + identity.runnerId !== input.priorIdentity.runnerInstanceId || identity.leaseId !== input.priorIdentity.environmentLeaseId || + identity.sessionId !== input.priorIdentity.normalizedSessionId || identity.runId !== input.priorIdentity.runId || + old.lease.metadata?.driver !== "plugin" || next.lease.metadata?.driver !== "plugin" || + !next.lease.providerLeaseId || next.lease.providerLeaseId === old.lease.providerLeaseId || + (old.run.id !== next.run.id && !["succeeded", "interrupted", "failed", "cancelled", "timed_out"].includes(old.run.status)) || + old.run.agentId !== next.run.agentId || old.lease.issueId !== next.lease.issueId || + old.lease.executionWorkspaceId !== next.lease.executionWorkspaceId || + !old.lease.environmentId || old.lease.environmentId !== next.lease.environmentId || + !next.lease.provider?.startsWith("plugin:") || old.lease.provider !== next.lease.provider || + old.lease.metadata?.pluginId !== next.lease.metadata?.pluginId || old.lease.metadata?.driverKey !== next.lease.metadata?.driverKey || + next.lease.status !== "active" || next.lease.releasedAt || next.run.status !== "running" || + (next.lease.expiresAt && next.lease.expiresAt.getTime() <= Date.now()) || + !(old.run.id === next.run.id + ? rows.filter(row => row.lease.id !== next.lease.id).every(row => hasRemoteTerminationReceipt(row.lease)) + : await remoteExecutionHasStopped(db, input.companyId, old.run.id))) { + throw new Error("Plugin Runner retained session is not authorized"); + } + return old.lease.providerLeaseId; +}