From 349b8b3f7b1705d467f461e8072e40121ae13b47 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 17:19:23 -0600 Subject: [PATCH] Migrate settled controller checkpoints while unrelated work runs (#19631) ## Thinking Path > - Paperclip manages work performed by AI agents. > - Server replacement must preserve retained Runner sessions. > - Older sessions can retain controller checkpoints on the server volume. > - The existing importer requires all execution writers to stop, including unrelated runs. > - This change adds a separate online path for successful, settled sources. > - It locks the source task and finalizer while it reads and publishes that source. > - Unrelated work continues. The offline importer keeps its existing authority. ## Linked Issues or Issue Description **What happened?** A retained controller checkpoint cannot be converted by the existing maintenance importer while another task runs. Requiring a busy server to become globally idle prevents online migration. **Expected behavior** An operator can convert a successful retained source after its finalizer releases ownership, while unrelated tasks continue. An expired owner remains an owner. Related active work and uncertain evidence remain blocked. **Steps to reproduce** 1. Retain a valid legacy controller for a successful run with an accepted result and a released committed finalizer. 2. Start a different task. 3. Run controller preparation. The old importer refuses because execution writers are active. Related preservation contracts: #19038 and #19024. This change does not replace workspace migration or authorize volume removal. ## What Changed - Add `prestageNativeControllers` with task, finalizer, and source-run row locks. Recheck related runs and exact checkpoint identity under those locks. - Allow legacy checkpoint material through normal settled-owner import checks. Keep offline authority separate. - Compare captured local controller and supervisor bytes before commit. A changed source rolls back publication. - Add `--controllers-only` to the existing finite native preparation entry. This avoids re-reading provider archives during controller conversion. - Add PostgreSQL cases for unrelated work, expired owners, related active work, source-lock contention, and source changes. - Document the finite command and its limits. ## Verification - Real PostgreSQL affected suites: 56 passed across controller migration and recovery. A first per-server run was interrupted because it was slow; the completed run used the repository's shared PostgreSQL test wrapper. - Initial online cases: 3 passed, covering unrelated work, an expired owner, and related active work. - Source lock and changed-file cases: 2 passed. Publication rolled back on a source change. - Final source-evidence regression: 2 passed. A first unproven source deferred and a later valid source imported; changed-file rollback remained intact. - Server TypeScript check passed with the SDK and Runner declarations built. - Server package build passed, including the Runner vendor checks and compiled command. - Runtime bundling suite: 10 passed. - Full repository CI and exact-head automated review remain required before merge. This draft is not staging acceptance. - PASS: 7 CLI cases after adding the controller-only dependency fixture and explicit-mode coverage. The first CI run exposed missing mock/export coverage; the fixture repair passed after rebase. - The unrelated model-default failures were fixed on master; this branch now includes that fix. ## Risks - A source task can wait briefly on its own migration lock. A one-second lock acquisition budget defers a contended source. No global admission hold is acquired. - Only successful sources with a released committed finalizer are eligible. Failed or unsettled sources stay deferred and require their preservation owners. - The operator must run the command on the host that owns the original files. The command does not certify enrollment readiness or detach any volume. - The existing offline path remains unchanged and still requires stopped writers. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, tool use, and code execution. The exact serving model ID and context window are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...-existing-workspace-ephemeral-migration.md | 18 ++ .../native-harness-prestage.test.ts | 27 ++- .../maintenance/native-harness-prestage.ts | 25 ++- ...e-controller-migration.integration.test.ts | 95 +++++++++- .../native-controller-migration.ts | 173 ++++++++++++++---- .../native-controller-recovery.ts | 22 ++- 6 files changed, 301 insertions(+), 59 deletions(-) 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) {