fix(skills): bound cold response lifetime through cleanup

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

doc/skill-package-storage.md

@@ -112,6 +112,8 @@ pending delete/existence requests; incomplete cleanup retains the receipt for re
   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.
+  Cold administrative responses also have a two-response, 32-waiter bound held
+  through post-fill cleanup, so decoded inventories cannot accumulate behind storage work.
   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

@@ -1,4 +1,5 @@
 import { skillPackageWork } from "../services/skill-package-work.js";
+import * as packageCache from "../services/skill-package-cache.js";
 import * as recovery from "../services/skill-package-recovery.js";
 import type { GitHubRead } from "../services/github-skill-source.js";
 import { randomUUID } from "node:crypto";
@@ -546,6 +547,35 @@ describe.skipIf(!support.supported)("external skill packages", () => {
     } finally { clearTimeout(timeout); release(); await Promise.allSettled([reading, warming]); }
   });
 
+  it("bounds cold response lifetimes while cleanup is waiting without blocking warm preparation", async () => {
+    const fixtures = [await fixture(), await fixture(), await fixture()];
+    let bothEntered!: () => void;
+    const started = new Promise<void>(resolve => { bothEntered = resolve; });
+    let release!: () => void;
+    const blocked = new Promise<void>(resolve => { release = resolve; });
+    let decoded = 0;
+    const trim = vi.spyOn(packageCache, "trimSkillPackageCache").mockImplementation(async () => {
+      if (++decoded === 2) bothEntered();
+      await blocked;
+    });
+    const reads = fixtures.slice(0, 2).map(f => f.store.versionFiles(f.skills[0]!.version));
+    const storageRead = vi.spyOn(storage, "getObject");
+    try {
+      await started;
+      reads.push(fixtures[2]!.store.versionFiles(fixtures[2]!.skills[0]!.version));
+      expect((await fixtures[0]!.resolve()).every(entry => entry.sourceStatus === "available")).toBe(true);
+      // Let the third metadata/manifest read settle. Its archive and response
+      // must remain queued while two decoded responses are owned by cleanup.
+      await new Promise(resolve => setTimeout(resolve, 50));
+      expect(decoded).toBe(2);
+      const third = (await fixtures[2]!.store.descriptor(fixtures[2]!.companyId, fixtures[2]!.id)).storage!;
+      expect(storageRead.mock.calls.filter(call => call[1] === third.archiveKey)).toHaveLength(0);
+      release();
+      expect((await Promise.all(reads)).every(files => files[0]!.path === "SKILL.md")).toBe(true);
+      expect(trim).toHaveBeenCalledTimes(3);
+    } finally { release(); await Promise.allSettled(reads); }
+  });
+
   it("bounds concurrent cached inventories while leaving warm turn preparation responsive", async () => {
     const f = await fixture();
     await f.resolve();

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

@@ -16,7 +16,8 @@ export function touchSkillPackage(directory: string) {
  * 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>,
-  removeVerified: (action: () => Promise<void>) => Promise<void> = action => action()) {
+  removeVerified: (action: () => Promise<void>) => Promise<void> = action => action(), cancellation?: AbortSignal) {
+  cancellation?.throwIfAborted();
   const parent = path.dirname(directory);
   await ensureSkillCacheDirectory(parent);
   const contents = await fs.readdir(parent, { withFileTypes: true });
@@ -51,7 +52,8 @@ export async function trimSkillPackageCache(db: Db, companyId: string, directory
   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);
+  const timeout = AbortSignal.timeout(10_000);
+  const signal = cancellation ? AbortSignal.any([cancellation, timeout]) : timeout;
   let attempted = 0;
   for (const entry of entries.sort((a, b) => a.lastUsed - b.lastUsed)) {
     if (bytes <= MAX_BYTES && count <= MAX_PACKAGES) break;

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

@@ -91,6 +91,8 @@ function sharedFlight<T>(db: Db, key: string, cancellation: AbortSignal | undefi
 }
 type PackageIoQueue = { active: number; waiting: Array<() => void> };
 const storageIo: PackageIoQueue = { active: 0, waiting: [] };
+// A decoded cold response remains owned until its post-fill cleanup finishes.
+const coldResponses: PackageIoQueue = { active: 0, waiting: [] };
 // Bound local response buffers independently so stalled storage does not prevent
 // verified cached reads, including reads inside an existing DB transaction.
 const inventoryIo: PackageIoQueue = { active: 0, waiting: [] };
@@ -324,12 +326,17 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach
     }
     if (!await verifySkillPackageDirectory(staging, manifest.files)) throw new Error("Restored skill package does not match its manifest");
   }
-  // Admin reads and retention verification own a scratch directory. They never
+  // Archive reads and verification own private scratch space outside the warm
-  // wait for or hold the lock used to prepare and pin an agent's warm files.
+  // preparation lock. Administrative reads briefly try that lock to publish.
   async function withIsolatedPackage<T>(companyId: string, id: string,
     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 read = () => readIsolatedPackage(companyId, id, consume, allowRecovery, signal, cacheRead);
+    return cacheRead ? withPackageIo(read, signal, coldResponses) : read();
+  }
+  async function readIsolatedPackage<T>(companyId: string, id: string,
+    consume: (root: string, manifest: SkillPackageManifest) => Promise<T>, allowRecovery: boolean, signal: AbortSignal, cacheRead: boolean) {
     const row = await descriptor(companyId, id);
     const manifest = await manifestFor(row, signal);
     let published = false;
@@ -371,8 +378,9 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService, cach
         async remove => {
           try { await withSkillVersionCacheLock(cacheOwner, companyId, remove, true); }
           catch (error) { if (!(error instanceof SkillCacheBusy)) throw error; }
-        });
+        }, signal);
     }
+    signal.throwIfAborted();
     return result;
   }
   async function materialize(row: Descriptor, manifest: SkillPackageManifest) {