fix(skills): reuse restored packages and share manifest reads

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

doc/skill-package-storage.md

@@ -109,6 +109,9 @@ pending delete/existence requests; incomplete cleanup retains the receipt for re
   Cached version inventories and file previews have a separate two-reader, 32-waiter
   bound, so slow storage cannot prevent verified local reads. Both queues honor deadlines
   and cancellation; an overloaded local-read queue does not trigger a storage download.
+  Identical cold version/file reads are coalesced. Verified scratch restores repopulate
+  the shared cache under a short, nonblocking publication lock, so subsequent reads
+  reuse local bytes. Eviction verification runs outside that publication lock.
   Packages share the Git scanner's limits of 10,000 files and 100 MiB decoded bytes.
   Recovery splits reads within the Git reader's 9,990-file and 61 MiB batch limits.
   Individual manifests are limited to 48 MiB; the manifest cache holds 16 MiB of serialized data.

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

@@ -454,6 +454,98 @@ describe.skipIf(!support.supported)("external skill packages", () => {
     await expect(f.store.versionFiles(saved.version)).rejects.toThrow("storage offline");
   });
 
+  it("coalesces cold version reads and retains the verified package for subsequent reads", async () => {
+    const f = await fixture();
+    await f.resolve();
+    const root = skillPackagePaths(f.companyId, f.id).directory;
+    await fs.rm(root, { recursive: true });
+    const read = vi.spyOn(storage, "getObject");
+    const saved = f.skills[0]!;
+    const get = () => f.service.getVersion(f.companyId, saved.skill.id, saved.version.id);
+    const [first, second, third] = await Promise.all([get(), get(), get()]);
+    expect(first!.fileInventory).toEqual([{ path: "SKILL.md", kind: "skill", content: files[0]!.content }]);
+    expect(second).toEqual(first);
+    expect(third).toEqual(first);
+    expect(await get()).toEqual(first);
+    expect(await f.store.readVersionFile(saved.version, "SKILL.md")).toMatchObject({ content: files[0]!.content });
+    expect(read.mock.calls.filter(call => call[1].endsWith("files.tar.gz"))).toHaveLength(1);
+    expect(await verifySkillPackageDirectory(root, skillPackageManifest(files).files)).toBe(true);
+  });
+
+  it("shares a cold manifest while allowing one caller to cancel without aborting another", async () => {
+    const f = await fixture();
+    const get = storage.getObject.bind(storage);
+    let entered!: () => void;
+    const started = new Promise<void>(resolve => { entered = resolve; });
+    let release!: () => void;
+    const blocked = new Promise<void>(resolve => { release = resolve; });
+    let storageSignal: AbortSignal | undefined;
+    const read = vi.spyOn(storage, "getObject").mockImplementation(async (...args) => {
+      storageSignal = args[2]?.signal;
+      entered();
+      await blocked;
+      return get(...args);
+    });
+    const cancellation = new AbortController();
+    const first = f.store.manifest(f.companyId, f.id, cancellation.signal).catch(error => error);
+    await started;
+    const second = f.store.manifest(f.companyId, f.id);
+    try {
+      // The shared one-slot pool places this query after the second descriptor.
+      await db.select({ id: snapshots.id }).from(snapshots).where(eq(snapshots.id, f.id));
+      cancellation.abort(new Error("Caller stopped"));
+      expect((await first).message).toBe("Caller stopped");
+      expect(storageSignal?.aborted).toBe(false);
+      release();
+      expect(await second).toEqual(skillPackageManifest(files));
+      expect(read).toHaveBeenCalledOnce();
+    } finally { release(); await Promise.allSettled([first, second]); }
+  });
+
+  it("keeps warm preparation responsive while a read-triggered cache trim verifies storage", async () => {
+    const f = await fixture();
+    await f.resolve();
+    const paths = skillPackagePaths(f.companyId, f.id);
+    // Exceed the directory bound without manufacturing large file payloads.
+    for (let index = 0; index < 63; index++) {
+      const orphan = path.join(paths.parent, randomUUID());
+      await fs.mkdir(orphan);
+      const later = new Date(Date.now() + 60_000);
+      await fs.utimes(orphan, later, later);
+    }
+    const changed = files.map((file, index) => index ? file : { ...file, content: `${file.content}Updated\n` });
+    const id = await f.store.prepare(f.companyId, changed);
+    const [version] = await db.insert(companySkillVersions).values({ companyId: f.companyId,
+      companySkillId: f.skills[0]!.skill.id, revisionNumber: 2, packageSnapshotId: id, migrationAt: null,
+      packageFileMap: f.skills[0]!.version.packageFileMap }).returning();
+    const receipt = (await f.store.descriptor(f.companyId, f.id)).storage!;
+    const get = storage.getObject.bind(storage);
+    let entered!: () => void;
+    const started = new Promise<void>(resolve => { entered = resolve; });
+    let release!: () => void;
+    const blocked = new Promise<void>(resolve => { release = resolve; });
+    vi.spyOn(storage, "getObject").mockImplementation(async (...args) => {
+      if (args[1] === receipt.manifestKey) { entered(); await blocked; }
+      return get(...args);
+    });
+    const reading = f.store.versionFiles(version!);
+    let warming: ReturnType<typeof f.resolve> | undefined;
+    let timeout: ReturnType<typeof setTimeout> | undefined;
+    try {
+      await started;
+      warming = f.resolve();
+      expect((await Promise.race([warming, new Promise<never>((_resolve, reject) => {
+        timeout = setTimeout(() => reject(new Error("Warm preparation waited for eviction storage")), 1_000);
+      })])).every(entry => entry.sourceStatus === "available")).toBe(true);
+      // Preparation can reuse its separate projection. Also access the shared
+      // package itself to prove eviction rechecks use after verification.
+      await f.store.versionFiles(f.skills[0]!.version);
+      release();
+      expect((await reading)[0]!.content).toBe(changed[0]!.content);
+      expect((await fs.stat(paths.directory)).isDirectory()).toBe(true);
+    } finally { clearTimeout(timeout); release(); await Promise.allSettled([reading, warming]); }
+  });
+
   it("bounds concurrent cached inventories while leaving warm turn preparation responsive", async () => {
     const f = await fixture();
     await f.resolve();

server/src/services/company-skills.ts

@@ -2979,7 +2979,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
   const packages = skillPackageStore(packageDb);
   const packageRuntime = skillPackageRuntime(packageDb);
   // Admin reads keep SQL on the caller's transaction, which may own the pool's only slot.
-  const packageReader = db === packageDb ? packages : skillPackageStore(db);
+  const packageReader = db === packageDb ? packages : skillPackageStore(db, undefined, packageDb);
   const visibility = (database: DbOrTransaction = db) => skillAccessService(database as Db, options.actor);
   const folderSvc = folderService(db);
   const agents = agentService(db);

server/src/services/customer-success-inspection.ts

@@ -292,7 +292,7 @@ export async function readCustomerSuccessResource(
             String(versionId),
           );
           if (version.companySkillId !== skill.id) throw notFound("Skill version not found");
-          const file = await skillPackageStore(tx, storage).readVersionFile(
+          const file = await skillPackageStore(tx, storage, db).readVersionFile(
             version as typeof schema.companySkillVersions.$inferSelect, relative, false);
           if (file?.content !== undefined) {
             const bytes = Buffer.from(file.content, file.encoding === "base64" ? "base64" : "utf8");

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

@@ -15,7 +15,8 @@ export function touchSkillPackage(directory: string) {
 /** Called only after a cold fill, under the company cache lock. Never on a warm
  * check. Version projections and run pins have their own existing retention. */
 export async function trimSkillPackageCache(db: Db, companyId: string, directory: string,
-  verifyDurable: (id: string, signal: AbortSignal) => Promise<boolean>) {
+  verifyDurable: (id: string, signal: AbortSignal) => Promise<boolean>,
+  removeVerified: (action: () => Promise<void>) => Promise<void> = action => action()) {
   const parent = path.dirname(directory);
   await ensureSkillCacheDirectory(parent);
   const contents = await fs.readdir(parent, { withFileTypes: true });
@@ -24,20 +25,30 @@ export async function trimSkillPackageCache(db: Db, companyId: string, directory
     const target = path.join(parent, entry.name);
     // Every I/O attempt has a three-minute deadline; leave a much longer crash
     // grace so housekeeping cannot race a live publisher in another process.
-    if (Date.now() - (await fs.lstat(target)).mtimeMs > 60 * 60_000) await fs.rm(target, { recursive: true, force: true });
+    const stat = await fs.lstat(target).catch((error: NodeJS.ErrnoException) => {
+      if (error.code !== "ENOENT") throw error;
+      return null;
+    });
+    if (stat && Date.now() - stat.mtimeMs > 60 * 60_000) await fs.rm(target, { recursive: true, force: true });
   }
   const names = contents.filter(entry => entry.isDirectory() && /^[a-f0-9-]{36}$/i.test(entry.name)).map(entry => entry.name);
   if (!names.length) return;
   const rows = await db.select({ id: snapshots.id, storage: snapshots.storage }).from(snapshots)
     .where(and(eq(snapshots.companyId, companyId), inArray(snapshots.id, names)));
   const byId = new Map(rows.map(row => [row.id, row.storage]));
-  const entries = await Promise.all(names.map(async id => {
+  const candidates = await Promise.all(names.map(async id => {
     const location = path.join(parent, id);
     const receipt = byId.get(id);
-    return { id, location, receipt, lastUsed: used.get(location) ?? (await fs.lstat(location)).mtimeMs,
+    const stat = await fs.lstat(location).catch((error: NodeJS.ErrnoException) => {
+      if (error.code !== "ENOENT") throw error;
+      return null;
+    });
+    if (!stat) return null;
+    return { id, location, receipt, lastUsed: used.get(location) ?? stat.mtimeMs,
       // Include an upper bound on block rounding and directory/manifest overhead.
       bytes: receipt ? receipt.totalBytes + receipt.fileCount * 8192 : 64 * 1024 * 1024 };
   }));
+  const entries = candidates.filter(entry => entry !== null);
   let bytes = entries.reduce((total, entry) => total + entry.bytes, 0);
   let count = entries.length;
   const signal = AbortSignal.timeout(10_000);
@@ -50,9 +61,15 @@ export async function trimSkillPackageCache(db: Db, companyId: string, directory
     // A storage outage must not discard the last recoverable copy. Keep the
     // excess until the next cold fill can confirm a durable copy independently.
     if (byId.has(entry.id) && !await verifyDurable(entry.id, signal)) break;
-    await fs.rm(entry.location, { recursive: true, force: true });
+    let removed = false;
-    await fs.rm(path.join(parent, `${entry.id}.json`), { force: true });
+    await removeVerified(async () => {
-    used.delete(entry.location);
+      if ((used.get(entry.location) ?? entry.lastUsed) > entry.lastUsed) return;
+      await fs.rm(entry.location, { recursive: true, force: true });
+      await fs.rm(path.join(parent, `${entry.id}.json`), { force: true });
+      used.delete(entry.location);
+      removed = true;
+    });
+    if (!removed) continue;
     bytes -= entry.bytes;
     count--;
   }

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

@@ -13,7 +13,7 @@ import { getStorageService, type StorageService } from "../storage/index.js";
 import { resolvePaperclipInstanceRoot } from "../home-paths.js";
 import { conflict, notFound } from "../errors.js";
 import { assertSkillSnapshotPath, skillFileBytes } from "./skill-snapshot.js";
-import { ensureSkillCacheDirectory, withSkillVersionCacheLock } from "./skill-version-cache.js";
+import { ensureSkillCacheDirectory, SkillCacheBusy, withSkillVersionCacheLock } from "./skill-version-cache.js";
 import { boundedHashStream, extractSkillPackageArchive, MAX_SKILL_MANIFEST_BYTES, packageSha256,
   skillPackageManifest, skillPackageManifestBytes, validateSkillPackageManifest, verifySkillPackageDirectory, relocateSkillPackageVerification, writeSkillPackageArchive,
   type SkillPackageManifest } from "./skill-package-files.js";
@@ -50,6 +50,45 @@ function flight<T>(db: Db, key: string, action: () => Promise<T>): Promise<T> {
   pending.set(key, work);
   return work;
 }
+type SharedFlight = { work: Promise<unknown>; controller: AbortController; waiters: number };
+const sharedFlights = new WeakMap<Db, Map<string, SharedFlight>>();
+function sharedFlight<T>(db: Db, key: string, cancellation: AbortSignal | undefined, action: (signal: AbortSignal) => Promise<T>): Promise<T> {
+  cancellation?.throwIfAborted();
+  let pending = sharedFlights.get(db);
+  if (!pending) { pending = new Map(); sharedFlights.set(db, pending); }
+  let entry = pending.get(key);
+  if (!entry) {
+    const controller = new AbortController();
+    entry = { controller, waiters: 0, work: Promise.resolve() };
+    const created = entry;
+    entry.work = action(controller.signal).finally(() => {
+      if (pending!.get(key) === created) pending!.delete(key);
+    });
+    pending.set(key, entry);
+  }
+  const current = entry;
+  current.waiters++;
+  return new Promise<T>((resolve, reject) => {
+    let settled = false;
+    const release = () => {
+      if (settled) return false;
+      settled = true;
+      cancellation?.removeEventListener("abort", aborted);
+      current.waiters--;
+      return true;
+    };
+    const aborted = () => {
+      if (!release()) return;
+      if (current.waiters) { reject(cancellation!.reason); return; }
+      if (pending!.get(key) === current) pending!.delete(key);
+      current.controller.abort(cancellation!.reason);
+      // The last caller owns cancellation and waits for storage/queue cleanup.
+      void current.work.then(() => reject(cancellation!.reason), () => reject(cancellation!.reason));
+    };
+    cancellation?.addEventListener("abort", aborted, { once: true });
+    current.work.then(value => { if (release()) resolve(value as T); }, error => { if (release()) reject(error); });
+  });
+}
 type PackageIoQueue = { active: number; waiting: Array<() => void> };
 const storageIo: PackageIoQueue = { active: 0, waiting: [] };
 // Bound local response buffers independently so stalled storage does not prevent
@@ -113,8 +152,10 @@ async function readBoundedObject(storage: StorageService, companyId: string, key
 }
 
 /** Owns immutable package storage. Authorization belongs to the referencing
- * skill/source, and must be checked even when package bytes are already cached. */
+ * skill/source, and must be checked even when package bytes are already cached.
-export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
+ * Transaction readers supply their originating DB as cacheOwner for dedicated
+ * cache locks; metadata queries remain on the caller's transaction. */
+export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cacheOwner: Db = db) {
   const storage = () => suppliedStorage ?? getStorageService();
   const scope = (companyId: string, id: string) => and(eq(snapshots.companyId, companyId), eq(snapshots.id, id));
   async function descriptor(companyId: string, id: string): Promise<Descriptor> {
@@ -191,8 +232,8 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     const key = `${paths.manifest}:${row.storage?.manifestSha256 ?? row.contentHash}`;
     const cached = manifests.get(key);
     if (cached) return cached.manifest;
-    const load = async () => {
+    const load = async (sharedSignal: AbortSignal) => {
-      const signal = cancellation ? AbortSignal.any([cancellation, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000);
+      const signal = AbortSignal.any([sharedSignal, 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.
@@ -227,7 +268,7 @@ 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, cancellation);
+        if (row.storage) return manifestFor(row, signal);
         bytes = skillPackageManifestBytes(skillPackageManifest(files));
       }
       signal.throwIfAborted();
@@ -237,14 +278,12 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
         || manifest.totalBytes !== row.storage.totalBytes)) throw new Error("Skill package receipt does not match its manifest");
       await ensureSkillCacheDirectory(paths.parent, true);
       const temp = `${paths.manifest}.${randomUUID()}`;
-      try { await fs.writeFile(temp, bytes, { flag: "wx", mode: 0o600 }); await fs.rename(temp, paths.manifest); }
+      try { await fs.writeFile(temp, bytes, { flag: "wx", mode: 0o600, signal }); signal.throwIfAborted(); await fs.rename(temp, paths.manifest); }
       finally { await fs.rm(temp, { force: true }); }
       rememberManifest(key, manifest, bytes.length);
       return manifest;
     };
-    // Optional work must neither inherit an uncancellable interactive read nor
+    return sharedFlight(db, `manifest:${key}`, cancellation, load);
-    // 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) {
@@ -288,21 +327,53 @@ 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, cancellation?: AbortSignal) {
+    consume: (root: string, manifest: SkillPackageManifest) => Promise<T>, allowRecovery = true, cancellation?: AbortSignal, cacheRead = false) {
     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, signal);
-    return withPackageIo(async () => {
+    let published = false;
+    const result = await withPackageIo(async () => {
       const parent = skillPackagePaths(companyId, id).parent;
       await ensureSkillCacheDirectory(parent, true);
       const staging = await fs.mkdtemp(path.join(parent, ".verify-"));
       const archive = `${staging}.tar.gz`;
       try {
         await restoreInto(row, manifest, staging, archive, signal, allowRecovery);
-        return await consume(staging, manifest);
+        const result = await consume(staging, manifest);
+        if (cacheRead) {
+          // Network, extraction, hashing and response reads all happen before
+          // this short publication lock. Busy preparation keeps its priority.
+          const paths = skillPackagePaths(companyId, id);
+          const previous = await fs.mkdtemp(path.join(parent, ".verify-"));
+          try {
+            await withSkillVersionCacheLock(cacheOwner, companyId, async () => {
+              let moved = false;
+              try { await fs.rename(paths.directory, previous); moved = true; }
+              catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }
+              try { await fs.rename(staging, paths.directory); }
+              catch (error) { if (moved) await fs.rename(previous, paths.directory); throw error; }
+              relocateSkillPackageVerification(staging, paths.directory, manifest.files);
+              touchSkillPackage(paths.directory);
+              published = true;
+            }, true);
+          } catch (error) { if (!(error instanceof SkillCacheBusy)) throw error; }
+          finally { await fs.rm(previous, { recursive: true, force: true }); }
+        }
+        return result;
       } finally { await fs.rm(staging, { recursive: true, force: true }); await fs.rm(archive, { force: true }); }
     }, signal);
+    if (published) {
+      // Verify possible eviction candidates outside both the preparation lock
+      // and the storage slot used for this fill; take the lock only to remove.
+      await trimSkillPackageCache(db, companyId, skillPackagePaths(companyId, id).directory,
+        (id, signal) => verifyDurable(companyId, id, signal).catch(() => false),
+        async remove => {
+          try { await withSkillVersionCacheLock(cacheOwner, companyId, remove, true); }
+          catch (error) { if (!(error instanceof SkillCacheBusy)) throw error; }
+        });
+    }
+    return result;
   }
   async function materialize(row: Descriptor, manifest: SkillPackageManifest) {
     const paths = skillPackagePaths(row.companyId, row.id);
@@ -534,7 +605,11 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
       return result;
     }
     signal.throwIfAborted();
-    return withIsolatedPackage(version.companyId, id, read, allowRecovery, signal);
+    const acquire = async () => {
+      const cached = await withPackageIo(() => read(root).catch(() => null), signal, inventoryIo);
+      return cached ?? withIsolatedPackage(version.companyId, id, read, allowRecovery, signal, true);
+    };
+    return cancellation ? acquire() : flight(db, `version-read:${version.companyId}:${version.id}:${allowRecovery}`, acquire);
   }
   async function readVersionFile(version: StoredSkillVersion, relativePath: string, allowRecovery = true) {
     assertSkillSnapshotPath(relativePath);
@@ -557,7 +632,11 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     const cached = await withPackageIo(() => read(skillPackagePaths(version.companyId, id).directory).catch(() => null), signal, inventoryIo);
     if (cached) return cached;
     signal.throwIfAborted();
-    return withIsolatedPackage(version.companyId, id, read, allowRecovery, signal);
+    return flight(db, `file-read:${version.companyId}:${version.id}:${relativePath}:${allowRecovery}`,
+      async () => {
+        const cached = await withPackageIo(() => read(skillPackagePaths(version.companyId, id).directory).catch(() => null), signal, inventoryIo);
+        return cached ?? withIsolatedPackage(version.companyId, id, read, allowRecovery, signal, true);
+      });
   }
   async function versionInventory(version: StoredSkillVersion): Promise<Array<{ path: string; kind: CompanySkillVersionFileInventoryEntry["kind"] }>> {
     if (version.packageSnapshotId) {