diff --git a/doc/plans/2026-11-04-preserve-native-workspace-migration.md b/doc/plans/2026-11-04-preserve-native-workspace-migration.md index 0f316804906..92cb05cd216 100644 --- a/doc/plans/2026-11-04-preserve-native-workspace-migration.md +++ b/doc/plans/2026-11-04-preserve-native-workspace-migration.md @@ -117,3 +117,28 @@ with matching, mismatched, and absent migration receipts. An executor-level test restores real backup files into a created warm lease only for the exact migrated session, verifies the retained provider identity, and checks the new lease's backup stamp. Neither test replaces the migration decision helper. + +### Live enrollment: protocol and physical lease identities + +The first full legacy import exposed a false binding failure: the controller's +`environmentLeaseId` retained its logical workspace identity, while +`nativeWorkspaceSync.leaseId` identified a distinct physical sandbox lease row. +The migration now validates the sync record's `workspaceId` against the frozen +execution workspace. The saved provider-pack authority still independently +verifies the exact protocol lease, runner, session, company, agent, provider and +remote path. A database-backed fixture uses distinct IDs, rejects foreign +workspaces and protocol identities, and restores the original checkpoint after +the predecessor home is deleted. The original failed live attempt and its +attached volume remain preserved until the corrected owner passes verification. + +The next live import exposed a second boundary: one insert of 538 encrypted +checkpoint records (about 2.2 MB) repeatedly closed the database connection on +both pooled and direct endpoints, including after warming the database cache. +The identical import completed with smaller inserts in a forced-rollback +transaction. The importer now limits each statement to 64 records and about +512 KiB of serialized rows; an individual larger record is sent alone. All +statements remain in the original owner transaction. A real PostgreSQL test +forces a conflict in a later batch, verifies that earlier batches roll back, +then imports and decrypts the complete original checkpoint. The precise +server-side reason for the connection closure is not yet established; the +unbatched failures and batched comparison remain in the verification record. diff --git a/server/src/services/native-runtime/native-controller-migration.integration.test.ts b/server/src/services/native-runtime/native-controller-migration.integration.test.ts index ea5d334e0ec..1c01589d8df 100644 --- a/server/src/services/native-runtime/native-controller-migration.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-migration.integration.test.ts @@ -4,11 +4,12 @@ import { join } from "node:path"; import { eq } from "drizzle-orm"; import { afterEach, expect, it, vi } from "vitest"; import { agentTaskSessions, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db"; -import { DurablePrpControlPlane, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; +import { captureLegacyDurablePrpCheckpoint, DurablePrpControlPlane, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; import { nativeToolContractFingerprintForTarget } from "./native-session-resume.js"; import { nativeControllerMigrationDescriptor } from "./native-session-executor.js"; import { hasLocalNativeControllerObligations, migrateNativeControllers } from "./native-controller-migration.js"; +import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; import { resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js"; import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js"; @@ -18,10 +19,10 @@ import { createStorageService } from "../../storage/service.js"; import { legacyNativeControllerArchive } from "./native-controller-archive.js"; import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; -describeEmbeddedPostgres.each(["import", "archive", "success"])("offline native controller migration (%s)", scenario => { +describeEmbeddedPostgres.each(["import", "archive", "success", "batch"])("offline native controller migration (%s)", scenario => { const environment = nativeRestartFixture(); afterEach(() => vi.unstubAllEnvs()); - async function fixture() { + async function fixture(commandCount = 1) { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "73".repeat(32)); const f = await environment.seedRun(randomUUID()); const { companyId, agentId, paperclipHome, runtimeRoot } = environment.current; @@ -42,7 +43,10 @@ describeEmbeddedPostgres.each(["import", "archive", "success"])("offline native environmentLeaseId: f.native.environmentLeaseId, turnId: f.native.turnId, itemId: f.native.itemId }; const core = new DurablePrpControlPlane({ stateDirectory: join(paths.root, "control-plane"), identity, expectedRunnerVersion: "migration-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); - core.issueBootstrapTicket(); core.queueCommand("session.snapshot", {}, "retained-command"); await core.stop(); + core.issueBootstrapTicket(); + for (let index = 0; index < commandCount; index++) core.queueCommand("session.snapshot", + { diagnostic: "x".repeat(commandCount > 1 ? 6000 : 0) }, `retained-command-${index}`); + await core.stop(); const file = join(paths.root, "control-plane/control-plane-state.json"); const state = JSON.parse(await readFile(file, "utf8")); state.tickets = {}; @@ -66,7 +70,7 @@ describeEmbeddedPostgres.each(["import", "archive", "success"])("offline native identity: { companyId, agentId, issueId: f.issueId, runId: f.runId, sessionId: f.native.normalizedSessionId } }; const profile = { ...f.preparedProfile, nativeExecutionInput: execution, nativeToolContractFingerprint: fingerprint, sessionCheckpoint: snapshot, nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared", descriptorSha256: "a".repeat(64), - baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: randomUUID(), leaseId: f.native.environmentLeaseId, + baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: binding.executionWorkspaceId, leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/remote/workspace", resourceDisposition: "destroy" } }; await f.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); await f.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId)); @@ -82,10 +86,19 @@ describeEmbeddedPostgres.each(["import", "archive", "success"])("offline native const { companyId, agentId, paperclipHome, runtimeRoot, profile, file, source, session, queuedId, identity, snapshot } = f; expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, - nativeWorkspaceSync: { ...profile.nativeWorkspaceSync, leaseId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId)); + nativeWorkspaceSync: { ...profile.nativeWorkspaceSync, workspaceId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId)); await expect(migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) })) .rejects.toThrow("native_controller_migration_artifact_binding"); await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + expect(profile.nativeWorkspaceSync.leaseId).not.toBe(identity.environmentLeaseId); + expect(profile.nativeWorkspaceSync.workspaceId).toBe(identity.environmentLeaseId); + const foreignProtocol = JSON.parse(source); + foreignProtocol.identity.environmentLeaseId = randomUUID(); + await writeFile(file, JSON.stringify(foreignProtocol), { mode: 0o600 }); + await expect(migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) })) + .rejects.toThrow("saved provider pack authority is invalid or does not match this execution"); + expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toHaveLength(0); + await writeFile(file, source, { mode: 0o600 }); await f.db.update(nativeRunFinalizations).set({ leaseOwner: "unsettled-owner" }).where(eq(nativeRunFinalizations.runId, f.runId)); await expect(migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) })).rejects.toThrow("normal_ownership_recovery_required"); expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toHaveLength(0); @@ -112,6 +125,41 @@ describeEmbeddedPostgres.each(["import", "archive", "success"])("offline native expect(JSON.parse(await readFile(join(replacement, "control-plane/control-plane-state.json"), "utf8")).identity).toEqual(identity); expect(recovered.snapshot).toEqual(snapshot); }); + if (scenario === "batch") it("imports a large checkpoint with bounded writes and rolls back every batch on a late conflict", async () => { + const f = await fixture(160); + const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId }; + const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, { runId: f.runId, + runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId }, + { version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }); + const last = [...checkpoint.records].at(-1)!; + await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, { + ...checkpoint, records: new Map([last]), + })); + const before = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + const parameters: number[] = []; + const payloadBytes: number[] = []; + const previousDebug = f.db.$client.options.debug; + f.db.$client.options.debug = (_connection, query, values) => { + if (query.startsWith('insert into "native_runner_checkpoints"')) { + parameters.push(values.length); + payloadBytes.push(values.reduce((sum, value) => sum + Buffer.byteLength(JSON.stringify(value)), 0)); + } + }; + try { + await expect(f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, checkpoint))) + .rejects.toThrow(); + expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(before); + expect(parameters.length).toBeGreaterThan(1); + expect(Math.max(...parameters)).toBeLessThanOrEqual(320); + expect(Math.max(...payloadBytes)).toBeLessThanOrEqual(512 * 1024); + await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + expect(await migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) })) + .toMatchObject({ imported: 1 }); + const restored = await loadNativeRunnerCheckpoint(f.db, binding); + expect(restored?.records).toEqual(checkpoint.records); + expect(await readFile(f.file, "utf8")).toBe(f.source); + } finally { f.db.$client.options.debug = previousDebug; } + }, 60_000); if (scenario === "archive") it("preserves an already-failed legacy controller without inventing executable recovery authority", async () => { const f = await fixture(); await rm(join(f.paths.root, "provider-pack-authority.json")); diff --git a/server/src/services/native-runtime/native-controller-migration.ts b/server/src/services/native-runtime/native-controller-migration.ts index e760f7071dd..6665845fb4c 100644 --- a/server/src/services/native-runtime/native-controller-migration.ts +++ b/server/src/services/native-runtime/native-controller-migration.ts @@ -124,7 +124,7 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped: const workspace = readNativeWorkspaceSyncReference(admission?.workspace); const remoteCwd = workspace?.remoteCwd ?? execution.workspace.cwd; if (typeof identity?.environmentLeaseId !== "string" || !remoteCwd - || (workspace && workspace.leaseId !== identity.environmentLeaseId)) { + || (workspace && workspace.workspaceId !== execution.binding.executionWorkspaceId)) { throw new Error("native_controller_migration_artifact_binding"); } const unavailable = (): never => { throw new Error("native_controller_migration_artifact_unavailable"); }; diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index 8c568cb6772..5d38781748f 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -20,7 +20,19 @@ function encryptParts(binding: Binding, records: ReadonlyMap) { export async function insertMigratedNativeRunnerCheckpoint(tx: CheckpointTransaction, binding: Binding, checkpoint: DurablePrpCheckpoint) { if (checkpoint.runId !== binding.runId) throw new Error("native_runner_checkpoint_binding_mismatch"); - await tx.insert(nativeRunnerCheckpoints).values(encryptParts(binding, checkpoint.records)); + let batch: ReturnType = []; + let batchBytes = 0; + for (const part of encryptParts(binding, checkpoint.records)) { + const bytes = Buffer.byteLength(JSON.stringify(part)); + if (batch.length && (batch.length >= 64 || batchBytes + bytes > 512 * 1024)) { + await tx.insert(nativeRunnerCheckpoints).values(batch); + batch = []; + batchBytes = 0; + } + batch.push(part); + batchBytes += bytes; + } + if (batch.length) await tx.insert(nativeRunnerCheckpoints).values(batch); } function decryptPart(cipher: ReturnType, binding: Binding,