fix(skills): read cached versions and cancel conversion reads

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/skill-package-store.test.ts

@@ -438,6 +438,22 @@ describe.skipIf(!support.supported)("external skill packages", () => {
     expect((await f.resolve()).every(entry => entry.sourceStatus === "available")).toBe(true);
   });
 
+  it("reads verified cached versions offline while independently checking durable bytes and rejecting corrupt files", async () => {
+    const f = await fixture();
+    await f.store.files(f.companyId, f.id);
+    const read = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("storage offline"));
+    const saved = f.skills[0]!;
+    expect((await f.service.getVersion(f.companyId, saved.skill.id, saved.version.id))!.fileInventory)
+      .toEqual([{ path: "SKILL.md", kind: "skill", content: files[0]!.content }]);
+    expect(await f.store.versionFiles({ ...saved.version, packageFileMap: null })).toEqual(files);
+    expect(read).not.toHaveBeenCalled();
+    await expect(f.store.verifyDurable(f.companyId, f.id)).rejects.toThrow("storage offline");
+    const target = path.join(skillPackagePaths(f.companyId, f.id).directory, files[0]!.path);
+    await fs.chmod(target, 0o600);
+    await fs.writeFile(target, "x".repeat(Buffer.byteLength(files[0]!.content)));
+    await expect(f.store.versionFiles(saved.version)).rejects.toThrow("storage offline");
+  });
+
   it("retries expired upload claims with fresh keys and only deletes the abandoned attempt", async () => {
     const companyId = await company();
     const store = skillPackageStore(db, storage);

server/src/services/skill-package-maintenance.ts

@@ -57,9 +57,9 @@ export function skillPackageMaintenance(db: Db, storage?: StorageService) {
       if (row.repositorySnapshotId) await migrateSnapshots({ id: row.repositorySnapshotId, apply: true, clearLegacy: true, signal: options.signal });
       let packageId = row.packageSnapshotId;
       let fileMap = row.packageFileMap;
-      if (packageId && skillSnapshotHash(await store.versionFiles(row)) !== skillSnapshotHash(files)) throw new Error(`Legacy version differs from its package: ${row.id}`);
+      if (packageId && skillSnapshotHash(await store.versionFiles(row, true, options.signal)) !== skillSnapshotHash(files)) throw new Error(`Legacy version differs from its package: ${row.id}`);
       if (!packageId && row.repositorySnapshotId && row.repositorySkillPath) {
-        const manifest = await store.manifest(row.companyId, row.repositorySnapshotId);
+        const manifest = await store.manifest(row.companyId, row.repositorySnapshotId, options.signal);
         const byPath = new Map(manifest.files.map(file => [file.path, file]));
         const mappings = files.map(file => ({ path: file.path, kind: file.kind, encoding: file.encoding ?? "utf8" as const,
           sourcePath: file.path === "SKILL.md" ? row.repositorySkillPath! : path.posix.join(path.posix.dirname(row.repositorySkillPath!), file.path) }));
@@ -71,7 +71,7 @@ export function skillPackageMaintenance(db: Db, storage?: StorageService) {
       packageId ??= await store.prepare(row.companyId, files, true, options.signal);
       const legacyMetadata = new Map(files.map(file => [file.path, file]));
       fileMap = fileMap ? fileMap.map(file => ({ ...file, kind: legacyMetadata.get(file.path)?.kind ?? file.kind,
-        encoding: legacyMetadata.get(file.path)?.encoding ?? "utf8" })) : skillVersionFileMap(files, await store.manifest(row.companyId, packageId));
+        encoding: legacyMetadata.get(file.path)?.encoding ?? "utf8" })) : skillVersionFileMap(files, await store.manifest(row.companyId, packageId, options.signal));
       const receipt = await store.descriptor(row.companyId, packageId);
       if (!receipt.storage || !await store.verifyDurable(row.companyId, packageId, options.signal)) throw new Error(`Package must be migrated and verified first: ${packageId}`);
       await db.update(versions).set({ packageSnapshotId: packageId, packageFileMap: fileMap, ...(options.clearLegacy ? { fileInventory: [], migrationAt: null } : {}) })

server/src/services/skill-package-store.ts

@@ -182,12 +182,14 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     if (legacy.storage) { row.storage = legacy.storage; row.upload = legacy.upload; }
     return legacy.files;
   }
-  async function manifestFor(row: Descriptor): Promise<SkillPackageManifest> {
+  async function manifestFor(row: Descriptor, cancellation?: AbortSignal): Promise<SkillPackageManifest> {
+    cancellation?.throwIfAborted();
     const paths = skillPackagePaths(row.companyId, row.id);
     const key = `${paths.manifest}:${row.storage?.manifestSha256 ?? row.contentHash}`;
     const cached = manifests.get(key);
     if (cached) return cached.manifest;
-    return flight(db, `manifest:${key}`, async () => {
+    const load = async () => {
+      const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000);
       await ensureSkillCacheDirectory(paths.parent, true);
       // Legacy manifests have no independently stored checksum. Derive them once
       // per process from the authoritative row until the operator migrates it.
@@ -205,12 +207,12 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
       }
       if (row.storage) {
         if (!bytes || packageSha256(bytes) !== row.storage.manifestSha256) {
-          const signal = AbortSignal.timeout(180_000);
           try {
             if (row.storage.provider !== storage().provider) throw new Error("Skill package storage provider changed");
             bytes = await withPackageIo(() => readBoundedObject(storage(), row.companyId, row.storage!.manifestKey,
               row.storage!.manifestBytes, row.storage!.manifestSha256, signal), signal);
           } catch (error) {
+            signal.throwIfAborted();
             const [legacy] = await db.select({ files: snapshots.fileInventory }).from(snapshots).where(scope(row.companyId, row.id));
             const saved = legacy?.files.length ? skillPackageManifestBytes(skillPackageManifest(legacy.files)) : null;
             if (saved && packageSha256(saved) === row.storage!.manifestSha256) bytes = saved;
@@ -222,9 +224,10 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
         bytes = skillPackageManifestBytes((await restoreRecipe(row)).manifest);
       } else if (!bytes) {
         const files = await legacyFiles(row);
-        if (row.storage) return manifestFor(row);
+        if (row.storage) return manifestFor(row, cancellation);
         bytes = skillPackageManifestBytes(skillPackageManifest(files));
       }
+      signal.throwIfAborted();
       if (bytes.length > MAX_SKILL_MANIFEST_BYTES) throw new Error("Skill manifest exceeds its byte bound");
       const manifest = validateSkillPackageManifest(JSON.parse(bytes.toString("utf8")));
       if (row.storage && (manifest.contentHash !== row.contentHash || manifest.files.length !== row.storage.fileCount
@@ -235,7 +238,10 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
       finally { await fs.rm(temp, { force: true }); }
       rememberManifest(key, manifest, bytes.length);
       return manifest;
-    });
+    };
+    // Optional work must neither inherit an uncancellable interactive read nor
+    // cancel a shared read that another caller still needs.
+    return cancellation ? load() : flight(db, `manifest:${key}`, load);
   }
   async function restoreInto(row: Descriptor, manifest: SkillPackageManifest, staging: string,
     archivePath: string, signal: AbortSignal, allowRecovery = true) {
@@ -279,10 +285,11 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
   // Admin reads and retention verification own a scratch directory. They never
   // wait for or hold the lock used to prepare and pin an agent's warm files.
   async function withIsolatedPackage<T>(companyId: string, id: string,
-    consume: (root: string, manifest: SkillPackageManifest) => Promise<T>, allowRecovery = true) {
+    consume: (root: string, manifest: SkillPackageManifest) => Promise<T>, allowRecovery = true, cancellation?: AbortSignal) {
+    const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000);
+    signal.throwIfAborted();
     const row = await descriptor(companyId, id);
-    const manifest = await manifestFor(row);
-    const signal = AbortSignal.timeout(180_000);
+    const manifest = await manifestFor(row, signal);
     return withPackageIo(async () => {
       const parent = skillPackagePaths(companyId, id).parent;
       await ensureSkillCacheDirectory(parent, true);
@@ -433,10 +440,11 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
   async function verifyDurable(companyId: string, id: string, cancellation?: AbortSignal,
     validate?: (root: string, manifest: SkillPackageManifest) => Promise<boolean>): Promise<boolean> {
     const action = async () => {
+      cancellation?.throwIfAborted();
       const row = await descriptor(companyId, id);
       if (!row.storage) {
-        await manifestFor(row);
-        if (!row.storage) return validate ? withIsolatedPackage(companyId, id, validate, false) : true;
+        await manifestFor(row, cancellation);
+        if (!row.storage) return validate ? withIsolatedPackage(companyId, id, validate, false, cancellation) : true;
       }
       if (row.storage.provider !== storage().provider) throw new Error("Skill package storage provider changed; migrate its objects before switching storage");
       const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000);
@@ -464,18 +472,41 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     };
     const context = readScope.getStore();
     const key = `${companyId}:${id}`;
-    if (context?.db !== db || validate) return action();
+    if (context?.db !== db || validate || cancellation) return action();
     if (!context.durable.has(key)) context.durable.set(key, action());
     return context.durable.get(key)!;
   }
-  async function versionFiles(version: StoredSkillVersion, allowRecovery = true): Promise<CompanySkillVersionFileInventoryEntry[]> {
+  async function readPackageFile(root: string, entry: SkillPackageManifest["files"][number], signal?: AbortSignal) {
+    signal?.throwIfAborted();
+    const filename = path.join(root, entry.path);
+    await ensureSkillCacheDirectory(path.dirname(filename));
+    const handle = await fs.open(filename, constants.O_RDONLY | constants.O_NOFOLLOW);
+    try {
+      const stat = await handle.stat();
+      if (!stat.isFile() || stat.size !== entry.size) throw new Error("Invalid cached skill file");
+      const bytes = Buffer.alloc(entry.size);
+      let offset = 0;
+      while (offset < bytes.length) {
+        signal?.throwIfAborted();
+        const { bytesRead } = await handle.read(bytes, offset, bytes.length - offset, offset);
+        if (!bytesRead) throw new Error("Invalid cached skill file");
+        offset += bytesRead;
+      }
+      if (packageSha256(bytes) !== entry.sha256) throw new Error("Invalid cached skill file");
+      return bytes;
+    } finally { await handle.close(); }
+  }
+  async function versionFiles(version: StoredSkillVersion, allowRecovery = true, cancellation?: AbortSignal): Promise<CompanySkillVersionFileInventoryEntry[]> {
+    cancellation?.throwIfAborted();
     if (!version.packageSnapshotId) {
       const [row] = await db.select({ ...skillVersionMetadataColumns, files: companySkillVersions.fileInventory }).from(companySkillVersions).where(and(
         eq(companySkillVersions.companyId, version.companyId), eq(companySkillVersions.id, version.id)));
-      if (row?.packageSnapshotId) return versionFiles(row, allowRecovery);
+      if (row?.packageSnapshotId) return versionFiles(row, allowRecovery, cancellation);
       return row?.files ?? [];
     }
-    return withIsolatedPackage(version.companyId, version.packageSnapshotId, async (root, manifest) => {
+    const id = version.packageSnapshotId;
+    const manifest = await manifestFor(await descriptor(version.companyId, id), cancellation);
+    const read = async (root: string) => {
       const byPath = new Map(manifest.files.map(file => [file.path, file]));
       const result: CompanySkillVersionFileInventoryEntry[] = [];
       for (const mapping of version.packageFileMap ?? manifest.files.map(file => ({ ...file, sourcePath: file.path }))) {
@@ -485,11 +516,22 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
         if (!file) throw new Error("Skill version references a missing package file");
         const encoding = mapping.encoding ?? file.encoding;
         result.push({ path: mapping.path, kind: mapping.kind,
-          content: (await fs.readFile(path.join(root, file.path))).toString(encoding),
+          content: (await readPackageFile(root, file, cancellation)).toString(encoding),
           ...(encoding === "base64" ? { encoding } : {}), ...(file.executable ? { executable: true } : {}) });
       }
       return result;
-    }, allowRecovery);
+    };
+    // Verify bytes as they are read, without holding the warm preparation lock.
+    // A descriptor survives eviction; a missing/racing file falls back to storage.
+    const root = skillPackagePaths(version.companyId, id).directory;
+    try {
+      const result = await read(root);
+      touchSkillPackage(root);
+      return result;
+    } catch {
+      cancellation?.throwIfAborted();
+      return withIsolatedPackage(version.companyId, id, read, allowRecovery, cancellation);
+    }
   }
   async function readVersionFile(version: StoredSkillVersion, relativePath: string, allowRecovery = true) {
     assertSkillSnapshotPath(relativePath);
@@ -501,24 +543,10 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     const entry = manifest.files.find(file => file.path === (mapping?.sourcePath ?? relativePath));
     if (!entry) return undefined;
     const read = async (root: string) => {
-      const filename = path.join(root, entry.path);
-      await ensureSkillCacheDirectory(path.dirname(filename));
-      const handle = await fs.open(filename, constants.O_RDONLY | constants.O_NOFOLLOW);
-      try {
-        const stat = await handle.stat();
-        if (!stat.isFile() || stat.size !== entry.size) throw new Error("Invalid cached skill file");
-        const bytes = Buffer.alloc(entry.size);
-        let offset = 0;
-        while (offset < bytes.length) {
-          const { bytesRead } = await handle.read(bytes, offset, bytes.length - offset, offset);
-          if (!bytesRead) throw new Error("Invalid cached skill file");
-          offset += bytesRead;
-        }
-        if (packageSha256(bytes) !== entry.sha256) throw new Error("Invalid cached skill file");
-        const encoding = mapping?.encoding ?? entry.encoding;
-        return { path: relativePath, kind: mapping?.kind ?? entry.kind, content: bytes.toString(encoding),
-          encoding, executable: entry.executable };
-      } finally { await handle.close(); }
+      const bytes = await readPackageFile(root, entry);
+      const encoding = mapping?.encoding ?? entry.encoding;
+      return { path: relativePath, kind: mapping?.kind ?? entry.kind, content: bytes.toString(encoding),
+        encoding, executable: entry.executable };
     };
     // Immutable cache files can be read without the preparation lock. An open
     // descriptor survives eviction; missing/racing files fall back to storage.
@@ -538,5 +566,5 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     return row?.files ?? [];
   }
   return { descriptor, prepare, bindRestore, files, version, prefetch, versionFiles, versionInventory, readVersionFile, withPackage, withIsolatedPackage, verifyDurable,
-    manifest: async (companyId: string, id: string) => manifestFor(await descriptor(companyId, id)) };
+    manifest: async (companyId: string, id: string, cancellation?: AbortSignal) => manifestFor(await descriptor(companyId, id), cancellation) };
 }

server/src/services/skill-package-work.test.ts

@@ -4,7 +4,7 @@ import os from "node:os";
 import path from "node:path";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { eq, sql } from "drizzle-orm";
-import { companies, companySkillRepositorySnapshots as snapshots, skillPackageDeletions, createDb } from "@paperclipai/db";
+import { companies, companySkills, companySkillVersions as versions, companySkillRepositorySnapshots as snapshots, skillPackageDeletions, createDb } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
 import { createStorageService } from "../storage/service.js";
 import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
@@ -103,12 +103,47 @@ describe.skipIf(!support.supported)("automatic skill package work", () => {
   it("reports a safe conversion failure reason without logging the legacy payload", async () => {
     const companyId = await company();
     const legacy = [{ ...files[0]!, content: "private payload" }];
-    const [row] = await db.insert(snapshots).values({ companyId, contentHash: skillSnapshotHash(files), fileInventory: legacy }).returning();
+    const [row] = await db.insert(snapshots).values({ companyId, contentHash: skillSnapshotHash(files), fileInventory: legacy, migrationAt: new Date(0) }).returning();
     const warning = vi.spyOn(logger, "warn").mockImplementation(() => {});
     await skillPackageWork(db, storage).tick(new AbortController().signal);
     expect(warning).toHaveBeenCalledWith({ packageId: row!.id, failure: { reason: "legacy_hash_mismatch", errorCode: "unknown" } }, expect.any(String));
     expect(JSON.stringify(warning.mock.calls)).not.toContain("private payload");
   });
+  it.each(["manifest.json", "files.tar.gz"])("stops conversion during a stalled %s read and retains legacy bytes", async object => {
+    const companyId = await company();
+    const packageId = await skillPackageStore(db, storage).prepare(companyId, files);
+    const [skill] = await db.insert(companySkills).values({ companyId, key: "legacy", slug: "legacy", name: "Legacy", markdown: files[0]!.content }).returning();
+    const [version] = await db.insert(versions).values({ companyId, companySkillId: skill!.id, revisionNumber: 1, fileInventory: files, migrationAt: new Date(0),
+      ...(object === "manifest.json" ? { repositorySnapshotId: packageId, repositorySkillPath: "SKILL.md" } : { packageSnapshotId: packageId }) }).returning();
+    const get = storage.getObject.bind(storage);
+    let signalStarted!: () => void;
+    const started = new Promise<void>(resolve => { signalStarted = resolve; });
+    let pendingSignal: AbortSignal | undefined;
+    let releasePending: () => void = () => {};
+    vi.spyOn(storage, "getObject").mockImplementation(async (companyId, key, options) => {
+      if (!key.endsWith(object)) return get(companyId, key, options);
+      pendingSignal = options?.signal;
+      signalStarted();
+      return new Promise<never>((_resolve, reject) => {
+        releasePending = () => reject(new Error("Test released stalled read"));
+        pendingSignal?.addEventListener("abort", () => reject(pendingSignal!.reason), { once: true });
+      });
+    });
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: () => {} });
+    skillPackageWork(db, storage).start(coordinator);
+    await started;
+    const stopping = coordinator.stop();
+    let timeout: ReturnType<typeof setTimeout> | undefined;
+    try {
+      await Promise.race([stopping, new Promise<never>((_resolve, reject) => {
+        timeout = setTimeout(() => reject(new Error("Conversion did not stop promptly")), 1_000);
+      })]);
+      expect(pendingSignal?.aborted).toBe(true);
+      const [saved] = await db.select().from(versions).where(eq(versions.id, version!.id));
+      expect(saved!.fileInventory).toEqual(files);
+      expect(saved!.migrationAt).not.toBeNull();
+    } finally { clearTimeout(timeout); releasePending(); await stopping; }
+  });
   it.each([
     ["deleteObject", "stop"], ["headObject", "stop"], ["deleteObject", "deadline"], ["headObject", "deadline"],
   ] as const)("releases a stalled %s on %s and retains its deletion receipt", async (method, cancellation) => {