diff --git a/doc/agent-files.md b/doc/agent-files.md index 92bea64d590..4b20e623c58 100644 --- a/doc/agent-files.md +++ b/doc/agent-files.md @@ -339,12 +339,12 @@ unit or matcher results; a passing matcher does not prove a live run. Run `server/src/maintenance/agent-files.ts` with the instance's database, storage, secrets and original `PAPERCLIP_HOME`. The default migration phase is -read-only. Before applying, stop ingress and execution writers on that instance -and set `PAPERCLIP_DURABLE_AGENT_FILES=true`: +read-only. Set `PAPERCLIP_DURABLE_AGENT_FILES=true` before applying on the original +host. Migration uses the same agent row lock as file edits and admission: ```sh pnpm exec tsx server/src/maintenance/agent-files.ts --phase=migration -pnpm exec tsx server/src/maintenance/agent-files.ts --phase=migration --apply --writers-stopped +pnpm exec tsx server/src/maintenance/agent-files.ts --phase=migration --apply ``` The production runtime build includes the same command without a TypeScript @@ -355,8 +355,11 @@ Eligibility is a point-in-time check; keep writers stopped through volume remova Follow each returned `next` cursor with `--after=`. All pages must report `pageReady: true` before removing the volume. A missing legacy directory, external -instructions or active execution prevents migration; resolve that condition on -the original host. Repeating apply verifies/restores the committed projection. +instructions, active execution without a durable head, or an unsettled private +local working copy prevents first adoption. An existing durable head needs no +local rewrite while its agent is active. Migration keeps copy recovery states +and retained provider references. The live eligibility gate separately refuses +unsettled local instruction copies, even after canonical agent files are durable. This migrates agent files only; other storage families have their own eligibility checks. Keep the original volume until the deployment migration is qualified. diff --git a/doc/skill-ephemeral-web-servers.md b/doc/skill-ephemeral-web-servers.md index f3c5b4e9a30..28194dca51b 100644 --- a/doc/skill-ephemeral-web-servers.md +++ b/doc/skill-ephemeral-web-servers.md @@ -90,15 +90,17 @@ use the latest branch as a replacement for missing installed content. `versions` phases. The latter also fills missing version metadata indexes so subsequent library/history reads need no object downloads. Keep original bytes until independent verification succeeds. -3. Stop admission and skill/source writers; retire every filesystem-authoritative - process. The `--writers-stopped` flag attests this operator action; it does not - stop the application itself. -4. Apply final installation conversion on the host with the legacy files: +3. Copy installations on the host with the legacy files. This phase shares the + cache, publication and row locks with normal writers. It checks the exact + local source again before publication: ```sh - pnpm --filter @paperclipai/server exec tsx scripts/migrate-skill-packages.ts --phase=installations --company= --limit=10 --apply --writers-stopped + pnpm --filter @paperclipai/server exec tsx scripts/migrate-skill-packages.ts --phase=installations --company= --limit=10 --apply ``` +4. Keep all original directories, versions and inline bytes. Conversion does not + settle old execution, recovery or pin uncertainty. Preserve or settle every + remaining local dependency through its owner before removing the volume. 5. Sweep `eligibility` and `verification`, including all historical versions. The latter verifies object bytes independently of caches. Follow all cursors; a successful first batch is not a complete readiness report. @@ -109,9 +111,10 @@ Managed conversion preserves skill IDs, keys, history, assignments, pins and acc provenance. Identical complete snapshots reuse their version IDs. Catalog conversion preserves the original reset version as well as installed content. A missing legacy directory can use a complete verified saved version. A partial Markdown-only rescue, -unknown revision, unsafe source tree or active legacy selection without exact version -identity is a blocker. Settle or snapshot such runs while A is still available; do -not cancel them simply to pass preflight. +unknown revision or unsafe source tree blocks conversion. An active legacy selection +without exact version identity remains a retention and volume-removal blocker even +after copy-only conversion. Preserve or settle such runs while A is still available; +do not cancel them simply to pass preflight. Report statuses distinguish linked-local sources, legacy managed/catalog records, missing current versions, unknown active-run selections, missing/corrupt packages, diff --git a/server/src/__tests__/agent-file-durability.test.ts b/server/src/__tests__/agent-file-durability.test.ts index eb23f65b520..c20e29ac96d 100644 --- a/server/src/__tests__/agent-file-durability.test.ts +++ b/server/src/__tests__/agent-file-durability.test.ts @@ -6,7 +6,7 @@ import { Readable } from "node:stream"; import { setTimeout as delay } from "node:timers/promises"; import { beforeAll, beforeEach, afterEach, afterAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; -import { agents, agentFileRoots, authUsers, companies, companyMemberships, principalPermissionGrants, heartbeatRuns, createDb, closeFileStorageDb } from "@paperclipai/db"; +import { agents, agentFileRoots, agentInstructionWorkingCopies, authUsers, companies, companyMemberships, principalPermissionGrants, heartbeatRuns, createDb, closeFileStorageDb } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { agentFileStore, adoptAgentFiles, fileHash } from "../services/agent-file-store.js"; import { invalidateAgentFileProjection, persistDurableAgentFiles } from "../services/agent-file-durability.js"; @@ -193,17 +193,42 @@ describe("durable agent file authority", () => { expect(await store().read(companyId, agentId, "failed.txt", actor())).toBeNull(); }); - it("dry-runs migration without adopting files and requires stopped writers for publication", async () => { + it("dry-runs without adopting files and publishes settled legacy files online", async () => { await materializeInstructionBytes(root, "AGENTS.md", Buffer.from("Legacy")); const input = { companyId, apply: false, writersStopped: false }; expect(await migrateAgentFiles(db, input)).toMatchObject({ results: [{ agentId, state: "needs_import" }], pageReady: false }); expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, agentId))).toEqual([]); - await expect(migrateAgentFiles(db, { ...input, apply: true })).rejects.toThrow("stopped writers"); - expect(await migrateAgentFiles(db, { ...input, apply: true, writersStopped: true })).toMatchObject({ pageReady: true }); + expect(await migrateAgentFiles(db, { ...input, apply: true })).toMatchObject({ pageReady: true }); await replaceFilesystem(); expect(await store().read(companyId, agentId, "AGENTS.md", actor())).toEqual(Buffer.from("Legacy")); }); + it("recognizes an active agent's existing durable head without restoring its filesystem projection", async () => { + await store().write({ companyId, agentId, path: "AGENTS.md", bytes: Buffer.from("Durable"), baseHash: null }, actor()); + const [active] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running", runtimeMode: "native" }).returning(); + const before = await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, agentId)); + await fs.rm(root, { recursive: true }); + expect(await migrateAgentFiles(db, { companyId, apply: true, writersStopped: false })) + .toMatchObject({ results: [{ agentId, state: "durable" }], pageReady: true }); + expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, agentId))).toEqual(before); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, active!.id)))[0]).toEqual(active); + await expect(fs.stat(root)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it.each(["pending_collection", "unavailable"])("preserves a %s working copy and keeps first adoption blocked", async state => { + await materializeInstructionBytes(root, "AGENTS.md", Buffer.from("Canonical files")); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", finishedAt: new Date() }).returning(); + const [copy] = await db.insert(agentInstructionWorkingCopies).values({ companyId, agentId, runId: run!.id, + responsibleUserId: userId, entryFile: "AGENTS.md", baseHash: "a".repeat(64), localRoot: path.join(home, "retained-copy"), + executionRoot: "/retained/agent-files", location: "local", state, + errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE", receipt: { schema: "paperclip.agent-files.v1" } }).returning(); + expect(await migrateAgentFiles(db, { companyId, apply: true, writersStopped: false })) + .toMatchObject({ results: [{ agentId, state: "working_copy_unsettled" }], pageReady: false }); + expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, agentId))).toEqual([]); + expect((await db.select().from(agentInstructionWorkingCopies).where(eq(agentInstructionWorkingCopies.runId, run!.id)))[0]).toEqual(copy); + expect(await fs.readFile(path.join(root, "AGENTS.md"), "utf8")).toBe("Canonical files"); + }); + it("refuses migration while the agent has queued or running execution", async () => { const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning(); for (const status of ["running", "queued"]) { diff --git a/server/src/__tests__/ephemeral-server-data.test.ts b/server/src/__tests__/ephemeral-server-data.test.ts index 21dd5b3db95..0baadc1068c 100644 --- a/server/src/__tests__/ephemeral-server-data.test.ts +++ b/server/src/__tests__/ephemeral-server-data.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { beforeAll, afterAll, afterEach, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { agentFileRoots, agents, companies, companySkills, companySkillVersions, executionWorkspaces, heartbeatRuns, providerTraceRecords, createDb } from "@paperclipai/db"; +import { agentFileRoots, agentInstructionWorkingCopies, agents, companies, companySkills, companySkillVersions, executionWorkspaces, heartbeatRuns, providerTraceRecords, createDb } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { assertEphemeralServerData, readEphemeralServerDataEligibility } from "../services/ephemeral-server-contract.js"; import { initializeDurableAgentFiles } from "../services/agent-file-durability.js"; @@ -66,3 +66,22 @@ it("keeps replacement readiness independent of a serving controller's unfinished await db.update(companySkills).set({ currentVersionId: null }).where(eq(companySkills.id, skillId)); await expect(assertEphemeralServerData(db)).rejects.toThrow("skills_require_migration"); }); + +it("keeps an unsettled local instruction copy visible after its canonical files become durable", async () => { + vi.stubEnv("PAPERCLIP_DURABLE_AGENT_FILES", "true"); + const companyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Private copy preservation", issuePrefix: "COPY" }); + await db.insert(agents).values({ id: agentId, companyId, name: "File owner" }); + await db.transaction(tx => initializeDurableAgentFiles(tx as unknown as Db, { id: agentId, companyId })); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", finishedAt: new Date() }).returning(); + const [copy] = await db.insert(agentInstructionWorkingCopies).values({ runId: run.id, companyId, agentId, + responsibleUserId: "preservation-fixture", entryFile: "AGENTS.md", baseHash: "a".repeat(64), + localRoot: "/private/unsaved-copy", executionRoot: "/private/unsaved-copy", location: "local", state: "pending_collection" }).returning(); + expect((await readEphemeralServerDataEligibility(db, { purpose: "startup" })).blockers) + .toContain("local_instruction_copies_require_settlement"); + expect((await db.select().from(agentInstructionWorkingCopies).where(eq(agentInstructionWorkingCopies.runId, run.id)))[0]).toEqual(copy); + await db.update(agentInstructionWorkingCopies).set({ state: "unavailable", processStoppedAt: new Date(), + receipt: { schema: "paperclip.agent-files.v1", cleanupPending: false } }).where(eq(agentInstructionWorkingCopies.runId, run.id)); + expect((await readEphemeralServerDataEligibility(db, { purpose: "startup" })).blockers) + .not.toContain("local_instruction_copies_require_settlement"); +}); diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index 0fb3ebcdc49..946c929b497 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -1,6 +1,8 @@ import { createServer } from "node:http"; import { createS3StorageProvider } from "../storage/s3-provider.js"; import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; +import { withSkillVersionCacheLock } from "../services/skill-version-cache.js"; +import { skillPackageStore } from "../services/skill-package-store.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; import { writePaperclipSkillSyncPreference } from "@paperclipai/adapter-utils/server-utils"; @@ -384,8 +386,7 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () await db.update(companySkillVersions).set({ packageFileMap: null }).where(eq(companySkillVersions.id, version.id)); const maintenance = skillInstallationMaintenance(db); expect((await maintenance.installations({ id: legacy!.id })).results[0]!.status).toBe("conversion_required"); - await expect(maintenance.installations({ companyId, apply: true })).rejects.toMatchObject({ status: 409 }); - expect((await maintenance.installations({ id: legacy!.id, apply: true, writersStopped: true })).results) + expect((await maintenance.installations({ id: legacy!.id, apply: true })).results) .toEqual([expect.objectContaining({ skillId: legacy!.id, versionId: version.id, status: "converted" })]); await fs.rm(directory, { recursive: true }); expect(await service.getById(companyId, legacy!.id)).toMatchObject({ sourceType: "managed", sourceLocator: null, currentVersionId: version.id }); @@ -393,6 +394,163 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () expect((await maintenance.eligibility({ id: legacy!.id })).results[0]!.status).toBe("version_backed"); }); + it("copies an installation online while preserving unknown old pins, old inline bytes, files, and active work", async () => { + const companyId = await company(), directory = path.join(root, `copy-only-${randomUUID()}`); + await fs.mkdir(directory); + await fs.writeFile(path.join(directory, "SKILL.md"), "# Preserved skill"); + const [skill] = await db.insert(companySkills).values({ companyId, key: "copy-only", slug: "copy-only", name: "Copy only", + markdown: "# Preserved skill", sourceType: "local_path", sourceLocator: directory, + fileInventory: [{ path: "SKILL.md", kind: "skill" }], metadata: { sourceKind: "managed_local" } }).returning(); + const files = [{ path: "SKILL.md", kind: "skill" as const, content: "# Preserved skill" }]; + const [version] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skill.id, + revisionNumber: 1, fileInventory: files }).returning(); + await db.update(companySkills).set({ currentVersionId: version.id }).where(eq(companySkills.id, skill.id)); + const [agent] = await db.insert(agents).values({ companyId, name: "Unrelated active worker" }).returning(); + const [active] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, runtimeMode: "native", + status: "running", contextSnapshot: {} }).returning(); + const maintenance = skillInstallationMaintenance(db); + expect((await maintenance.installations({ id: skill.id, apply: true })).results) + .toEqual([expect.objectContaining({ skillId: skill.id, status: "converted" })]); + expect((await db.select().from(companySkillVersions).where(eq(companySkillVersions.id, version.id)))[0].fileInventory).toEqual(files); + expect(await fs.readFile(path.join(directory, "SKILL.md"), "utf8")).toBe("# Preserved skill"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, active.id)))[0]).toEqual(active); + // Copy-only publication does not release retention or volume-removal guards. + expect((await maintenance.eligibility({ id: skill.id })).results[0].status).toBe("unsupported_active_run"); + }); + + it.each(["files", "source_locator"])("keeps a concurrent %s change instead of publishing the earlier conversion", async change => { + const companyId = await company(), directory = path.join(root, `conversion-race-${randomUUID()}`); + await fs.mkdir(directory); + await fs.writeFile(path.join(directory, "SKILL.md"), "# Race"); + await fs.writeFile(path.join(directory, "support.txt"), "old version"); + const [skill] = await db.insert(companySkills).values({ companyId, key: "conversion-race", slug: "conversion-race", name: "Race", + markdown: "# Race", sourceType: "local_path", sourceLocator: directory, + fileInventory: [{ path: "SKILL.md", kind: "skill" }, { path: "support.txt", kind: "other" }], + metadata: { sourceKind: "managed_local" } }).returning(); + const version = await companySkillService(db).createVersion(companyId, skill.id); + await fs.writeFile(path.join(directory, "support.txt"), "captured unsaved edit"); + const storage = storageModule.getStorageService(), put = storage.putFile.bind(storage); + let changed = false; + const nextDirectory = path.join(root, `new-source-${randomUUID()}`); + const upload = vi.spyOn(storage, "putFile").mockImplementation(async input => { + const saved = await put(input); + if (!changed) { + changed = true; + if (change === "files") await fs.writeFile(path.join(directory, "support.txt"), "newer unsaved edit"); + else { + await fs.mkdir(nextDirectory); + await fs.writeFile(path.join(nextDirectory, "SKILL.md"), "# New source"); + await db.transaction(async tx => { + await tx.select().from(companySkills).where(eq(companySkills.id, skill.id)).for("update"); + await tx.update(companySkills).set({ sourceLocator: nextDirectory }).where(eq(companySkills.id, skill.id)); + }); + } + } + return saved; + }); + try { + const result = await skillInstallationMaintenance(db, storage).installations({ id: skill.id, apply: true }); + expect(changed).toBe(true); + expect(result.results).toEqual([expect.objectContaining({ status: "blocked" })]); + const [current] = await db.select().from(companySkills).where(eq(companySkills.id, skill.id)); + expect(current).toMatchObject({ sourceType: "local_path", currentVersionId: version.id, + sourceLocator: change === "files" ? directory : nextDirectory }); + expect(await fs.readFile(path.join(directory, "support.txt"), "utf8")) + .toBe(change === "files" ? "newer unsaved edit" : "captured unsaved edit"); + } finally { upload.mockRestore(); } + }); + + it("holds a hard-linked local skill instead of adopting another file owner's mutable bytes", async () => { + const companyId = await company(), directory = path.join(root, `linked-conversion-${randomUUID()}`); + await fs.mkdir(directory); + const original = path.join(root, `shared-content-${randomUUID()}.md`); + await fs.writeFile(original, "# Shared source"); + await fs.link(original, path.join(directory, "SKILL.md")); + const [skill] = await db.insert(companySkills).values({ companyId, key: "linked-conversion", slug: "linked-conversion", name: "Linked", + markdown: "# Shared source", sourceType: "local_path", sourceLocator: directory, + fileInventory: [{ path: "SKILL.md", kind: "skill" }], metadata: { sourceKind: "managed_local" } }).returning(); + expect((await skillInstallationMaintenance(db).installations({ id: skill.id, apply: true })).results) + .toEqual([expect.objectContaining({ status: "blocked" })]); + expect((await db.select().from(companySkills).where(eq(companySkills.id, skill.id)))[0]).toEqual(skill); + expect(await fs.readFile(original, "utf8")).toBe("# Shared source"); + }); + + it("allows new durable skill pinning while an installation upload is blocked", async () => { + const companyId = await company(), directory = path.join(root, `pinning-during-copy-${randomUUID()}`); + await fs.mkdir(directory); + await fs.writeFile(path.join(directory, "SKILL.md"), "# Old selection"); + const [skill] = await db.insert(companySkills).values({ companyId, key: "online-pinning", slug: "online-pinning", name: "Pinning", + markdown: "# Old selection", sourceType: "local_path", sourceLocator: directory, + fileInventory: [{ path: "SKILL.md", kind: "skill" }], metadata: { sourceKind: "managed_local" } }).returning(); + const version = await companySkillService(db).createVersion(companyId, skill.id); + await fs.writeFile(path.join(directory, "SKILL.md"), `# Current local files ${randomUUID()}`); + const [agent] = await db.insert(agents).values({ companyId, name: "New work" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "queued" }).returning(); + const storage = storageModule.getStorageService(), put = storage.putFile.bind(storage); + let release!: () => void, ready!: () => void, blocked = false; + const barrier = new Promise(resolve => { release = resolve; }), uploading = new Promise(resolve => { ready = resolve; }); + const upload = vi.spyOn(storage, "putFile").mockImplementation(async input => { + if (!blocked) { blocked = true; ready(); await barrier; } + return put(input); + }); + const conversion = skillInstallationMaintenance(db, storage).installations({ id: skill.id, apply: true }); + try { + await uploading; + // This uses the real nonblocking PostgreSQL owner, not a timing assertion. + await withSkillVersionCacheLock(db, companyId, async () => { + await db.update(heartbeatRuns).set({ status: "running", contextSnapshot: { skillVersions: { [skill.key]: version.id } } }) + .where(eq(heartbeatRuns.id, run.id)); + }, true); + } finally { release(); await conversion; upload.mockRestore(); } + expect((await conversion).results).toEqual([expect.objectContaining({ status: "converted" })]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]) + .toMatchObject({ status: "running", contextSnapshot: { skillVersions: { [skill.key]: version.id } } }); + expect((await db.select().from(companySkillVersions).where(eq(companySkillVersions.id, version.id)))[0]).toBeDefined(); + }); + + it.each([true, false])("keeps catalog history unchanged on rejection and preserves its reset version (concurrent edit: %s)", async concurrentEdit => { + const companyId = await company(), directory = path.join(root, `catalog-copy-${randomUUID()}`), origin = `${directory}-original`; + await fs.mkdir(directory); await fs.mkdir(origin); + await fs.writeFile(path.join(directory, "SKILL.md"), "# Installed edit"); + await fs.writeFile(path.join(origin, "SKILL.md"), "# Original package"); + const [skill] = await db.insert(companySkills).values({ companyId, key: "catalog-copy", slug: "catalog-copy", name: "Catalog", + markdown: "# Installed edit", sourceType: "catalog", sourceLocator: directory, + fileInventory: [{ path: "SKILL.md", kind: "skill" }], metadata: { + sourceKind: "catalog", originSnapshotLocator: origin, userModifiedAt: new Date().toISOString(), + } }).returning(); + const [version] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skill.id, revisionNumber: 1, + fileInventory: [{ path: "SKILL.md", kind: "skill", content: "# Installed edit" }] }).returning(); + await db.update(companySkills).set({ currentVersionId: version.id }).where(eq(companySkills.id, skill.id)); + const storage = storageModule.getStorageService(), put = storage.putFile.bind(storage); + let changed = false; + const upload = vi.spyOn(storage, "putFile").mockImplementation(async input => { + const saved = await put(input); + if (!changed) { + changed = true; + if (concurrentEdit) await fs.writeFile(path.join(origin, "SKILL.md"), "# Concurrent original edit"); + } + return saved; + }); + try { + expect((await skillInstallationMaintenance(db, storage).installations({ id: skill.id, apply: true })).results) + .toEqual([expect.objectContaining({ status: concurrentEdit ? "blocked" : "converted" })]); + expect(changed).toBe(true); + const history = await db.select().from(companySkillVersions).where(eq(companySkillVersions.companySkillId, skill.id)); + expect(history).toHaveLength(concurrentEdit ? 1 : 2); + expect(history.find(row => row.id === version.id)).toMatchObject({ fileInventory: version.fileInventory }); + const [current] = await db.select().from(companySkills).where(eq(companySkills.id, skill.id)); + if (concurrentEdit) expect(current.metadata?.originVersionId).toBeUndefined(); + else { + const original = history.find(row => row.id !== version.id)!; + expect(current.metadata?.originVersionId).toBe(original.id); + const store = skillPackageStore(db, storage), saved = await store.version(companyId, skill.id, original.id); + expect((await store.versionFiles(saved!)).find(file => file.path === "SKILL.md")?.content).toBe("# Original package"); + } + expect(await fs.readFile(path.join(origin, "SKILL.md"), "utf8")) + .toBe(concurrentEdit ? "# Concurrent original edit" : "# Original package"); + } finally { upload.mockRestore(); } + }); + it("reports an unrecoverable legacy installation and leaves linked sources alone", async () => { const companyId = await company(); const [legacy] = await db.insert(companySkills).values({ companyId, key: "legacy-missing", slug: "legacy-missing", name: "Missing", diff --git a/server/src/services/agent-file-maintenance.ts b/server/src/services/agent-file-maintenance.ts index eec32d2fbed..ad10a5b1f2c 100644 --- a/server/src/services/agent-file-maintenance.ts +++ b/server/src/services/agent-file-maintenance.ts @@ -8,17 +8,18 @@ import type { StorageProvider, StorageService } from "../storage/types.js"; import { adoptAgentFiles, inspectAgentDirectory } from "./agent-file-store.js"; import { agentInstructionsBundleMode, deriveBundleState, resolveManagedInstructionsRoot } from "./agent-instructions.js"; import { durableAgentFilesEnabled } from "./agent-file-durability.js"; +import { hasUnsettledLocalInstructionCopies } from "./agent-instruction-copy-preservation.js"; export const AGENT_FILE_OBJECT_GRACE_MS = 30 * 24 * 60 * 60_000; -/** One-time operator migration on the host that owns the volume. Cloud must - * stop ingress and execution writers before apply; row locks also serialize - * ordinary file mutations. A dry run never adopts or rewrites local files. */ +/** Run on the host that owns the legacy volume. The canonical agent row lock + * serializes file writers and new admission. First adoption also requires its + * local working copies to be settled. A dry run never adopts local files. */ export async function migrateAgentFiles(db: Db, input: { apply: boolean; writersStopped: boolean; companyId?: string; after?: string; limit?: number; }) { - if (input.apply && (!input.writersStopped || !durableAgentFilesEnabled())) { - throw new Error("Agent file migration requires stopped writers and PAPERCLIP_DURABLE_AGENT_FILES=true"); + if (input.apply && !durableAgentFilesEnabled()) { + throw new Error("Agent file migration requires PAPERCLIP_DURABLE_AGENT_FILES=true"); } const limit = input.limit ?? 25; if (!Number.isSafeInteger(limit) || limit < 1 || limit > 100) throw new Error("Migration limit must be between 1 and 100"); @@ -32,19 +33,22 @@ export async function migrateAgentFiles(db: Db, input: { const [agent] = await tx.select().from(agents).where(eq(agents.id, candidate.id)).for("update"); if (!agent) return "removed"; if (agentInstructionsBundleMode(agent) === "external") return "external_instructions"; + const [head] = await tx.select({ generation: agentFileRoots.generation }).from(agentFileRoots) + .where(and(eq(agentFileRoots.companyId, agent.companyId), eq(agentFileRoots.agentId, agent.id))); + // Current shared authority needs no local rewrite, even during a run. + if (head) return "durable"; const [run] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, agent.companyId), eq(heartbeatRuns.agentId, agent.id), input.writersStopped && process.env.PAPERCLIP_EPHEMERAL_MIGRATION === "true" ? offlineMigrationExecutionWhere() : inArray(heartbeatRuns.status, ["running", "queued"]), )).limit(1); if (run) return "active_execution"; - const [head] = await tx.select({ generation: agentFileRoots.generation }).from(agentFileRoots) - .where(and(eq(agentFileRoots.companyId, agent.companyId), eq(agentFileRoots.agentId, agent.id))); + if (await hasUnsettledLocalInstructionCopies(tx, { companyId: agent.companyId, agentId: agent.id })) + return "working_copy_unsettled"; if (input.apply) { await adoptAgentFiles(tx, agent); return "durable"; } - if (head) return "durable"; const root = resolveManagedInstructionsRoot(agent); const stored = agent.adapterConfig.instructionsRootPath; if (typeof stored === "string" && (path.resolve(stored) !== root diff --git a/server/src/services/agent-instruction-copy-preservation.ts b/server/src/services/agent-instruction-copy-preservation.ts new file mode 100644 index 00000000000..5b8997ef22f --- /dev/null +++ b/server/src/services/agent-instruction-copy-preservation.ts @@ -0,0 +1,16 @@ +import { and, eq, isNotNull, not, notInArray, sql } from "drizzle-orm"; +import { agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db"; + +/** Terminal run status does not settle private edits. Only the copy owner's + * save/resolution or confirmed stopped no-save cleanup releases local bytes. */ +export async function hasUnsettledLocalInstructionCopies(db: Pick, + scope?: { companyId: string; agentId?: string }): Promise { + return (await db.select({ runId: copies.runId }).from(copies).where(and( + scope ? eq(copies.companyId, scope.companyId) : undefined, + scope?.agentId ? eq(copies.agentId, scope.agentId) : undefined, + 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; +} diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts index 586bd63dbfb..a3161d2e347 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, not, notInArray, lte, or, isNull, isNotNull, sql, type SQL } from "drizzle-orm"; +import { and, asc, desc, eq, gt, inArray, lte, or, isNull, sql, type SQL } from "drizzle-orm"; import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, databaseWorkScopeKey, type Db } from "@paperclipai/db"; import { prepareAdapterExecutionTargetRuntime, @@ -41,18 +41,7 @@ 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; -} +export { hasUnsettledLocalInstructionCopies } from "./agent-instruction-copy-preservation.js"; function serializeCopySave(db: Db, companyId: string, runId: string, save: () => Promise): Promise { const scope = databaseWorkScopeKey(db); diff --git a/server/src/services/ephemeral-server-contract.ts b/server/src/services/ephemeral-server-contract.ts index 78cc9c54502..ca9209eadc9 100644 --- a/server/src/services/ephemeral-server-contract.ts +++ b/server/src/services/ephemeral-server-contract.ts @@ -6,6 +6,7 @@ import { agentInstructionsBundleMode } from "./agent-instructions.js"; import { skillInstallationMaintenance } from "./skill-installation-maintenance.js"; import { hasLocalNativeHarnessObligations } from "./native-runtime/native-harness-backup-migration.js"; import { hasLocalNativeControllerObligations } from "./native-runtime/native-controller-migration.js"; +import { hasUnsettledLocalInstructionCopies } from "./agent-instruction-copy-preservation.js"; export const ephemeralServerEnabled = () => process.env.PAPERCLIP_EPHEMERAL_SERVER === "true"; @@ -49,6 +50,7 @@ export async function readEphemeralServerDataEligibility(db: Db, options: { purp and(eq(agentFileRoots.companyId, agents.companyId), eq(agentFileRoots.agentId, agents.id))) .where(isNull(agentFileRoots.agentId)).limit(1); if (missingFiles) blockers.push("agent_files_require_migration"); + if (await hasUnsettledLocalInstructionCopies(db)) blockers.push("local_instruction_copies_require_settlement"); // Only inspect configuration here; filesystem reads must never adopt an // empty directory on the replacement machine as the old server's content. const configured = await db.select({ id: agents.id, companyId: agents.companyId, name: agents.name, adapterConfig: agents.adapterConfig }).from(agents); diff --git a/server/src/services/skill-installation-maintenance.ts b/server/src/services/skill-installation-maintenance.ts index 9b9f7218ac0..2aef82362ba 100644 --- a/server/src/services/skill-installation-maintenance.ts +++ b/server/src/services/skill-installation-maintenance.ts @@ -24,6 +24,18 @@ function coversInstalledInventory(skill: typeof companySkills.$inferSelect, file return paths.has("SKILL.md") && skill.fileInventory.every(file => typeof file.path === "string" && paths.has(file.path)); } +async function legacySourceIdentity(skill: typeof companySkills.$inferSelect) { + if (!skill.sourceLocator || !path.isAbsolute(skill.sourceLocator)) return null; + const source = path.resolve(skill.sourceLocator), root = path.basename(source) === "SKILL.md" ? path.dirname(source) : source; + const stat = await fs.lstat(root).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (!stat) return null; + if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("Legacy skill root is not a safe directory"); + return { root: await fs.realpath(root), device: stat.dev, inode: stat.ino }; +} + async function legacyFiles(skill: typeof companySkills.$inferSelect, signal?: AbortSignal) { if (!skill.sourceLocator || !path.isAbsolute(skill.sourceLocator)) return null; const source = path.resolve(skill.sourceLocator); @@ -48,7 +60,7 @@ async function legacyFiles(skill: typeof companySkills.$inferSelect, signal?: Ab const handle = await fs.open(absolute, constants.O_RDONLY | constants.O_NOFOLLOW); try { const stat = await handle.stat(); - if (!stat.isFile()) throw new Error("Legacy skill contains an unsupported file"); + if (!stat.isFile() || stat.nlink !== 1) throw new Error("Legacy skill contains an unsupported file"); bytes += stat.size; if (files.length >= MAX_SKILL_SNAPSHOT_FILES || bytes > MAX_SKILL_SNAPSHOT_BYTES) throw new Error("Legacy skill exceeds package limits"); const body = Buffer.alloc(stat.size); @@ -80,7 +92,6 @@ export function skillInstallationMaintenance(db: Db, storage?: StorageService) { options.after ? gt(companySkills.id, options.after) : undefined); } async function installations(options: Conversion = {}) { - if (options.apply && !options.writersStopped) throw conflict("Installation conversion requires stopped admission and retired legacy writers (--writers-stopped)."); const skills = await db.select().from(companySkills).where(scope(options)).orderBy(asc(companySkills.id)).limit(limit(options)); const results: Array<{ skillId: string; companyId: string; status: string; versionId?: string | null; reason?: string }> = []; for (const skill of skills) { @@ -97,60 +108,89 @@ export function skillInstallationMaintenance(db: Db, storage?: StorageService) { results.push({ skillId: skill.id, companyId: skill.companyId, status: "unchanged", versionId: skill.currentVersionId }); continue; } - const pins = await protectedVersions(db, skill.companyId); - if (pins.versions === null) throw new Error("Active run lacks an exact durable skill selection; settle or snapshot it on the legacy server first"); - const saved = version ? await store.versionFiles(version) : null; - const local = await legacyFiles(skill, options.signal); - const completeSaved = saved && coversInstalledInventory(skill, saved) ? saved : null; - const files = local ?? completeSaved ?? ((skill.sourceType === "github" || skill.sourceType === "skills_sh" || skill.sourceType === "url") - ? await companySkillService(db).captureLegacyInstallation(skill.companyId, skill.id) : null); - const origin = catalogOrigin && typeof skill.metadata?.originSnapshotLocator === "string" - ? await legacyFiles({ ...skill, sourceLocator: skill.metadata.originSnapshotLocator }, options.signal) : null; - if (catalogOrigin && !origin && skill.metadata?.userModifiedAt) { - throw new Error("Catalog reset snapshot is unavailable; preserve the original installed package before conversion"); - } - if (!files?.length || !files.some(file => file.path === "SKILL.md")) throw new Error("Neither complete legacy files nor a saved version is available"); - if (!local && !coversInstalledInventory(skill, files)) { - throw new Error("Saved version does not cover the installed inventory"); - } - if (!options.apply) { - results.push({ skillId: skill.id, companyId: skill.companyId, status: "conversion_required", versionId: skill.currentVersionId }); - continue; - } - if (version && saved && !version.packageSnapshotId && skillSnapshotHash(saved) === skillSnapshotHash(files)) { - await skillPackageMaintenance(db, storage).migrateVersions({ id: version.id, apply: true, clearLegacy: true, signal: options.signal }); - } - const candidate = await content.prepare(skill.companyId, files, options.signal); - if (!candidate.packageSnapshotId || !await store.verifyDurable(skill.companyId, candidate.packageSnapshotId, options.signal)) { - throw new Error("Final snapshot is not independently recoverable from object storage"); - } - let originVersionId = typeof skill.metadata?.originVersionId === "string" ? skill.metadata.originVersionId : null; - if (catalogOrigin && origin && skillSnapshotHash(origin) !== skillSnapshotHash(files)) { - originVersionId = (await companySkillService(db).createVersion(skill.companyId, skill.id, - { label: "Original catalog package" }, { type: "system" }, { - fileInventory: origin, updateCurrentVersion: false, skipInventoryRefresh: true, - })).id; - } - const markdown = files.find(file => file.path === "SKILL.md")!.content; - const parsed = parseFrontmatterMarkdown(markdown); - const { missingSource: _missing, ...metadata } = skill.metadata ?? {}; - const result = await withSkillVersionCacheLock(db, skill.companyId, () => content.publish(skill, candidate, { - ...(legacyManaged(skill) ? { sourceType: "managed", sourceLocator: null } : {}), markdown, - name: typeof parsed.frontmatter.name === "string" ? parsed.frontmatter.name : skill.name, - description: typeof parsed.frontmatter.description === "string" ? parsed.frontmatter.description : skill.description, - fileInventory: files.map(({ path, kind }) => ({ path, kind })), - trustLevel: files.some(file => file.kind === "script" || file.executable) ? "scripts_executables" - : files.some(file => file.kind === "asset" || file.kind === "other") ? "assets" : "markdown_only", - metadata: { ...metadata, ...(legacyManaged(skill) ? { sourceKind: "managed" } : {}), - ...(originVersionId ? { originVersionId } : {}) }, - }, { type: "system" }, { label: "Legacy installation conversion", - reuseCurrentVersion: Boolean(saved && skillSnapshotHash(saved) === skillSnapshotHash(files)), - afterPublish: async (tx, row, version) => { - if (catalogOrigin && !originVersionId) await tx.update(companySkills) - .set({ metadata: { ...row.metadata, originVersionId: version.id } }).where(eq(companySkills.id, row.id)); - }, - })); - results.push({ skillId: skill.id, companyId: skill.companyId, status: "converted", versionId: result.version.id }); + // Conversion copies current content. Keep old directories, versions and + // inline bytes; unknown historical pins still block retention/eviction. + const convert = async () => { + const saved = version ? await store.versionFiles(version) : null; + const sourceIdentity = await legacySourceIdentity(skill); + const local = await legacyFiles(skill, options.signal); + const completeSaved = saved && coversInstalledInventory(skill, saved) ? saved : null; + const files = local ?? completeSaved ?? ((skill.sourceType === "github" || skill.sourceType === "skills_sh" || skill.sourceType === "url") + ? await companySkillService(db).captureLegacyInstallation(skill.companyId, skill.id) : null); + const originSource = catalogOrigin && typeof skill.metadata?.originSnapshotLocator === "string" + ? { ...skill, sourceLocator: skill.metadata.originSnapshotLocator } : null; + const originIdentity = originSource ? await legacySourceIdentity(originSource) : null; + const origin = originSource ? await legacyFiles(originSource, options.signal) : null; + if (catalogOrigin && !origin && skill.metadata?.userModifiedAt) { + throw new Error("Catalog reset snapshot is unavailable; preserve the original installed package before conversion"); + } + if (!files?.length || !files.some(file => file.path === "SKILL.md")) throw new Error("Neither complete legacy files nor a saved version is available"); + if (!local && !coversInstalledInventory(skill, files)) { + throw new Error("Saved version does not cover the installed inventory"); + } + if (!options.apply) { + return { skillId: skill.id, companyId: skill.companyId, status: "conversion_required", versionId: skill.currentVersionId }; + } + if (version && saved && !version.packageSnapshotId && !version.repositorySnapshotId + && skillSnapshotHash(saved) === skillSnapshotHash(files)) { + await skillPackageMaintenance(db, storage).migrateVersions({ id: version.id, apply: true, clearLegacy: false, signal: options.signal }); + } + const candidate = await content.prepare(skill.companyId, files, options.signal); + if (!candidate.packageSnapshotId || !await store.verifyDurable(skill.companyId, candidate.packageSnapshotId, options.signal)) { + throw new Error("Final snapshot is not independently recoverable from object storage"); + } + let originVersionId = typeof skill.metadata?.originVersionId === "string" ? skill.metadata.originVersionId : null; + const originCandidate = catalogOrigin && origin && skillSnapshotHash(origin) !== skillSnapshotHash(files) + ? await content.prepare(skill.companyId, origin, options.signal) : null; + if (originCandidate && !originCandidate.packageSnapshotId) throw new Error("Catalog origin is not durably available"); + const markdown = files.find(file => file.path === "SKILL.md")!.content; + const parsed = parseFrontmatterMarkdown(markdown); + const { missingSource: _missing, ...metadata } = skill.metadata ?? {}; + // Object preparation is private. Hold the shared cache lock only for + // publication so new run pinning does not wait for network uploads. + const result = await withSkillVersionCacheLock(db, skill.companyId, () => content.publish(skill, candidate, { + ...(legacyManaged(skill) ? { sourceType: "managed", sourceLocator: null } : {}), markdown, + name: typeof parsed.frontmatter.name === "string" ? parsed.frontmatter.name : skill.name, + description: typeof parsed.frontmatter.description === "string" ? parsed.frontmatter.description : skill.description, + fileInventory: files.map(({ path, kind }) => ({ path, kind })), + trustLevel: files.some(file => file.kind === "script" || file.executable) ? "scripts_executables" + : files.some(file => file.kind === "asset" || file.kind === "other") ? "assets" : "markdown_only", + metadata: { ...metadata, ...(legacyManaged(skill) ? { sourceKind: "managed" } : {}), + ...(originVersionId ? { originVersionId } : {}) }, + }, { type: "system" }, { label: "Legacy installation conversion", + reuseCurrentVersion: Boolean(saved && skillSnapshotHash(saved) === skillSnapshotHash(files)), + validate: async tx => { + // publish already holds the canonical slug and skill-row locks. + // Recheck filesystem writers after upload, before changing authority. + const [current] = await tx.select().from(companySkills).where(and( + eq(companySkills.companyId, skill.companyId), eq(companySkills.id, skill.id))); + if (!current || current.slug !== skill.slug || current.sourceLocator !== skill.sourceLocator + || current.sourceRef !== skill.sourceRef || JSON.stringify(current.fileInventory) !== JSON.stringify(skill.fileInventory) + || JSON.stringify(await legacySourceIdentity(current)) !== JSON.stringify(sourceIdentity)) + throw conflict("Legacy skill source changed during conversion. Retry the operation."); + const latest = await legacyFiles(current, options.signal); + if ((latest ? skillSnapshotHash(latest) : null) !== (local ? skillSnapshotHash(local) : null)) + throw conflict("Legacy skill files changed during conversion. Retry the operation."); + if (originSource && (JSON.stringify(await legacySourceIdentity(originSource)) !== JSON.stringify(originIdentity) + || (await legacyFiles(originSource, options.signal).then(value => value ? skillSnapshotHash(value) : null)) + !== (origin ? skillSnapshotHash(origin) : null))) + throw conflict("Catalog origin changed during conversion. Retry the operation."); + if (origin && originCandidate?.packageSnapshotId) { + originVersionId = (await companySkillService(db).createVersion(skill.companyId, skill.id, + { label: "Original catalog package" }, { type: "system" }, { + database: tx, fileInventory: origin, packageSnapshotId: originCandidate.packageSnapshotId, + packageFileMap: originCandidate.packageFileMap, updateCurrentVersion: false, skipInventoryRefresh: true, + })).id; + } + }, + afterPublish: async (tx, row, version) => { + if (catalogOrigin) await tx.update(companySkills) + .set({ metadata: { ...row.metadata, originVersionId: originVersionId ?? version.id } }).where(eq(companySkills.id, row.id)); + }, + })); + return { skillId: skill.id, companyId: skill.companyId, status: "converted", versionId: result.version.id }; + }; + results.push(await convert()); } catch (error) { results.push({ skillId: skill.id, companyId: skill.companyId, status: "blocked", reason: error instanceof Error ? error.message : "Conversion failed" }); }