diff --git a/doc/agent-files.md b/doc/agent-files.md index 2f6c83c46e1..765384f8fa0 100644 --- a/doc/agent-files.md +++ b/doc/agent-files.md @@ -362,3 +362,46 @@ protects interrupted one-time imports. Deleted-agent objects become eligible after the same grace. Deletion checks the current head under the agent lock and removes all versions. There is no automatic idle scan; operators own scheduling this maintenance during the bounded rollout. + + +### Existing-volume Cloud enrollment + +The runtime bundle also contains `node server/dist-runtime/ephemeral-migration.mjs`. +Cloud starts this dedicated process on the original volume after confirming that +all application deployments have stopped. It never starts app routes, queues or +agent execution. Set `PAPERCLIP_EPHEMERAL_MIGRATION=true`, a UUID attempt ID in +`PAPERCLIP_EPHEMERAL_MIGRATION_ID`, and a fresh 32-byte base64url token in +`PAPERCLIP_EPHEMERAL_MIGRATION_TOKEN`. Use the original home and instance ID, +external database, shared S3 configuration and existing stable secrets. Durable +agent files and shared UI assets must be enabled; ephemeral-server mode stays +false until adoption completes. + +The process uses the existing agent-file, run-evidence and skill maintenance +owners, follows all pages, verifies stored agent bytes and skill versions, and +checks whole-database eligibility. The trace owner expires all due retention +pages before the final check. Active execution, unresolved local working copies, +missing local bytes, local workspaces and external instructions block enrollment. Those conditions must be +resolved by their owning migration paths before detaching the volume. + +`GET /health` is a public maintenance liveness check. `GET /status` and +`POST /export` require the attempt token as Bearer authorization. Status includes +only the attempt, stamped Core commit, phase, counts and safe error code. Export +is available only after all checks succeed; it also returns the original decision +signing key for transfer into the stack vault. The process never generates a +replacement key. Every response disables caching. Never log or retain the export +body outside the vault. A 25-minute deadline blocks export even after migration +has succeeded; SIGTERM stops the +process, with a ten-second exit deadline. + +Cloud must verify the attempt and image commit, preserve the exact key, and +confirm this process is stopped before detaching the volume and starting the +normal ephemeral server. Keep tenant routing closed throughout this one-time +maintenance. Retain the old volume for recovery evidence; it must not become +writable authority again after shared state advances. + +The opt-in `ephemeral-migration.test.ts` uses disposable PostgreSQL and an +S3-compatible endpoint. Set `PAPERCLIP_TEST_POSTGRES_ADMIN_URL`, +`PAPERCLIP_TEST_S3_ENDPOINT` and normal AWS credentials to run it. It verifies +multi-page adoption, filesystem destruction, fresh reads, corrupt object rejection, +and missing-byte/active-execution refusals. The default suite does not contact +external storage. diff --git a/server/scripts/bundle-runtime.mjs b/server/scripts/bundle-runtime.mjs index 45754641187..e36a8c8749a 100644 --- a/server/scripts/bundle-runtime.mjs +++ b/server/scripts/bundle-runtime.mjs @@ -71,10 +71,11 @@ function externalizeThirdParty(isBundled, runtimeDir) { }; } -export async function bundleRuntime({ entry, outdir, maintenanceEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) { +export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) { await rm(outdir, { recursive: true, force: true }); const result = await build({ - entryPoints: { index: "paperclip-runtime-entry", server: entry, ...(maintenanceEntry ? { "agent-files": maintenanceEntry } : {}) }, + entryPoints: { index: "paperclip-runtime-entry", server: entry, ...(maintenanceEntry ? { "agent-files": maintenanceEntry } : {}), + ...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}) }, outdir, outExtension: { ".js": ".mjs" }, chunkNames: "chunk-[hash]", @@ -103,7 +104,8 @@ export async function bundleRuntime({ entry, outdir, maintenanceEntry, isBundled async function main() { const started = performance.now(); const summary = await bundleRuntime({ entry: join(serverDir, "dist", "index.js"), outdir: join(serverDir, "dist-runtime"), - maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js") }); + maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js"), + ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js") }); console.log(`[bundle-runtime] ${summary.inputs} inputs -> ${summary.chunks} chunks in ${Math.round(performance.now() - started)} ms: ${summary.outfile}`); } diff --git a/server/scripts/bundle-runtime.test.mjs b/server/scripts/bundle-runtime.test.mjs index 78f4072ac84..a634d11b2ac 100644 --- a/server/scripts/bundle-runtime.test.mjs +++ b/server/scripts/bundle-runtime.test.mjs @@ -80,6 +80,15 @@ describe("bundleRuntime", () => { expect(JSON.parse(stdout)).toEqual({ asset: "asset beside its source", phase: "--phase=eligibility" }); }); + it("packages standalone enrollment without starting the app", async () => { + const ephemeralMigrationEntry = path.join(root, "server", "dist", "migration.js"); + write(ephemeralMigrationEntry, 'import { values } from "@paperclipai/fixture"; console.log(JSON.stringify({ migrated: values.doubled }));'); + const outdir = path.join(root, "server", "dist-runtime"); + await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir, ephemeralMigrationEntry }); + const { stdout } = await execFileAsync(process.execPath, [path.join(outdir, "ephemeral-migration.mjs")], { timeout: 30_000 }); + expect(JSON.parse(stdout)).toEqual({ migrated: 42 }); + }); + it("bundles workspace TypeScript and runs with plain node", async () => { const outdir = path.join(root, "server", "dist-runtime"); const summary = await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir }); diff --git a/server/src/__tests__/ephemeral-migration.test.ts b/server/src/__tests__/ephemeral-migration.test.ts new file mode 100644 index 00000000000..69733b6eb8d --- /dev/null +++ b/server/src/__tests__/ephemeral-migration.test.ts @@ -0,0 +1,194 @@ +import fs from "node:fs/promises"; +import { existsSync } from "node:fs"; +import { spawn } from "node:child_process"; +import { createServer } from "node:net"; +import { fileURLToPath } from "node:url"; +import { setTimeout as delay } from "node:timers/promises"; +import os from "node:os"; +import path from "node:path"; +import { randomBytes, randomUUID } from "node:crypto"; +import { S3Client, CreateBucketCommand, DeleteBucketCommand, DeleteObjectsCommand, ListObjectsV2Command, PutObjectCommand } from "@aws-sdk/client-s3"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { activityLog, agents, agentFileRoots, agentInstructionWorkingCopies, companies, createDb, closeFileStorageDb, heartbeatRuns, providerTraceRecords } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { migrateEphemeralServerData } from "../maintenance/ephemeral-migration-data.js"; +import { agentFileStore } from "../services/agent-file-store.js"; +import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js"; +import { materializeInstructionBytes } from "../services/agent-instruction-files.js"; +import { verifyAgentFileObjects } from "../services/agent-file-maintenance.js"; +import { getStorageService } from "../storage/index.js"; +import { resolveDecisionSigningSecret } from "../services/decision-signing.js"; + +// Explicit real-dependency qualification. Never contacts off-host storage in +// the default test path; use a disposable S3-compatible service and PostgreSQL. +describe.skipIf(!process.env.PAPERCLIP_TEST_S3_ENDPOINT)("existing-volume enrollment with PostgreSQL and S3", () => { + let database: Awaited>; + let db: ReturnType, s3: S3Client; + let home: string, companyId: string, bucket: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("ephemeral-migration-"); db = createDb(database.connectionString); + s3 = new S3Client({ endpoint: process.env.PAPERCLIP_TEST_S3_ENDPOINT, region: "us-east-1", forcePathStyle: true }); + }, 90_000); + beforeEach(async () => { + companyId = randomUUID(); bucket = `enrollment-${randomUUID()}`; + home = await fs.mkdtemp(path.join(os.tmpdir(), "ephemeral-migration-")); + await s3.send(new CreateBucketCommand({ Bucket: bucket })); + vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_DURABLE_AGENT_FILES", "true"); + vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "false"); vi.stubEnv("PAPERCLIP_STORAGE_PROVIDER", "s3"); + vi.stubEnv("PAPERCLIP_STORAGE_S3_BUCKET", bucket); vi.stubEnv("PAPERCLIP_STORAGE_S3_REGION", "us-east-1"); + vi.stubEnv("PAPERCLIP_STORAGE_S3_ENDPOINT", process.env.PAPERCLIP_TEST_S3_ENDPOINT!); + vi.stubEnv("PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE", "true"); + await db.insert(companies).values({ id: companyId, name: "Enrollment fixture", issuePrefix: "MIG" }); + }); + afterEach(async () => { + await db.delete(activityLog).where(eq(activityLog.companyId, companyId)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + await db.delete(agents).where(eq(agents.companyId, companyId)); + await db.delete(companies).where(eq(companies.id, companyId)); + let token: string | undefined; + do { + const page = await s3.send(new ListObjectsV2Command({ Bucket: bucket, ContinuationToken: token })); + if (page.Contents?.length) await s3.send(new DeleteObjectsCommand({ Bucket: bucket, Delete: { Objects: page.Contents.map(row => ({ Key: row.Key! })) } })); + token = page.NextContinuationToken; + } while (token); + await s3.send(new DeleteBucketCommand({ Bucket: bucket })); + await fs.rm(home, { recursive: true, force: true }); vi.unstubAllEnvs(); + }); + afterAll(async () => { + if (db) { await closeFileStorageDb(db, 1); await db.$client.end({ timeout: 1 }); } + await database?.cleanup(); s3?.destroy(); + }); + const migrate = () => migrateEphemeralServerData(db, { signal: AbortSignal.timeout(120_000), phase: () => {} }); + async function agent(index = 0) { + const row = { id: randomUUID(), companyId, name: `Migrated ${index}`, adapterConfig: {} }; + const root = resolveManagedInstructionsRoot(row); + await materializeInstructionBytes(root, "AGENTS.md", Buffer.from(`Original ${index}`)); + await db.insert(agents).values({ ...row, adapterConfig: { instructionsBundleMode: "managed", instructionsRootPath: root } }); + return { ...row, root }; + } + it("migrates every page, verifies object bytes, and reads after deleting the entire predecessor filesystem", async () => { + const rows = []; + for (let index = 0; index < 26; index++) rows.push(await agent(index)); + const binary = Buffer.alloc(2 * 1024 * 1024, 0x81); + await materializeInstructionBytes(rows[0]!.root, "nested/data.bin", binary); + await materializeInstructionBytes(rows[0]!.root, "empty.txt", Buffer.alloc(0)); + expect(await migrate()).toMatchObject({ eligibility: { eligible: true }, counts: { agentFiles: 26, verifiedAgentObjects: 28 } }); + await fs.rm(home, { recursive: true, force: true }); + home = await fs.mkdtemp(path.join(os.tmpdir(), "ephemeral-successor-")); vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true"); vi.stubEnv("PAPERCLIP_DURABLE_AGENT_FILES", "false"); + // The service's normal board actor uses instance-operator authority. + const actor = { type: "board" as const, source: "local_implicit" as const, userId: "local-board" }; + const store = agentFileStore(db); + for (let index = 0; index < rows.length; index++) { + expect(await store.read(companyId, rows[index]!.id, "AGENTS.md", actor)).toEqual(Buffer.from(`Original ${index}`)); + } + expect(await store.read(companyId, rows[0]!.id, "nested/data.bin", actor)).toEqual(binary); + const [head] = await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, rows[0]!.id)); + const file = head!.entries.find(entry => entry.kind === "file" && entry.objectKey !== null)!; + if (file.kind !== "file" || !file.objectKey) throw new Error("Missing object"); + await s3.send(new PutObjectCommand({ Bucket: bucket, Key: file.objectKey, Body: "corrupt" })); + await expect(verifyAgentFileObjects(db, getStorageService(), AbortSignal.timeout(30_000))).rejects.toThrow(/agent_file_object_(size|digest)_mismatch/); + }, 120_000); + const entry = fileURLToPath(new URL("../../dist-runtime/ephemeral-migration.mjs", import.meta.url)); + it.skipIf(!existsSync(entry))("runs the actual bundled maintenance entry with plain Node and exits cleanly", async () => { + await agent(); + vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", ""); + const original = resolveDecisionSigningSecret(); + const listener = createServer(); + await new Promise(resolve => listener.listen(0, "127.0.0.1", resolve)); + const address = listener.address(); + if (!address || typeof address === "string") throw new Error("Missing listener"); + await new Promise(resolve => listener.close(() => resolve())); + const id = randomUUID(), token = randomBytes(32).toString("base64url"); + const stamp = JSON.parse(await fs.readFile(new URL("../../dist/build-info.json", import.meta.url), "utf8")); + const child = spawn(process.execPath, [entry], { env: { ...process.env, DATABASE_URL: database.connectionString, + PAPERCLIP_DATABASE_LISTEN_URL: database.connectionString, PAPERCLIP_SECRETS_MASTER_KEY: randomBytes(32).toString("hex"), + PAPERCLIP_SHARED_UI_ASSETS: "true", PAPERCLIP_EPHEMERAL_MIGRATION: "true", PAPERCLIP_EPHEMERAL_MIGRATION_ID: id, + PAPERCLIP_EPHEMERAL_MIGRATION_TOKEN: token, PORT: String(address.port) }, stdio: ["ignore", "pipe", "pipe"] }); + let diagnostic = ""; + child.stderr.on("data", chunk => { diagnostic = (diagnostic + String(chunk)).slice(-8_192); }); + child.stdout.resume(); + const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => { + child.once("error", reject); child.once("exit", (code, signal) => resolve({ code, signal })); + }); + const origin = `http://127.0.0.1:${address.port}`; + const headers = { authorization: `Bearer ${token}` }; + try { + let status: Record | undefined; + for (const until = Date.now() + 45_000; Date.now() < until;) { + if (child.exitCode !== null) throw new Error(`Maintenance exited: ${diagnostic}`); + status = await fetch(`${origin}/status`, { headers, signal: AbortSignal.timeout(1_000) }).then(r => r.json() as Promise>).catch(() => undefined); + if (status?.state === "ready" || status?.state === "blocked") break; + await delay(100); + } + expect(status).toMatchObject({ id, commit: stamp.commit, state: "ready", eligibility: { eligible: true } }); + expect(JSON.stringify(status)).not.toContain(original); + const exported = await fetch(`${origin}/export`, { method: "POST", headers }).then(r => r.json()); + expect(exported).toMatchObject({ id, commit: stamp.commit, decisionSigningSecret: original }); + expect((await fetch(`${origin}/api/companies`, { headers })).status).toBe(404); + child.kill("SIGTERM"); + expect(await Promise.race([exited, delay(12_000).then(() => { throw new Error("Maintenance did not exit gracefully"); })])).toEqual({ code: 0, signal: null }); + } finally { + if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + await exited; + } + }, 60_000); + + it("refuses active executions and missing predecessor bytes without publishing an empty head", async () => { + const row = await agent(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: row.id, status: "running" }).returning(); + await expect(migrate()).rejects.toThrow("active_execution_requires_drain"); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run!.id)); + await fs.rm(home, { recursive: true, force: true }); + await expect(migrate()).rejects.toThrow(); + expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, row.id))).toEqual([]); + }); + + it("retains unsaved terminal-run working copies until their owner records settlement", async () => { + const row = await agent(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: row.id, status: "succeeded" }).returning(); + const privateRoot = path.join(home, "private-copy"); + await materializeInstructionBytes(privateRoot, "AGENTS.md", Buffer.from("Unsaved work")); + await db.insert(agentInstructionWorkingCopies).values({ runId: run!.id, companyId, agentId: row.id, + responsibleUserId: "fixture", entryFile: "AGENTS.md", baseHash: "fixture", localRoot: privateRoot, + executionRoot: privateRoot, location: "local", state: "pending_collection" }); + await expect(migrate()).rejects.toThrow("local_instruction_copies_require_settlement"); + expect(await fs.readFile(path.join(privateRoot, "AGENTS.md"), "utf8")).toBe("Unsaved work"); + expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, row.id))).toEqual([]); + // Simulate only the owning service's completed collection receipt. The + // enrollment coordinator must not fabricate that save or erase the copy. + await materializeInstructionBytes(row.root, "AGENTS.md", Buffer.from("Unsaved work")); + await db.update(agentInstructionWorkingCopies).set({ state: "saved", processStoppedAt: new Date() }) + .where(eq(agentInstructionWorkingCopies.runId, run!.id)); + expect((await migrate()).eligibility.eligible).toBe(true); + }); + + it("expires every page of local traces through their owner before final eligibility", async () => { + const row = await agent(); + const runs = await db.insert(heartbeatRuns).values(Array.from({ length: 101 }, () => ({ + companyId, agentId: row.id, status: "succeeded", + }))).returning(); + await db.insert(providerTraceRecords).values(runs.map(run => ({ companyId, runId: run.id, provider: "codex", + traceRef: `${randomUUID()}.ndjson`, status: "complete", requestedBy: "fixture", expiresAt: new Date(0) }))); + expect(await migrate()).toMatchObject({ eligibility: { eligible: true }, counts: { expiredTraces: 101 } }); + expect((await db.select().from(providerTraceRecords).where(eq(providerTraceRecords.companyId, companyId))) + .every(trace => trace.status === "expired" && trace.deletedAt !== null)).toBe(true); + }, 120_000); + + it("accepts an unavailable directory only after its owner confirms stop and no-save cleanup", async () => { + const row = await agent(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: row.id, status: "succeeded" }).returning(); + await db.insert(agentInstructionWorkingCopies).values({ runId: run!.id, companyId, agentId: row.id, + responsibleUserId: "fixture", entryFile: "AGENTS.md", baseHash: "fixture", localRoot: path.join(home, "rejected-copy"), + executionRoot: path.join(home, "rejected-copy"), location: "local", state: "unavailable", + errorCode: "AGENT_FILES_LIMIT_EXCEEDED", receipt: { schema: "paperclip.agent-files.v1", cleanupPending: true } }); + await expect(migrate()).rejects.toThrow("local_instruction_copies_require_settlement"); + await db.update(agentInstructionWorkingCopies).set({ receipt: { schema: "paperclip.agent-files.v1", cleanupPending: false } }) + .where(eq(agentInstructionWorkingCopies.runId, run!.id)); + await expect(migrate()).rejects.toThrow("local_instruction_copies_require_settlement"); + await db.update(agentInstructionWorkingCopies).set({ processStoppedAt: new Date() }) + .where(eq(agentInstructionWorkingCopies.runId, run!.id)); + expect((await migrate()).eligibility.eligible).toBe(true); + }); +}); diff --git a/server/src/maintenance/ephemeral-migration-data.ts b/server/src/maintenance/ephemeral-migration-data.ts new file mode 100644 index 00000000000..17c713b0f12 --- /dev/null +++ b/server/src/maintenance/ephemeral-migration-data.ts @@ -0,0 +1,108 @@ +import { inArray } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { getStorageService } from "../storage/index.js"; +import { migrateAgentFiles, verifyAgentFileObjects } from "../services/agent-file-maintenance.js"; +import { readEphemeralServerDataEligibility } from "../services/ephemeral-server-contract.js"; +import { migrateRunEvidence } from "../services/run-evidence-migration.js"; +import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; +import { skillPackageMaintenance } from "../services/skill-package-maintenance.js"; +import { hasUnsettledLocalInstructionCopies } from "../services/agent-instruction-working-copies.js"; +import { providerTraceStore } from "../services/provider-trace-store.js"; + +/** Runs only in the maintenance process after Cloud has confirmed all legacy + * deployments stopped. Each owner retains its own migration/publication rules; + * a restart repeats idempotent pages, never reimports over durable authority. */ +export async function migrateEphemeralServerData(db: Db, input: { + signal: AbortSignal; phase: (name: string) => void; +}) { + const { signal, phase } = input; + const counts: Record = {}; + const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(inArray(heartbeatRuns.status, ["running", "queued"])).limit(1); + if (active) throw new Error("active_execution_requires_drain"); + if (await hasUnsettledLocalInstructionCopies(db)) throw new Error("local_instruction_copies_require_settlement"); + const before = await readEphemeralServerDataEligibility(db); + const unsupported = before.blockers.filter(blocker => ![ + "agent_files_require_migration", "run_logs_require_migration", "provider_traces_require_migration", "skills_require_migration", + ].includes(blocker)); + if (unsupported.length) throw new Error(unsupported[0]); + + phase("agent_files"); + let after: string | undefined; + do { + signal.throwIfAborted(); + const page = await migrateAgentFiles(db, { apply: true, writersStopped: true, after, limit: 25 }); + if (!page.pageReady) throw new Error("agent_file_migration_blocked"); + counts.agentFiles = (counts.agentFiles ?? 0) + page.results.length; + after = page.next ?? undefined; + } while (after); + + phase("run_evidence"); + // Separate cursors avoid restarting an exhausted kind when the other kind + // still has pages. A blocked item fails enrollment instead of being skipped. + for (const kind of ["log", "trace"] as const) { + after = undefined; + do { + signal.throwIfAborted(); + const page = await migrateRunEvidence(db, { apply: true, kinds: [kind], limit: 50, signal, + ...(kind === "log" ? { afterRunId: after } : { afterTraceId: after }) }); + if (page.items.some(item => item.state !== "migrated")) throw new Error(`${kind}_migration_blocked`); + counts[kind] = (counts[kind] ?? 0) + page.items.length; + after = kind === "log" ? page.nextRunId : page.nextTraceId; + } while (after); + } + + const packages = skillPackageMaintenance(db); + const skills = skillInstallationMaintenance(db); + for (const [name, migrate] of [["skill_packages", packages.migrateSnapshots], ["skill_versions", packages.migrateVersions]] as const) { + phase(name); + after = undefined; + for (;;) { + signal.throwIfAborted(); + const page: Awaited> = await migrate({ apply: true, after, limit: 50, signal }); + counts[name] = (counts[name] ?? 0) + page.inspected; + if (page.inspected < 50 || !page.after) break; + after = page.after; + } + } + phase("skill_installations"); + after = undefined; + for (;;) { + signal.throwIfAborted(); + const page = await skills.installations({ apply: true, writersStopped: true, after, limit: 50, signal }); + if (page.results.some(row => !["converted", "unchanged"].includes(row.status))) throw new Error("skill_installation_migration_blocked"); + counts.skillInstallations = (counts.skillInstallations ?? 0) + page.inspected; + if (page.inspected < 50 || !page.after) break; + after = page.after; + } + + phase("verify_agent_objects"); + const storage = getStorageService(); + if (storage.provider !== "s3") throw new Error("shared_object_storage_required"); + counts.verifiedAgentObjects = await verifyAgentFileObjects(db, storage, signal); + phase("verify_skill_versions"); + after = undefined; + for (;;) { + signal.throwIfAborted(); + const page = await skills.verifyVersions({ after, limit: 50, signal }); + if (page.results.some(row => !["remote_storage", "database_release_snapshot", "database_snapshot", "expired_git_history"].includes(row.status))) { + throw new Error("skill_version_verification_failed"); + } + counts.verifiedSkillVersions = (counts.verifiedSkillVersions ?? 0) + page.inspected; + if (page.inspected < 50 || !page.after) break; + after = page.after; + } + phase("eligibility"); + // A trace can expire after the app stops or while other owners migrate. + // Its owner tombstones and cleans all expired pages before final eligibility. + const traces = providerTraceStore(db), expiryAt = new Date(); + for (;;) { + signal.throwIfAborted(); + const removed = await traces.cleanupExpired(expiryAt, signal); + counts.expiredTraces = (counts.expiredTraces ?? 0) + removed.length; + if (removed.length < 100) break; + } + const eligibility = await readEphemeralServerDataEligibility(db); + if (!eligibility.eligible) throw new Error("final_data_eligibility_failed"); + return { counts, eligibility }; +} diff --git a/server/src/maintenance/ephemeral-migration-service.test.ts b/server/src/maintenance/ephemeral-migration-service.test.ts new file mode 100644 index 00000000000..4f11ca250cc --- /dev/null +++ b/server/src/maintenance/ephemeral-migration-service.test.ts @@ -0,0 +1,75 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { afterEach, expect, it } from "vitest"; +import { startEphemeralMigrationService, type EphemeralMigrationResult } from "./ephemeral-migration-service.js"; + +let service: Awaited> | undefined; +afterEach(async () => { await service?.close(); service = undefined; }); +const result = (): EphemeralMigrationResult => ({ counts: { files: 2 }, eligibility: { eligible: true, blockers: [] }, decisionSigningSecret: randomBytes(32).toString("base64") }); +async function start(migrate: Parameters[0]["migrate"], timeoutMs?: number) { + const id = randomUUID(), token = randomBytes(32).toString("base64url"), commit = "a".repeat(40); + service = await startEphemeralMigrationService({ id, commit, token, port: 0, host: "127.0.0.1", migrate, timeoutMs }); + const address = service.server.address(); + if (!address || typeof address === "string") throw new Error("Missing listener"); + const request = (pathname: string, method = "GET", authorized = true) => fetch(`http://127.0.0.1:${address.port}${pathname}`, { + method, headers: authorized ? { authorization: `Bearer ${token}` } : {}, + }); + return { request, id, commit }; +} + +it("keeps the original key out of health/status and exports only after authenticated successful migration", async () => { + const original = result(); + let finish!: (value: EphemeralMigrationResult) => void; + const { request, id, commit } = await start(() => new Promise(resolve => { finish = resolve; })); + expect((await request("/health", "GET", false)).status).toBe(200); + expect((await request("/status", "GET", false)).status).toBe(401); + expect((await request("/export", "POST", false)).status).toBe(401); + expect((await request("/export", "POST")).status).toBe(409); + expect((await request("/api/companies")).status).toBe(404); + finish(original); await service!.finished; + const status = await request("/status"); + expect(status.headers.get("cache-control")).toBe("no-store"); + const body = await status.json(); + expect(body).toMatchObject({ id, commit, state: "ready", counts: original.counts }); + expect(JSON.stringify(body)).not.toContain(original.decisionSigningSecret); + expect(await (await request("/export", "POST")).json()).toMatchObject({ id, commit, decisionSigningSecret: original.decisionSigningSecret }); +}); + +it("fails closed without exposing owner errors or credentials", async () => { + const { request } = await start(async () => { throw new Error("provider rejected secret-value at https://private.invalid"); }); + await service!.finished; + expect(await (await request("/status")).json()).toMatchObject({ state: "blocked", failure: "owner_migration_failed" }); + expect((await request("/export", "POST")).status).toBe(409); +}); + +it("rejects incomplete eligibility even when a key was recovered", async () => { + const { request } = await start(async () => ({ ...result(), eligibility: { eligible: false, blockers: ["legacy_data"] } })); + await service!.finished; + expect((await request("/export", "POST")).status).toBe(409); + expect(service!.receipt().failure).toBe("incomplete_migration_receipt"); +}); + +it("publishes a blocked deadline immediately and never exports a late completion", async () => { + let finish!: (value: EphemeralMigrationResult) => void; + let aborted!: Promise; + const { request } = await start(signal => { + aborted = new Promise(resolve => signal.addEventListener("abort", () => resolve(), { once: true })); + return new Promise(resolve => { finish = resolve; }); + }, 20); + await aborted; + expect(service!.receipt()).toMatchObject({ state: "blocked", failure: "maintenance_deadline_exceeded" }); + finish(result()); await service!.finished; + expect((await request("/export", "POST")).status).toBe(409); +}); + +it("expires successful export authority at the original maintenance deadline", async () => { + let expired!: Promise; + const { request } = await start(async signal => { + expired = new Promise(resolve => signal.addEventListener("abort", () => resolve(), { once: true })); + return result(); + }, 100); + await service!.finished; + expect(service!.receipt().state).toBe("ready"); + await expired; + expect(service!.receipt()).toMatchObject({ state: "blocked", failure: "maintenance_deadline_exceeded" }); + expect((await request("/export", "POST")).status).toBe(409); +}); diff --git a/server/src/maintenance/ephemeral-migration-service.ts b/server/src/maintenance/ephemeral-migration-service.ts new file mode 100644 index 00000000000..e4cfd957790 --- /dev/null +++ b/server/src/maintenance/ephemeral-migration-service.ts @@ -0,0 +1,82 @@ +import { timingSafeEqual } from "node:crypto"; +import { createServer } from "node:http"; + +export interface EphemeralMigrationResult { + counts: Record; + eligibility: { eligible: boolean; blockers: string[] }; + decisionSigningSecret: string; +} + +/** A dedicated maintenance process, never mounted on the ordinary app router. + * Cloud supplies a fresh attempt token and keeps tenant routing closed. Only + * the authenticated export transfers the original key to its durable vault. */ +export async function startEphemeralMigrationService(input: { + id: string; commit: string; token: string; port: number; host?: string; timeoutMs?: number; + migrate: (signal: AbortSignal, phase: (name: string) => void) => Promise; +}) { + if (!/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(input.id) || !/^[a-f0-9]{40}$/i.test(input.commit) + || !/^[A-Za-z0-9_-]{43,128}$/.test(input.token)) throw new Error("Invalid maintenance identity or credential"); + const authorization = Buffer.from(`Bearer ${input.token}`); + const cancellation = new AbortController(); + let state: "running" | "ready" | "blocked" = "running", phase = "configuration", failure: string | undefined; + let result: EphemeralMigrationResult | undefined; + const startedAt = new Date().toISOString(); + let completedAt: string | undefined; + const receipt = () => ({ schema: "paperclip.ephemeral-migration.v1", id: input.id, commit: input.commit, + state, phase, startedAt, completedAt, failure, counts: result?.counts, eligibility: result?.eligibility }); + const server = createServer((request, response) => { + response.setHeader("cache-control", "no-store"); + response.setHeader("content-type", "application/json"); + const send = (code: number, body: unknown) => { response.writeHead(code); response.end(JSON.stringify(body)); }; + if (request.method === "GET" && request.url === "/health") return send(200, { status: "maintenance" }); + const supplied = Buffer.from(request.headers.authorization ?? ""); + if (supplied.length !== authorization.length || !timingSafeEqual(supplied, authorization)) return send(401, { error: "unauthorized" }); + if (request.method === "GET" && request.url === "/status") return send(200, receipt()); + if (request.method === "POST" && request.url === "/export") { + if (state !== "ready" || !result) return send(409, receipt()); + return send(200, { ...receipt(), decisionSigningSecret: result.decisionSigningSecret }); + } + send(404, { error: "not_found" }); + }); + server.requestTimeout = 10_000; + server.headersTimeout = 10_000; + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(input.port, input.host ?? "0.0.0.0", () => { server.off("error", reject); resolve(); }); + }); + cancellation.signal.addEventListener("abort", () => { + state = "blocked"; + failure = cancellation.signal.reason instanceof Error ? cancellation.signal.reason.message : "maintenance_stopped"; + completedAt = new Date().toISOString(); + }, { once: true }); + const deadline = setTimeout(() => cancellation.abort(new Error("maintenance_deadline_exceeded")), input.timeoutMs ?? 25 * 60_000); + const finished = (async () => { + try { + const migrated = await input.migrate(cancellation.signal, next => { phase = next; }); + cancellation.signal.throwIfAborted(); + if (!migrated.eligibility.eligible || migrated.decisionSigningSecret.trim().length < 32) throw new Error("incomplete_migration_receipt"); + result = migrated; state = "ready"; + } catch (error) { + // Owner codes are safe diagnostics. Never echo arbitrary SDK errors, + // credential-bearing URLs, file contents or stack traces through HTTP. + const code = error instanceof Error ? error.message : "owner_migration_failed"; + failure = /^[a-z_]{1,80}$/.test(code) ? code : "owner_migration_failed"; + state = "blocked"; + } finally { + // Successful export authority expires at the original deadline too. + // A failed migration has no export authority to retain. + if (state !== "ready") clearTimeout(deadline); + completedAt = new Date().toISOString(); + } + })(); + return { server, finished, receipt, + async close() { + cancellation.abort(new Error("maintenance_stopped")); + clearTimeout(deadline); + server.closeIdleConnections(); + server.closeAllConnections(); + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + await finished; + }, + }; +} diff --git a/server/src/maintenance/ephemeral-migration.ts b/server/src/maintenance/ephemeral-migration.ts new file mode 100644 index 00000000000..cbef0d790e9 --- /dev/null +++ b/server/src/maintenance/ephemeral-migration.ts @@ -0,0 +1,42 @@ +import { closeRegisteredClients, createDb } from "@paperclipai/db"; +import { readBuildCommit } from "../build-commit.js"; +import { loadConfig } from "../config.js"; +import { cloudRuntimeConfigService } from "../services/cloud-runtime-config.js"; +import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js"; +import { isCloudManagedInstance } from "../services/cloud-instance.js"; +import { resolveDecisionSigningSecret } from "../services/decision-signing.js"; +import { assertEphemeralServerConfig } from "../services/ephemeral-server-contract.js"; +import { migrateEphemeralServerData } from "./ephemeral-migration-data.js"; +import { startEphemeralMigrationService } from "./ephemeral-migration-service.js"; + +// Cloud starts this image entry only after its provider has confirmed the web +// and every legacy worker stopped. It never starts schedulers or app routes. +if (process.env.PAPERCLIP_EPHEMERAL_MIGRATION !== "true") throw new Error("Explicit maintenance mode is required"); +const url = process.env.DATABASE_URL; +const commit = readBuildCommit(); +if (!url || !commit) throw new Error("Maintenance requires an explicit database and stamped image"); +const service = await startEphemeralMigrationService({ + id: process.env.PAPERCLIP_EPHEMERAL_MIGRATION_ID ?? "", token: process.env.PAPERCLIP_EPHEMERAL_MIGRATION_TOKEN ?? "", + commit, port: Number(process.env.PORT ?? 3100), + async migrate(signal, phase) { + const db = createDb(url, { maxConnections: 2 }); + try { + if (isCloudManagedInstance()) { + const config = await cloudRuntimeConfigService(db).inspectStored(); + if (!config) throw new Error("cloud_runtime_configuration_missing"); + setCloudRuntimeConfigSnapshot(config); + } + // Never generate a replacement key over a missing volume. The original + // key is exported only after every owner confirms durable eligibility. + const decisionSigningSecret = resolveDecisionSigningSecret({ allowCreate: false }); + assertEphemeralServerConfig(loadConfig(), { ...process.env, PAPERCLIP_EPHEMERAL_SERVER: "true", + PAPERCLIP_DECISION_SIGNING_SECRET: decisionSigningSecret }); + const result = await migrateEphemeralServerData(db, { signal, phase }); + return { ...result, decisionSigningSecret }; + } finally { await closeRegisteredClients(url, { scope: "database" }); } + }, +}); +for (const signal of ["SIGINT", "SIGTERM"] as const) process.once(signal, () => { + const deadline = setTimeout(() => process.exit(1), 10_000); + void service.close().then(() => { clearTimeout(deadline); process.exit(0); }, () => process.exit(1)); +}); diff --git a/server/src/services/agent-file-maintenance.ts b/server/src/services/agent-file-maintenance.ts index bafd76e7847..1888e742008 100644 --- a/server/src/services/agent-file-maintenance.ts +++ b/server/src/services/agent-file-maintenance.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; import { and, asc, eq, gt, inArray } from "drizzle-orm"; @@ -96,3 +97,34 @@ export async function collectAgentFileObjects(db: Db, provider: Pick { + if (storage.provider !== "s3") throw new Error("shared_object_storage_required"); + let after: string | undefined; + let verified = 0; + for (;;) { + signal.throwIfAborted(); + const heads = await db.select().from(agentFileRoots).where(after ? gt(agentFileRoots.agentId, after) : undefined) + .orderBy(asc(agentFileRoots.agentId)).limit(25); + for (const head of heads) for (const file of head.entries) { + if (file.kind !== "file") continue; + const hash = createHash("sha256"); + let bytes = 0; + if (file.objectKey !== null) { + const object = await storage.getObject(head.companyId, file.objectKey, { signal }); + for await (const chunk of object.stream) { + signal.throwIfAborted(); bytes += chunk.length; + if (bytes > file.size) throw new Error("agent_file_object_size_mismatch"); + hash.update(chunk); + } + } + if (bytes !== file.size || hash.digest("hex") !== file.sha256) throw new Error("agent_file_object_digest_mismatch"); + verified++; + } + if (heads.length < 25) break; + after = heads.at(-1)!.agentId; + } + return verified; +} diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts index 50e61014072..3e7a6363ecf 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -11,7 +11,7 @@ import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; import fs from "node:fs/promises"; import path from "node:path"; -import { and, asc, desc, eq, gt, inArray, lte, or, isNull, sql, type SQL } from "drizzle-orm"; +import { and, asc, desc, eq, gt, inArray, not, notInArray, lte, or, isNull, isNotNull, sql, type SQL } from "drizzle-orm"; import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, databaseWorkScopeKey, type Db } from "@paperclipai/db"; import { prepareAdapterExecutionTargetRuntime, @@ -41,6 +41,19 @@ const liveTargets = new Map(); const targetKey = (companyId: string, runId: string) => `${companyId}:${runId}`; const copySaves = new WeakMap>>(); +/** A terminal run does not establish that its private local edits were saved. + * Enrollment may discard these paths only after this owner's save/resolution + * or explicit ownership transfer. A rejected directory save is settled only + * after this owner confirms both process stop and completed no-save cleanup. */ +export async function hasUnsettledLocalInstructionCopies(db: Db): Promise { + return (await db.select({ runId: copies.runId }).from(copies).where(and( + eq(copies.location, "local"), notInArray(copies.state, ["saved", "unchanged", "resolved", "superseded"]), + not(and(eq(copies.state, "unavailable"), isNotNull(copies.processStoppedAt), + sql`coalesce(${copies.receipt}->>'schema', '') = 'paperclip.agent-files.v1'`, + sql`coalesce(${copies.receipt}->>'cleanupPending', '') = 'false'`)!), + )).limit(1)).length > 0; +} + function serializeCopySave(db: Db, companyId: string, runId: string, save: () => Promise): Promise { const scope = databaseWorkScopeKey(db); let active = copySaves.get(scope); diff --git a/server/src/services/decision-signing.test.ts b/server/src/services/decision-signing.test.ts new file mode 100644 index 00000000000..6b8abb40f15 --- /dev/null +++ b/server/src/services/decision-signing.test.ts @@ -0,0 +1,30 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { existsSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { resolveDefaultSecretsKeyFilePath } from "../home-paths.js"; +import { resolveDecisionSigningSecret, signDecisionSpec, verifyDecisionSpec } from "./decision-signing.js"; + +let home: string; +beforeEach(async () => { + home = await mkdtemp(path.join(tmpdir(), "decision-migration-")); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", ""); +}); +afterEach(async () => { vi.unstubAllEnvs(); await rm(home, { recursive: true, force: true }); }); +it("does not manufacture a replacement key when the legacy volume is missing", () => { + const key = path.join(path.dirname(resolveDefaultSecretsKeyFilePath()), "decision-signing.key"); + expect(() => resolveDecisionSigningSecret({ allowCreate: false })).toThrow(); + expect(existsSync(key)).toBe(false); +}); +it("preserves decision signatures after the original key moves to environment storage", async () => { + const original = resolveDecisionSigningSecret(); + const spec = { action: "approve", id: "retained-decision" }; + const signature = signDecisionSpec(spec); + expect(resolveDecisionSigningSecret({ allowCreate: false })).toBe(original); + await rm(home, { recursive: true, force: true }); + vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", original); + expect(resolveDecisionSigningSecret({ allowCreate: false })).toBe(original); + expect(verifyDecisionSpec(spec, signature)).toBe(true); +}); diff --git a/server/src/services/decision-signing.ts b/server/src/services/decision-signing.ts index 365e20ac7bb..69947f15a0a 100644 --- a/server/src/services/decision-signing.ts +++ b/server/src/services/decision-signing.ts @@ -120,7 +120,7 @@ function loadOrCreateGeneratedSecret(): string { } } -export function resolveDecisionSigningSecret(): string { +export function resolveDecisionSigningSecret(options: { allowCreate?: boolean } = {}): string { const fromEnv = process.env.PAPERCLIP_DECISION_SIGNING_SECRET?.trim(); if (fromEnv) { if (fromEnv.length < MIN_SECRET_LENGTH) { @@ -130,7 +130,9 @@ export function resolveDecisionSigningSecret(): string { } return fromEnv; } - return loadOrCreateGeneratedSecret(); + return options.allowCreate === false + ? readGeneratedSecret(resolveGeneratedSecretFilePath()) + : loadOrCreateGeneratedSecret(); } /**