diff --git a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md index 29c0d19f613..4a2f1f250d3 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -289,6 +289,24 @@ manifest and suspended state. A concurrent turn, moved source, revoked stamp or changed owner defers that source. Unrelated executions continue unchanged. No lease, result, task or lifecycle state is rewritten to make a source eligible. +Run the controller-only phase separately so it does not re-read provider +archives that were already verified: + +```sh +PAPERCLIP_NATIVE_HARNESS_PRESTAGE=true node --max-old-space-size=512 server/dist-runtime/native-harness-prestage.mjs --controllers-only +``` + +This finite entry adopts legacy controller checkpoints for successful +retained sources whose finalizer is committed, has an accepted result, and has +released its ownership. It locks only the source task and coordinator while +reading and publishing the original controller state. Related queued/running +work, checkout, failed sources, recovery state, or any remaining coordinator +owner (including an expired owner) defers that source. Local controller and warm +supervisor bytes are rechecked before commit; a changed source rolls back. +Unrelated runs remain active. Existing controller checkpoint identity, artifact +provenance, configuration and snapshot checks still apply. The `paperclip.native-controller-prestage.v1` counts describe this phase; neither successful prestaging +nor zero deferred controller sources authorizes volume detachment. + The final offline migration still stops all writers, verifies every durable object, imports deferred sources and all other data, and exports only its normal complete eligibility receipt. Qualification starts from the saved pristine diff --git a/server/src/maintenance/native-harness-prestage.test.ts b/server/src/maintenance/native-harness-prestage.test.ts index f114f60b822..4660034bd54 100644 --- a/server/src/maintenance/native-harness-prestage.test.ts +++ b/server/src/maintenance/native-harness-prestage.test.ts @@ -1,9 +1,9 @@ -import { afterEach, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; import { spawnSync } from "node:child_process"; import { fileURLToPath } from "node:url"; import { build } from "esbuild"; -const mocks = vi.hoisted(() => ({ close: vi.fn(async () => {}), prestage: vi.fn(), key: vi.fn(async () => {}), obligations: vi.fn(async () => false) })); +const mocks = vi.hoisted(() => ({ close: vi.fn(async () => {}), prestage: vi.fn(), controllers: vi.fn(), key: vi.fn(async () => {}), obligations: vi.fn(async () => false) })); vi.mock("@paperclipai/db", () => ({ createDb: () => ({}), closeRegisteredClients: mocks.close })); vi.mock("../config.js", () => ({ loadConfig: () => ({ secretsMasterKeyFilePath: "/fixture/key" }) })); vi.mock("../services/cloud-runtime-config.js", () => ({ cloudRuntimeConfigService: () => ({}) })); @@ -13,12 +13,15 @@ vi.mock("../storage/index.js", () => ({ getStorageService: () => ({ provider: "s vi.mock("../services/native-runtime/native-harness-backup-migration.js", () => ({ prestageNativeHarnessBackups: mocks.prestage, hasLocalNativeHarnessObligations: mocks.obligations, })); +vi.mock("../services/native-runtime/native-controller-migration.js", () => ({ prestageNativeControllers: mocks.controllers })); vi.mock("./native-harness-prestage-runtime.js", async importOriginal => ({ ...await importOriginal(), initializeNativeHarnessPrestageKey: mocks.key, })); const initialExitCode = process.exitCode; -afterEach(() => { process.exitCode = initialExitCode; vi.useRealTimers(); vi.unstubAllEnvs(); vi.restoreAllMocks(); vi.resetModules(); mocks.close.mockClear(); mocks.obligations.mockReset().mockResolvedValue(false); }); +const initialArgv = process.argv; +beforeEach(() => { process.argv = [process.execPath, "native-harness-prestage"]; }); +afterEach(() => { process.argv = initialArgv; mocks.controllers.mockReset(); process.exitCode = initialExitCode; vi.useRealTimers(); vi.unstubAllEnvs(); vi.restoreAllMocks(); vi.resetModules(); mocks.close.mockClear(); mocks.obligations.mockReset().mockResolvedValue(false); }); it.each([true, false])("checks unfulfilled database obligations for an empty source scan (pending: %s)", async pending => { vi.stubEnv("PAPERCLIP_NATIVE_HARNESS_PRESTAGE", "true"); @@ -69,6 +72,7 @@ it("reports a failed deadline with real timers when the dependency holds no even export const isCloudManagedInstance = () => false, getStorageService = () => ({ provider: 's3' }); export const initializeNativeHarnessPrestageKey = async () => {}, nativeHarnessPrestageDatabaseUrl = url => url; export const prestageNativeHarnessBackups = () => new Promise(() => {}); + export const prestageNativeControllers = () => new Promise(() => {}); export const hasLocalNativeHarnessObligations = async () => false; ` })); } }], @@ -111,3 +115,20 @@ it.each(["SIGTERM", "deadline"])("bounds shutdown if a dependency does not settl expect(process.exitCode).toBe(1); expect(vi.getTimerCount()).toBe(0); }); + +it("copies only settled controllers when the explicit mode is selected", async () => { + vi.stubEnv("PAPERCLIP_NATIVE_HARNESS_PRESTAGE", "true"); + vi.stubEnv("DATABASE_URL", "postgres://fixture:fixture@localhost/fixture"); + process.argv.push("--controllers-only"); + const counts = { inspected: 2, published: 1, current: 0, deferred: { source_active: 1 } }; + mocks.controllers.mockResolvedValue(counts); + mocks.prestage.mockClear(); + const success = vi.spyOn(console, "log").mockImplementation(() => {}); + await import("./native-harness-prestage.js"); + expect(mocks.controllers).toHaveBeenCalledOnce(); + expect(mocks.controllers.mock.calls[0][1].signal.aborted).toBe(false); + expect(mocks.prestage).not.toHaveBeenCalled(); + expect(mocks.obligations).not.toHaveBeenCalled(); + expect(success).toHaveBeenCalledWith(JSON.stringify({ schema: "paperclip.native-controller-prestage.v1", counts, enrollmentReady: false })); + expect(mocks.close).toHaveBeenCalled(); +}); diff --git a/server/src/maintenance/native-harness-prestage.ts b/server/src/maintenance/native-harness-prestage.ts index fba4095a408..5103f78c1f4 100644 --- a/server/src/maintenance/native-harness-prestage.ts +++ b/server/src/maintenance/native-harness-prestage.ts @@ -5,9 +5,14 @@ import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config- import { isCloudManagedInstance } from "../services/cloud-instance.js"; import { getStorageService } from "../storage/index.js"; import { hasLocalNativeHarnessObligations, prestageNativeHarnessBackups } from "../services/native-runtime/native-harness-backup-migration.js"; +import { prestageNativeControllers } from "../services/native-runtime/native-controller-migration.js"; import { initializeNativeHarnessPrestageKey, nativeHarnessPrestageDatabaseUrl } from "./native-harness-prestage-runtime.js"; if (process.env.PAPERCLIP_NATIVE_HARNESS_PRESTAGE !== "true") throw new Error("Explicit harness pre-copy mode is required"); +const args = process.argv.slice(2); +if (args.length > 1 || args.some(arg => arg !== "--controllers-only")) throw new Error("Unknown native pre-copy mode"); +const controllersOnly = args[0] === "--controllers-only"; +const schema = controllersOnly ? "paperclip.native-controller-prestage.v1" : "paperclip.native-harness-prestage.v1"; const sourceUrl = process.env.DATABASE_URL; if (!sourceUrl) throw new Error("Harness pre-copy requires an explicit database"); const url = nativeHarnessPrestageDatabaseUrl(sourceUrl); @@ -20,7 +25,7 @@ const reportBlocked = (error: unknown) => { blocked = true; const message = error instanceof Error ? error.message : "native_harness_prestage_failed"; const failure = /^[a-z_]{1,80}$/.test(message) ? message : "native_harness_prestage_failed"; - console.error(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", state: "blocked", failure, enrollmentReady: false })); + console.error(JSON.stringify({ schema, state: "blocked", failure, enrollmentReady: false })); }; const abort = (reason: string) => { if (cancellation.signal.aborted) return; @@ -42,14 +47,20 @@ try { await initializeNativeHarnessPrestageKey(loadConfig().secretsMasterKeyFilePath); const storage = getStorageService(); if (storage.provider !== "s3") throw new Error("shared_object_storage_required"); - const counts = await prestageNativeHarnessBackups(db, { storage, signal: cancellation.signal }); - cancellation.signal.throwIfAborted(); - if (counts.inspected === 0) { - const obligations = await hasLocalNativeHarnessObligations(db); + if (controllersOnly) { + const counts = await prestageNativeControllers(db, { signal: cancellation.signal }); cancellation.signal.throwIfAborted(); - if (obligations) throw new Error("native_harness_prestage_source_missing"); + console.log(JSON.stringify({ schema, counts, enrollmentReady: false })); + } else { + const counts = await prestageNativeHarnessBackups(db, { storage, signal: cancellation.signal }); + cancellation.signal.throwIfAborted(); + if (counts.inspected === 0) { + const obligations = await hasLocalNativeHarnessObligations(db); + cancellation.signal.throwIfAborted(); + if (obligations) throw new Error("native_harness_prestage_source_missing"); + } + console.log(JSON.stringify({ schema, counts, enrollmentReady: false })); } - console.log(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false })); } catch (error) { reportBlocked(error); } finally { 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 ff5a26e692e..4caf3e39967 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 @@ -1,14 +1,14 @@ import { createHash, randomUUID } from "node:crypto"; -import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; +import { mkdir, open, readFile, rm, stat, writeFile } from "node:fs/promises"; import { dirname, join } from "node:path"; import { eq } from "drizzle-orm"; import { afterEach, expect, it, vi } from "vitest"; -import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db"; +import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db"; 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 { hasLocalNativeControllerObligations, migrateNativeControllers, prestageNativeControllers } from "./native-controller-migration.js"; import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; import { createNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js"; @@ -22,7 +22,7 @@ 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", "batch", "durable_failed", +describeEmbeddedPostgres.each(["online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed", "stale_confirmed", "stale_foreign", "stale_changed", "stale_queued", "stale_overlap", "stale_success", "stale_busy", "stale_durable", "stale_durable_absent", "stale_durable_runner", "stale_durable_session", "stale_durable_snapshot", "controller_confirmed", "controller_foreign", "controller_thread", "controller_runner", "controller_success", "controller_busy", "controller_invalid", "controller_durable", @@ -30,7 +30,7 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable "controller_observed_live", "controller_observed_result", "controller_observed_retry", "controller_observed_task", "controller_observed_old_receipt", "controller_observed_revoked", "controller_observed_handoff", "controller_observed_handoff_stopped", "controller_observed_handoff_broken", "controller_observed_handoff_fork"])("offline native controller migration (%s)", scenario => { const environment = nativeRestartFixture(); - afterEach(() => vi.unstubAllEnvs()); + afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); }); async function fixture(commandCount = 1, warm = false) { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "73".repeat(32)); const f = await environment.seedRun(randomUUID()); @@ -90,6 +90,91 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable taskKey: f.issueId, lastRunId: f.runId }).returning(); return { ...f, companyId, agentId, paperclipHome, runtimeRoot, paths, file, source, profile, session, queuedId, identity, snapshot }; } + if (scenario.startsWith("online_")) it("prestages only a settled legacy controller while unrelated work continues", async () => { + const f = await fixture(); + const settle = async (f: Awaited>) => { + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: run.completionContractId!, serverFingerprint: "online-migration", schemaStatus: "accepted", + resultJson: {}, canonicalSha256: "online-migration" }).returning(); + await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + await f.db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, f.queuedId)); + await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, + leaseExpiresAt: null, recoveryState: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + }; + await settle(f); + const unrelated = await environment.seedRun(randomUUID()); + const [unrelatedBefore] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)); + if (scenario === "online_owned") await f.db.update(nativeRunFinalizations).set({ + leaseOwner: "expired-owner-is-still-an-owner", leaseExpiresAt: new Date(0), + }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario === "online_same_source") await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId, + agentId: f.agentId, issueId: f.issueId, nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, + status: "running", runtimeMode: "native" }); + const before = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario === "online_evidence") { + await rm(join(f.paths.root, "provider-pack-authority.json")); + await f.db.update(agentTaskSessions).set({ id: "00000000-0000-4000-8000-000000000001" }).where(eq(agentTaskSessions.id, f.session.id)); + const next = await fixture(); + await settle(next); + const actual = await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) }); + expect(actual).toMatchObject({ inspected: 2, imported: 1, deferred: { artifact_unproven: 1 } }); + expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).toBeNull(); + expect(await loadNativeRunnerCheckpoint(f.db, { companyId: next.companyId, issueId: next.issueId, runId: next.runId })).not.toBeNull(); + expect(await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore); + return; + } + if (scenario === "online_changed") { + const inode = (await stat(f.file)).ino; + const probe = await open(f.file, "r"); + const prototype = Object.getPrototypeOf(probe); + const originalRead = prototype.readFile; + await probe.close(); + let changed = false; + vi.spyOn(prototype, "readFile").mockImplementation(async function(this: { stat: () => Promise<{ ino: number }> }, ...args: unknown[]) { + const bytes = await originalRead.apply(this, args); + if (!changed && (await this.stat()).ino === inode) { + changed = true; + await writeFile(f.file, f.source + "\n", { mode: 0o600 }); + } + return bytes; + }); + expect(await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) })) + .toMatchObject({ imported: 0, deferred: { source_changed: 1 } }); + expect(changed).toBe(true); + expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).toBeNull(); + expect(await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore); + return; + } + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let acquired!: () => void; + const locked = new Promise(resolve => { acquired = resolve; }); + const writer = scenario === "online_busy" ? f.db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); + acquired(); + await gate; + }) : null; + if (writer) await locked; + let actual: Awaited>; + try { actual = await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) }); } + finally { release(); await writer; } + if (scenario === "online_busy") expect(actual.deferred.source_busy).toBe(1); + if (scenario === "online_success") { + expect(actual.imported).toBe(1); + expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).not.toBeNull(); + expect((await prestageNativeControllers(f.db, { signal: AbortSignal.timeout(30_000) })).unchanged).toBeGreaterThanOrEqual(1); + } else { + expect(actual.imported).toBe(0); + expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).toBeNull(); + } + expect(await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore); + expect(await readFile(f.file, "utf8")).toBe(f.source); + }); if (scenario === "import") it("imports an exact settled authority once, preserves queued work, and resumes from an empty home", async () => { const f = await fixture(); const { companyId, agentId, paperclipHome, runtimeRoot, profile, file, source, session, queuedId, identity, snapshot } = f; diff --git a/server/src/services/native-runtime/native-controller-migration.ts b/server/src/services/native-runtime/native-controller-migration.ts index 443ffb73bed..77d2af449c3 100644 --- a/server/src/services/native-runtime/native-controller-migration.ts +++ b/server/src/services/native-runtime/native-controller-migration.ts @@ -1,8 +1,8 @@ import { constants } from "node:fs"; import { lstat, open, readdir } from "node:fs/promises"; import { join } from "node:path"; -import { and, eq, sql } from "drizzle-orm"; -import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { captureLegacyDurablePrpCheckpoint, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; import { findNativeSessionResumeRun } from "./native-session-resume.js"; @@ -157,6 +157,54 @@ export async function hasLocalNativeControllerObligations(db: Db, storage?: Stor return false; } +function hasMatchingControllerRecovery(selected: Selected, + source: NonNullable>>, + checkpoint: Awaited>) { + const metadata = readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint); + const snapshot = source.runnerProfileJson.sessionCheckpoint as { identity?: Record } | null; + return Boolean(checkpoint && metadata?.checkpoint?.runId === source.id + && metadata.checkpoint.runnerInstanceId === source.runnerInstanceId + && metadata.checkpoint.normalizedSessionId === selected.sessionId + && snapshot?.identity?.runId === source.id && snapshot.identity.companyId === selected.companyId + && snapshot.identity.agentId === selected.agentId && snapshot.identity.issueId === selected.issueId + && snapshot.identity.sessionId === selected.sessionId + && checkpoint.recoveryEvidence.runnerInstanceId === source.runnerInstanceId + && checkpoint.recoveryEvidence.normalizedSessionId === selected.sessionId); +} + +async function readMigrationMaterial(db: Db, selected: Selected, + source: NonNullable>>, + checkpoint: Awaited>) { + const paths = await localPaths(source); + const legacyControllerState = checkpoint ? undefined : await readPrivateSource(paths.controlFile, 256 * 1024 * 1024); + let legacyRunnerArtifact: { version: string; digest: string } | undefined; + if (!checkpoint && legacyControllerState === undefined) return { paths, legacyControllerState, legacyRunnerArtifact, supervisor: undefined }; + if (legacyControllerState !== undefined && await readPrivateSource(join(paths.root, "provider-pack-authority.json"), 512 * 1024)) { + const identity = JSON.parse(legacyControllerState).identity; + const execution = paths.execution; + const [admission] = await db.select({ workspace: sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'` }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, selected.companyId), + eq(heartbeatRuns.agentId, selected.agentId))); + const workspace = readNativeWorkspaceSyncReference(admission?.workspace); + const remoteCwd = workspace?.remoteCwd ?? execution.workspace.cwd; + if (typeof identity?.environmentLeaseId !== "string" || !remoteCwd + || (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"); }; + legacyRunnerArtifact = resolveRemoteProviderPackAuthority({ stateDirectory: paths.root, reattach: true, + binding: { companyId: selected.companyId, agentId: selected.agentId, + workspaceId: providerPackWorkspaceId(execution.binding.executionWorkspaceId, source.id, identity.environmentLeaseId), + sessionId: selected.sessionId, runnerInstanceId: source.runnerInstanceId!, + environmentLeaseId: identity.environmentLeaseId, remoteCwd, + providerDigest: nativeSha256(execution.provider) }, + current: unavailable, runnerArtifact: unavailable }).runnerArtifact; + } + const supervisor = !checkpoint && paths.execution.session.lifecyclePolicy.mode === "warm" + ? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined; + return { paths, legacyControllerState, legacyRunnerArtifact, supervisor }; +} + export async function migrateNativeControllers(db: Db, input: { writersStopped: true; signal: AbortSignal; storage?: StorageService }) { const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1); if (!input.writersStopped || active) throw new Error("native_controller_migration_requires_offline_writers"); @@ -166,43 +214,9 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped: const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id }; const checkpoint = await loadNativeRunnerCheckpoint(db, binding); if (preservedFailedCheckpoint(source, selected.sessionId, checkpoint)) { counts.preservedFailed++; continue; } - const metadata = readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint); - const snapshot = source.runnerProfileJson.sessionCheckpoint as { identity?: Record } | null; - if (checkpoint && metadata?.checkpoint?.runId === source.id - && metadata.checkpoint.runnerInstanceId === source.runnerInstanceId - && metadata.checkpoint.normalizedSessionId === selected.sessionId - && snapshot?.identity?.runId === source.id && snapshot.identity.companyId === selected.companyId - && snapshot.identity.agentId === selected.agentId && snapshot.identity.issueId === selected.issueId - && snapshot.identity.sessionId === selected.sessionId - && checkpoint.recoveryEvidence.runnerInstanceId === source.runnerInstanceId - && checkpoint.recoveryEvidence.normalizedSessionId === selected.sessionId) { counts.unchanged++; continue; } - const paths = await localPaths(source); - const legacyControllerState = checkpoint ? undefined : await readPrivateSource(paths.controlFile, 256 * 1024 * 1024); + if (hasMatchingControllerRecovery(selected, source, checkpoint)) { counts.unchanged++; continue; } + const { paths, legacyControllerState, legacyRunnerArtifact, supervisor } = await readMigrationMaterial(db, selected, source, checkpoint); if (!checkpoint && legacyControllerState === undefined) { counts.sourceAlreadyAbsent++; continue; } - let legacyRunnerArtifact: { version: string; digest: string } | undefined; - if (legacyControllerState !== undefined && await readPrivateSource(join(paths.root, "provider-pack-authority.json"), 512 * 1024)) { - const identity = JSON.parse(legacyControllerState).identity; - const execution = paths.execution; - const [admission] = await db.select({ workspace: sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'` }) - .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, selected.companyId), - eq(heartbeatRuns.agentId, selected.agentId))); - const workspace = readNativeWorkspaceSyncReference(admission?.workspace); - const remoteCwd = workspace?.remoteCwd ?? execution.workspace.cwd; - if (typeof identity?.environmentLeaseId !== "string" || !remoteCwd - || (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"); }; - legacyRunnerArtifact = resolveRemoteProviderPackAuthority({ stateDirectory: paths.root, reattach: true, - binding: { companyId: selected.companyId, agentId: selected.agentId, - workspaceId: providerPackWorkspaceId(execution.binding.executionWorkspaceId, source.id, identity.environmentLeaseId), - sessionId: selected.sessionId, runnerInstanceId: source.runnerInstanceId!, - environmentLeaseId: identity.environmentLeaseId, remoteCwd, - providerDigest: nativeSha256(execution.provider) }, - current: unavailable, runnerArtifact: unavailable }).runnerArtifact; - } - const supervisor = !checkpoint && paths.execution.session.lifecyclePolicy.mode === "warm" - ? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined; input.signal.throwIfAborted(); let result: Awaited>; try { @@ -240,3 +254,86 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped: } return counts; } + +function onlineControllerSourceDeferral(error: unknown): string | null { + if (!(error instanceof Error)) return null; + const reasons: Record = { + native_runner_checkpoint_artifact_unproven: "artifact_unproven", + native_runner_checkpoint_binding_mismatch: "checkpoint_binding", + native_controller_migration_source_changed: "source_changed", + native_controller_migration_unsafe_source: "unsafe_source", + native_controller_migration_artifact_binding: "artifact_binding", + native_controller_migration_artifact_unavailable: "artifact_unavailable", + }; + if (reasons[error.message]) return reasons[error.message]; + if (error.message.startsWith("runner_remote_provider_artifact_incompatible:")) return "artifact_incompatible"; + if (error instanceof SyntaxError) return "invalid_source_json"; + if (["ENOENT", "ENOTDIR", "EISDIR", "ELOOP", "EACCES", "EPERM"].includes((error as NodeJS.ErrnoException).code ?? "")) return "source_unavailable"; + return null; +} + +export async function prestageNativeControllers(db: Db, input: { signal: AbortSignal }) { + const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0, deferred: {} as Record }; + const defer = (reason: string) => { counts.deferred[reason] = (counts.deferred[reason] ?? 0) + 1; }; + for await (const { selected, source } of retainedSources(db, input.signal)) { + counts.inspected++; + input.signal.throwIfAborted(); + const outcome = await db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout = '1s'`); + const [task] = await tx.select().from(issues).where(and(eq(issues.companyId, selected.companyId), + eq(issues.id, selected.issueId))).for("update"); + if (!task || task.executionRunId || task.checkoutRunId) return "source_active"; + const [owner] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, selected.companyId), eq(nativeRunFinalizations.issueId, selected.issueId), + eq(nativeRunFinalizations.runId, source.id))).for("update"); + if (!owner || owner.leaseOwner || owner.phase !== "committed" || !owner.resultId || owner.recoveryState) return "owner_not_released"; + const [current] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id), + eq(heartbeatRuns.companyId, selected.companyId), eq(heartbeatRuns.agentId, selected.agentId), + eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.nativeSessionId, selected.sessionId))).for("update"); + if (current?.status !== "succeeded" || !current.finishedAt) return "source_not_successful"; + const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, selected.companyId), inArray(heartbeatRuns.status, ["queued", "running"]), + or(eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.issueId, selected.issueId), + eq(heartbeatRuns.nativeSessionId, selected.sessionId)))).limit(1); + if (active) return "source_active"; + const scoped = tx as unknown as Db; + const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id }; + const checkpoint = await loadNativeRunnerCheckpoint(scoped, binding); + const profile = record(current.runnerProfileJson); + const freshSource = { ...source, status: current.status, runnerInstanceId: current.runnerInstanceId, + runnerProfileJson: { nativeToolContractFingerprint: profile.nativeToolContractFingerprint, + nativeExecutionInput: profile.nativeExecutionInput, sessionCheckpoint: profile.sessionCheckpoint, + nativeControllerRecovery: profile.nativeControllerRecovery } }; + if (hasMatchingControllerRecovery(selected, freshSource, checkpoint)) return "unchanged"; + const material = await readMigrationMaterial(scoped, selected, freshSource, checkpoint); + if (!checkpoint && material.legacyControllerState === undefined) return "source_already_absent"; + input.signal.throwIfAborted(); + const result = await importNativeControllerRecovery(scoped, { ...binding, agentId: selected.agentId, + normalizedSessionId: selected.sessionId, expectedConfigDigest: material.paths.configDigest, + supervisor: material.supervisor === undefined ? undefined : JSON.parse(material.supervisor), + legacyCheckpoint: { legacyControllerState: material.legacyControllerState, + legacyRunnerArtifact: material.legacyRunnerArtifact } }); + if (result.kind !== "imported" && result.kind !== "unchanged") return "source_unproven"; + if (material.legacyControllerState !== undefined + && await readPrivateSource(material.paths.controlFile, 256 * 1024 * 1024) !== material.legacyControllerState) { + throw new Error("native_controller_migration_source_changed"); + } + if (material.supervisor !== undefined + && await readPrivateSource(material.paths.supervisorPath, 4 * 1024 * 1024) !== material.supervisor) { + throw new Error("native_controller_migration_source_changed"); + } + input.signal.throwIfAborted(); + return result.kind; + }).catch(error => { + input.signal.throwIfAborted(); + const reason = onlineControllerSourceDeferral(error); + if (reason) return reason; + if (error?.code === "55P03" || error?.cause?.code === "55P03") return "source_busy"; + throw error; + }); + if (outcome === "imported" || outcome === "unchanged") counts[outcome]++; + else if (outcome === "source_already_absent") counts.sourceAlreadyAbsent++; + else defer(outcome); + } + return counts; +} diff --git a/server/src/services/native-runtime/native-controller-recovery.ts b/server/src/services/native-runtime/native-controller-recovery.ts index e78beedfbe3..e4fbad449e0 100644 --- a/server/src/services/native-runtime/native-controller-recovery.ts +++ b/server/src/services/native-runtime/native-controller-recovery.ts @@ -1,4 +1,4 @@ -import { and, eq, ne, sql } from "drizzle-orm"; +import { and, eq, inArray, ne, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db"; import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import { captureLegacyDurablePrpCheckpoint, parseNativeExecutionInput, type DurablePrpCheckpoint, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; @@ -226,13 +226,16 @@ export async function importNativeControllerRecovery(db: Db, input: { expectedConfigDigest: string; supervisor?: unknown; offline?: { writersStopped: true; legacyControllerState?: string; legacyRunnerArtifact?: { version: string; digest: string } }; + legacyCheckpoint?: { legacyControllerState?: string; legacyRunnerArtifact?: { version: string; digest: string } }; }): Promise<{ kind: "imported" | "unchanged" } | Exclude> { + if (input.offline && input.legacyCheckpoint) throw new Error("Conflicting native controller migration authority"); + const legacy = input.offline ?? input.legacyCheckpoint; const supervisor = record(input.supervisor); if (input.supervisor !== undefined && (supervisor.schema !== "paperclip.native-session-supervisor.v1" || supervisor.configDigest !== input.expectedConfigDigest)) { return { kind: "unsupported", reason: "supervisor_compatibility" }; } return db.transaction(async tx => { - const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) + const [task] = await tx.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues) .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("share"); const [owner] = await tx.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId), @@ -252,6 +255,13 @@ export async function importNativeControllerRecovery(db: Db, input: { if (!input.offline && (!(owner.phase === "committed" && owner.resultId) || owner.recoveryState)) { return { kind: "missing_evidence" as const, reason: "pending_finalization" }; } + if (input.legacyCheckpoint) { + const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), inArray(heartbeatRuns.status, ["queued", "running"]), + or(eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.issueId, input.issueId), + eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId)))).limit(1); + if (task.checkoutRunId || active) return { kind: "missing_evidence" as const, reason: "legacy_source_active" }; + } const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId))).for("update"); @@ -266,7 +276,7 @@ export async function importNativeControllerRecovery(db: Db, input: { if (input.supervisor === undefined) { try { const admitted = parseNativeExecutionInput(profile.nativeExecutionInput); - if (!input.offline || (!checkpoint && admitted.session.lifecyclePolicy.mode !== "per_turn") + if (!legacy || (!checkpoint && admitted.session.lifecyclePolicy.mode !== "per_turn") || admitted.session.normalizedSessionId !== input.normalizedSessionId || admitted.binding.runId !== input.runId || admitted.binding.companyId !== input.companyId || admitted.binding.agentId !== input.agentId || admitted.binding.issueId !== input.issueId) { @@ -274,12 +284,12 @@ export async function importNativeControllerRecovery(db: Db, input: { } } catch { return { kind: "unsupported" as const, reason: "admitted_input_format" }; } } - const importing = !checkpoint && input.offline?.legacyControllerState !== undefined; + const importing = !checkpoint && legacy?.legacyControllerState !== undefined; if (importing) { if (!run.runnerInstanceId) return { kind: "missing_evidence" as const, reason: "checkpoint_binding" }; - checkpoint = captureLegacyDurablePrpCheckpoint(input.offline!.legacyControllerState!, { + checkpoint = captureLegacyDurablePrpCheckpoint(legacy!.legacyControllerState!, { runId: input.runId, runnerInstanceId: run.runnerInstanceId, normalizedSessionId: input.normalizedSessionId, - }, input.offline!.legacyRunnerArtifact); + }, legacy!.legacyRunnerArtifact); } if (!checkpoint || checkpoint.recoveryEvidence.normalizedSessionId !== input.normalizedSessionId || checkpoint.recoveryEvidence.runnerInstanceId !== run.runnerInstanceId) {