diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index d5a3f0b0fb2..f84fbf74e42 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3211,6 +3211,14 @@ ownership checks. The backup transaction checks this authority again after upload, and a late predecessor cannot replace a newer conversation backup. This grants no new lease or permission to start provider work. +The remote backend captures its backup fence while its run still owns the +controller lease. A suspended provider can then publish its backup while an +accepted result is still saving the workspace, before finalization commits. +Warm attachments replace that captured fence with the current run's authority; +an older backend's close callback uses the current attachment's fence. Publication +still revalidates the boot, generation, accepted result, and successor authority +before and after upload. A released, uncommitted run cannot acquire a new fence. + Local durable control-plane state is bounded at 256 MiB in the server, runnerd recovery, and durable control-plane readers. These paths synchronously read and parse the full JSON file, so memory use and parse time grow with file diff --git a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts index ab53c6e1b5a..182f14d03fb 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts @@ -180,6 +180,15 @@ describeEmbeddedPostgres("durable native provider backups", () => { leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); return result; } + async function releaseWhileSavingWorkspace(f: Awaited>) { + const result = await commitWarmRun(f); + await f.db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(heartbeatRuns).set({ status: "running", finishedAt: null }) + .where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + return result; + } async function inactiveFixture() { const [f] = await offlineFixtures(1); await commitWarmRun(f); @@ -338,6 +347,47 @@ describeEmbeddedPostgres("durable native provider backups", () => { expect(rows.filter(row => row.status === "pending")).toHaveLength(1); expect(await readFile(join(f.root, "manifest.json"), "utf8")).toContain("paperclip.native-harness-backup.v1"); }); + it("publishes with the captured provider owner while an accepted result is still saving its workspace", async () => { + const f = await fixture(); + const result = await releaseWhileSavingWorkspace(f); + await expect(nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, bootId: f.owner.bootId })).rejects.toThrow("native_harness_storage_owner_missing"); + await persistNativeHarnessBackup(f); + await removeNativeHarnessBackup(f.home); + const replacement = join(environment.current.runtimeRoot, randomUUID(), "restored-pending-workspace"); + ownedHomes.push(replacement); + expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).toBe(true); + expect(JSON.parse(await readFile(join(replacement, "current", "codex", "sessions", "history.json"), "utf8"))).toEqual(f.identity); + const [coordinator] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator).toMatchObject({ phase: "workspace_finalizing", resultId: result.id, leaseOwner: null, + leaseExpiresAt: null, controllerBootId: f.owner.bootId, controllerGeneration: f.owner.generation }); + }); + it.each(["boot", "generation", "result", "successor"])("rejects a captured provider fence when its %s changes during upload", async change => { + const f = await fixture(); + await persistNativeHarnessBackup(f); + const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); + await releaseWhileSavingWorkspace(f); + const put = f.storage.putFile.bind(f.storage); + vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + const object = await put(input); + if (change === "successor") { + const nextRunId = randomUUID(); + await f.db.insert(heartbeatRuns).values({ id: nextRunId, companyId: f.companyId, + agentId: environment.current.agentId, issueId: f.issueId, status: "running" }); + await f.db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, f.issueId)); + } else { + await f.db.update(nativeRunFinalizations).set(change === "boot" ? { controllerBootId: randomUUID() } + : change === "generation" ? { controllerGeneration: 2 } : { resultId: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + } + return object; + }); + await expect(persistNativeHarnessBackup(f)).rejects.toThrow(change === "successor" + ? "native_harness_storage_owner_superseded" : "native_runner_checkpoint_owner_lost"); + const selected = await f.db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest), + eq(nativeHarnessBackups.status, "current"))); + expect(selected.map(row => row.id)).toEqual([before.id]); + }); it.each(["idle", "preparing", "native"])("preserves a released committed warm session with a %s successor", async state => { const f = await fixture(); const result = await commitWarmRun(f); diff --git a/server/src/services/native-runtime/native-harness-backup-storage.ts b/server/src/services/native-runtime/native-harness-backup-storage.ts index ae7ee97d9af..4d35d2d5482 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -174,6 +174,10 @@ async function fence(tx: Parameters[0]>[0], owner: } } +export async function assertNativeHarnessBackupOwner(db: Db, owner: NativeRunnerCheckpointFenceOwner): Promise { + await db.transaction(tx => fence(tx, owner)); +} + export async function persistNativeHarnessBackup(input: { db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner; scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number; 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 1a6bdf4d5b8..977a426dccb 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -2,6 +2,7 @@ import { activeNativeExecutionRunIds } from "./native-active-runs.js"; import { onRetainedNativeSessionCleanupRelease } from "../retained-native-session-cleanup-notifications.js"; import { readIdleLocalWork, markIdleStartupComplete, markIdleIngressTracked } from "../idle-local-work.js"; import * as runnerCheckpoint from "./native-runner-checkpoint.js"; +import * as harnessStorage from "./native-harness-backup-storage.js"; import * as stagedAttachmentRecovery from "./native-staged-attachment-recovery.js"; import * as controllerRecovery from "./native-controller-recovery.js"; import { isSettledRemoteWarmControlState } from "./remote-warm-session-recovery.js"; @@ -17430,6 +17431,64 @@ describe("runnerd provider runtime wiring", () => { expect(runnerEnvironment).toMatchObject(paths); }); + it("keeps the provider backup fence while its accepted result is saving the workspace", async () => { + const current = { ...execution, binding: { ...execution.binding, runId: randomUUID(), issueId: randomUUID(), + executionWorkspaceId: randomUUID() }, session: { ...execution.session, normalizedSessionId: randomUUID() } }; + const coordinator = { phase: "observed", resultId: null as string | null, leaseOwner: "provider-owner" as string | null, + controllerGeneration: 1, controllerBootId: "backup-boot" }; + const db = leaseDb(current), select = db.select.bind(db); + db.select = ((fields?: Record) => ({ from: (table: unknown) => { + if (table !== nativeRunFinalizations) return select(fields as never).from(table as never); + const promise = Promise.resolve([coordinator]); + const query = { then: promise.then.bind(promise), where: () => query, limit: () => query, for: () => query }; + return query; + } })) as unknown as Db["select"]; + const storage = vi.spyOn(harnessStorage, "durableNativeHarnessStorage").mockReturnValue({ provider: "s3" } as never); + const validateOwner = vi.spyOn(harnessStorage, "assertNativeHarnessBackupOwner").mockResolvedValue(undefined); + const checkpointRegistration = vi.spyOn(runnerCheckpoint, "nativeRunnerCheckpointRegistration") + .mockReturnValue(async () => undefined); + onTestFinished(() => { storage.mockRestore(); validateOwner.mockRestore(); checkpointRegistration.mockRestore(); }); + let savingWorkspace = false; + const remoteExecute = vi.fn(async (command: { args?: string[] }) => { + if (savingWorkspace && command.args?.[1]?.includes("base64 <")) throw new Error("stopped-provider-inspection-reached"); + return { exitCode: command.args?.[1]?.includes("base64") ? 1 : 0, timedOut: false, stderr: "", stdout: + command.args?.[0] === "--build-metadata" ? JSON.stringify({ + schema: "paperclip-runner/runnerd-build-metadata/v1", binaryName: "paperclip-runnerd", packageName: "@paperclipai/paperclip-runner", + binaryContractVersion: 2, durableSessionCapabilities: ["unlimited_runtime", "connection_lease_renewal"], prpTransportModes: ["listen_ws"], + }) : command.args?.[0] === "--version" ? "codex-cli 0.160.0" : "" }; + }); + const controllerArtifact = join(isolatedStateDirectory, "backup-fixture-runnerd"); + await writeFile(controllerArtifact, "fixture bootstrap artifact"); + await createRunnerdBackend({ db, execution: current, runnerInstanceId: "backup-fence-runner", + runnerRemoteBinaryPath: controllerArtifact, runnerRemoteCodexPath: controllerArtifact, + controllerPersistence: { bootId: "backup-boot" }, runnerIngressAuthorized: true, runnerExecutionTarget: { + kind: "remote", transport: "sandbox", remoteCwd: "/workspace", environmentId: "environment", leaseId: "backup-fence-lease", + providerKey: "daytona", sandboxLeaseAcquisition: { outcome: "created", providerLeaseId: "backup-fence-sandbox" }, + effectiveCapabilities: { runnerWebSocketIngress: true }, runner: { execute: remoteExecute, syncIn: vi.fn(async () => undefined) }, + getRunnerIngressEndpoint: async () => ({ kind: "authenticated_websocket", websocketUrl: "wss://sandbox.invalid/runner", + secretHeaders: [{ name: "Authorization", value: "fixture-ingress" }], generation: "1", refresh: vi.fn(), close: vi.fn(async () => undefined) }), + } as never }); + state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + const options = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & { + controlPlaneRegistration: NonNullable; + }; + const controller = new DurablePrpControlPlane({ stateDirectory: join(options.stateDirectory!, "control-plane"), + identity: options.prpIdentity! as ConstructorParameters[0]["identity"], + expectedRunnerVersion: "backup-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + controller.queueCommand("run.prepare", {}); + controller.queueCommand("session.open", { reuse: "same_session" }); + let registration: Awaited> | undefined; + onTestFinished(async () => { await registration?.release(); await controller.stop(); }); + registration = await options.controlPlaneRegistration(controller); + coordinator.phase = "workspace_finalizing"; + coordinator.resultId = "accepted-result"; + coordinator.leaseOwner = null; + savingWorkspace = true; + await expect(registration.checkpoint!("settled")).rejects.toThrow("stopped-provider-inspection-reached"); + expect(validateOwner).toHaveBeenCalledWith(db, expect.objectContaining({ runId: current.binding.runId, + leaseOwner: "provider-owner", bootId: "backup-boot", generation: 1 })); + }); + it.each([true, false])("restores a migrated warm backup into a created lease only for its exact session (matching: %s)", async matching => { const remoteCwd = join(isolatedStateDirectory, "remote"); const current: NativeExecutionInputV1 = { ...execution, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 03507c2b2b3..c40cbcf8d7e 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -11,7 +11,7 @@ import { readProviderFailure } from "@paperclipai/shared"; import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js"; import { readClaimedNativeStopRecovery } from "./native-cancellation-request.js"; import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; -import { durableNativeHarnessStorage, nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; +import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; import { rehydratePinnedSkillAssets } from "./runtime-context.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeRecoveryUsageDelivery } from "./native-recovery-usage-delivery.js"; @@ -1931,6 +1931,7 @@ class SessionAuthorityEpoch { } const sessionToolAuthorityEpochs = new Map(); +const sessionHarnessBackupOwners = new WeakMap>>(); const initializingSessionToolAuthorities = new Set(); const executingRunnerdSessionScopes = new Map(); const executingNativeOwnerScopes = new Map(); @@ -13741,6 +13742,11 @@ async function createRunnerdBackendWithinSessionClaim( return { ...backend, descriptor: () => backend.descriptor(), openSession: input => backend.openSession(input), recoverSession: (snapshot, options) => backend.recoverSession!(snapshot, options), bindManagedSession: session => session }; } + if (remoteTarget && input.controllerPersistence && durableNativeHarnessStorage()) { + sessionHarnessBackupOwners.set(authorityEpoch, await nativeHarnessBackupOwner(input.db, { + ...input.execution.binding, bootId: input.controllerPersistence.bootId, + })); + } const remoteRuntimeRoot = remoteTarget ? posix.join( remoteTarget.remoteCwd, @@ -14594,10 +14600,10 @@ async function createRunnerdBackendWithinSessionClaim( ) return; const durableStorage = durableNativeHarnessStorage(); - if (durableStorage && !input.controllerPersistence) throw new Error("native_harness_storage_owner_missing"); - const backupOwner = durableStorage ? await nativeHarnessBackupOwner(input.db, { ...input.execution.binding, - runId: sessionToolAuthorityEpochs.get(sessionScopeId)?.runId ?? input.execution.binding.runId, - bootId: input.controllerPersistence!.bootId }) : null; + const currentAuthority = sessionToolAuthorityEpochs.get(sessionScopeId); + const backupOwner = durableStorage && currentAuthority ? sessionHarnessBackupOwners.get(currentAuthority) : null; + if (durableStorage && !backupOwner) throw new Error("native_harness_storage_owner_missing"); + if (backupOwner) await assertNativeHarnessBackupOwner(input.db, backupOwner); // Transport release also runs when provider bootstrap failed. In that case // runnerd has no durable provider identity (and may not have created the // provider persistence directory at all), so attempting a failover backup diff --git a/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts b/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts index 95f0c955fe0..b19150beafb 100644 --- a/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts +++ b/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts @@ -198,7 +198,7 @@ describeEmbeddedPostgres("managed warm checkpoint ownership", () => { expect(result.errorCode).toBe("native_provider_overloaded"); } expect(state.checkpoint).toBeTypeOf("function"); - if (invalidated) expect(closeFailures).toEqual(["Error: native_harness_storage_owner_missing"]); + if (invalidated || renewalLoss) expect(closeFailures).toEqual(["Error: native_runner_checkpoint_owner_lost"]); else expect(closeFailures).toEqual([]); expect(ownersAtClose).toHaveLength(1); expect(ownersAtClose[0].leaseOwner).not.toBeNull();