perf(skills): coalesce inventory before run cache locking (#16576)

Share each active company skill-inventory refresh before taking the version-cache lock. Read actor visibility under the lock and keep snapshot publication protected through durable run pinning. Preserve empty-selection behavior and the list API.

Validation: 176 affected cases, full repository typecheck/build, fresh full CI/E2E, security and exact-head Apex 5/5 pass. The local six-reader latency experiment improves from 50.4s to 18.3s; live response latency is a separate acceptance check.

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

doc/DEVELOPING.md

@@ -1506,6 +1506,12 @@ as they are available, so later setup failures still clean up acquired resources
 Busy-account deferral returns explicitly before session/workspace preparation.
 Public-path tests are in `server/src/__tests__/heartbeat-execution-configuration.test.ts`.
 
+Concurrent skill preparations join the company inventory refresh before taking
+the version-cache lock. Each preparation then reads its actor's current skill
+visibility under that lock. Snapshot materialization and the durable run version
+pins stay in the same locked callback. Only the in-flight inventory work is
+shared; completed refreshes and permission results are not cached.
+
 Browser Use skill assignment first checks for an enabled, active Browser Use
 connection in the run's company. With none, it skips full tool discovery.
 With one, the existing task, tool-policy and credential-grant checks still apply.

maps/agent-access-map/skills.md

@@ -102,6 +102,13 @@ exclusion and immutable version pinning. This changes no agent endpoint or grant
 the [configuration tests](../../server/src/__tests__/heartbeat-execution-configuration.test.ts)
 cover the held-lock boundary with real PostgreSQL.
 
+Concurrent preparations share only in-flight inventory reconciliation. After
+that work, each actor's visible skill rows are read again under the company cache
+lock. Snapshot delivery and durable run pins complete before the lock is released.
+No permission result is shared or kept between requests. The
+[concurrency tests](../../server/src/__tests__/company-skills-runtime-concurrency.test.ts)
+cover separate actors and project-access revocation; this adds no agent API.
+
 ## `policy`
 
 Review: `baseline`.

maps/feature-map/skills.md

@@ -63,6 +63,12 @@ later assignments. Selected skills and the legacy operational skill retain cache
 exclusion through version pinning. The [configuration regression](../../server/src/__tests__/heartbeat-execution-configuration.test.ts)
 uses a real PostgreSQL lock; live provider latency remains a separate check.
 
+Concurrent selected-skill preparations share one in-flight company inventory
+refresh before cache locking. Each run retains its own selection and durable
+version pins. The [concurrency regression](../../server/src/__tests__/company-skills-runtime-concurrency.test.ts)
+checks shared work and separate actor access; the cache tests check that cleanup
+cannot remove a snapshot before the run stores its pins.
+
 ## Gotchas
 
 - The release picker requires `enableBetaSkills`. Do not invent an Instructions

server/src/__tests__/company-skills-runtime-concurrency.test.ts

@@ -0,0 +1,80 @@
+import { randomUUID } from "node:crypto";
+import { promises as fs } from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { agents, companies, companyMemberships, companySkills, companySkillSources, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS,
+  projectAccessMembers, projects, projectSkillSources } from "@paperclipai/db";
+import { and, eq } from "drizzle-orm";
+import { PAPERCLIP_FEEDBACK_SKILL_KEYS, PAPERCLIP_OPERATIONAL_SKILL_KEY } from "@paperclipai/adapter-utils/server-utils";
+import { companySkillService } from "../services/company-skills.js";
+import { withSkillActor } from "../services/skill-access.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+const describeDatabase = support.supported ? describe : describe.skip;
+
+describeDatabase("concurrent runtime skill preparation", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  let home: string;
+  const previousHome = process.env.PAPERCLIP_HOME;
+
+  beforeAll(async () => {
+    database = await startEmbeddedPostgresTestDatabase("runtime-skill-concurrency-");
+    db = createDb(database.connectionString);
+    home = await fs.mkdtemp(path.join(os.tmpdir(), "runtime-skill-concurrency-"));
+    process.env.PAPERCLIP_HOME = home;
+  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
+
+  afterAll(async () => {
+    if (previousHome === undefined) delete process.env.PAPERCLIP_HOME;
+    else process.env.PAPERCLIP_HOME = previousHome;
+    await db?.$client.end({ timeout: 0 });
+    await database?.cleanup();
+    if (home) await fs.rm(home, { recursive: true, force: true });
+  });
+
+  it("shares an in-flight inventory refresh across service instances before snapshot locking", async () => {
+    const companyId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Concurrent skill readers", issuePrefix: `SC${companyId.slice(0, 6)}` });
+    const select = vi.spyOn(db, "select");
+    try {
+      const selections = await Promise.all(Array.from({ length: 6 }, () => companySkillService(db).listRuntimeSkillEntries(companyId, {
+        selectionConfig: { paperclipSkillSync: { desiredSkills: [] } }, includeLegacy: true, snapshotLatest: true,
+      })));
+      for (const entries of selections) {
+        expect(entries.map(entry => entry.key).sort()).toEqual([PAPERCLIP_OPERATIONAL_SKILL_KEY, ...PAPERCLIP_FEEDBACK_SKILL_KEYS].sort());
+        expect(entries.every(entry => entry.sourceStatus === "available" && entry.versionId)).toBe(true);
+      }
+      expect(select.mock.calls.filter(([fields]) => fields && Object.keys(fields).length === 1 && fields.id === companies.id)).toHaveLength(1);
+      await companySkillService(db).listRuntimeSkillEntries(companyId, {
+        selectionConfig: { paperclipSkillSync: { desiredSkills: [] } }, includeLegacy: true, snapshotLatest: true,
+      });
+      expect(select.mock.calls.filter(([fields]) => fields && Object.keys(fields).length === 1 && fields.id === companies.id)).toHaveLength(2);
+    } finally { select.mockRestore(); }
+  }, 30_000);
+
+  it("shares inventory work without sharing actor visibility or retaining revoked project access", async () => {
+    const companyId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Separate skill readers", issuePrefix: `SV${companyId.slice(0, 6)}` });
+    const readers = await db.insert(agents).values(["Allowed", "Denied"].map(name => ({ companyId, name, adapterType: "codex_local" }))).returning();
+    await db.insert(companyMemberships).values(readers.map(reader => ({ companyId, principalType: "agent", principalId: reader.id, membershipRole: "member" })));
+    const [project] = await db.insert(projects).values({ companyId, name: "Private skill source", visibility: "private" }).returning();
+    const [source] = await db.insert(companySkillSources).values({ companyId, repositoryUrl: "https://github.com/fixture/private-skills",
+      fullName: "fixture/private-skills", trackingRef: "main", accessScope: "projects" }).returning();
+    await db.insert(projectSkillSources).values({ companyId, projectId: project!.id, sourceId: source!.id,
+      repositoryUrl: source!.repositoryUrl, trackingRef: "main", status: "ready" });
+    await db.insert(projectAccessMembers).values({ companyId, projectId: project!.id, subjectType: "agent", subjectId: readers[0]!.id });
+    const skill = await companySkillService(db).createLocalSkill(companyId, { name: "Private local skill" });
+    await db.update(companySkills).set({ accessSourceId: source!.id }).where(eq(companySkills.id, skill.id));
+    const read = (agentId: string) => withSkillActor({ type: "agent", companyId, agentId, source: "agent_jwt" },
+      () => companySkillService(db).listRuntimeSkillEntries(companyId, { onlyKeys: new Set([skill.key]), snapshotLatest: true }));
+    const [allowed, denied] = await Promise.all(readers.map(reader => read(reader.id)));
+    expect(allowed!.map(entry => entry.key)).toEqual([skill.key]);
+    expect(allowed![0]!.sourceStatus).toBe("available");
+    expect(denied).toEqual([]);
+    await db.delete(projectAccessMembers).where(and(eq(projectAccessMembers.projectId, project!.id), eq(projectAccessMembers.subjectId, readers[0]!.id)));
+    expect(await read(readers[0]!.id)).toEqual([]);
+  });
+});

server/src/__tests__/heartbeat-execution-configuration.test.ts

@@ -49,10 +49,12 @@ vi.mock("../services/company-skills.js", async importOriginal => {
   const actual = await importOriginal<typeof import("../services/company-skills.js")>();
   return { ...actual, companySkillService: (...args: Parameters<typeof actual.companySkillService>) => {
     const service = actual.companySkillService(...args);
-    return { ...service, listRuntimeSkillEntries: async (...input: Parameters<typeof service.listRuntimeSkillEntries>) => {
-      const failure = skillFailures.get(input[0]);
+    return { ...service, withRuntimeSkillEntries: async <T>(companyId: string,
+      options: Parameters<typeof service.withRuntimeSkillEntries>[1],
+      consume: (entries: Awaited<ReturnType<typeof service.listRuntimeSkillEntries>>) => T | Promise<T>) => {
+      const failure = skillFailures.get(companyId);
       if (failure) throw new Error(failure);
-      return service.listRuntimeSkillEntries(...input);
+      return service.withRuntimeSkillEntries(companyId, options, consume);
     } };
   } };
 });

server/src/__tests__/skill-version-cache.test.ts

@@ -142,8 +142,10 @@ describePostgres("immutable skill version cache", () => {
     const run = await activeRun(f.companyId, null);
     const ready = Promise.withResolvers<void>();
     const release = Promise.withResolvers<void>();
-    const preparing = withSkillVersionCacheLock(db, f.companyId, async () => {
-      await f.resolve(f.versions[0]!.id);
+    const preparing = companySkillService(db).withRuntimeSkillEntries(f.companyId, {
+      onlyKeys: new Set([f.key]), versionSelections: new Map([[f.key, f.versions[0]!.id]]),
+    }, async entries => {
+      expect(entries).toMatchObject([{ key: f.key, versionId: f.versions[0]!.id, sourceStatus: "available" }]);
       await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
       ready.resolve();
       await release.promise;
@@ -159,6 +161,21 @@ describePostgres("immutable skill version cache", () => {
     } finally { release.resolve(); }
     await preparing;
     expect(await sweepSkillVersionCache(other, { companyId: f.companyId })).toMatchObject({ evicted: 0, skipped: { protected_version: 1 } });
+    await other.$client.end({ timeout: 0 });
+  });
+
+  it("releases the cache lock after a runtime selection consumer fails", async () => {
+    const f = await fixture();
+    await expect(companySkillService(db).withRuntimeSkillEntries(f.companyId, {
+      onlyKeys: new Set([f.key]), versionSelections: new Map([[f.key, f.versions[0]!.id]]),
+    }, async entries => {
+      expect(entries[0]!.sourceStatus).toBe("available");
+      throw new Error("durable pin write failed");
+    })).rejects.toThrow("durable pin write failed");
+    const other = createDb(database.connectionString);
+    try {
+      expect(await withSkillVersionCacheLock(other, f.companyId, async () => "released", true)).toBe("released");
+    } finally { await other.$client.end({ timeout: 0 }); }
   });
 
   it("reads company-wide protection twice per batch regardless of version count", async () => {

server/src/services/company-skills.ts

@@ -3383,6 +3383,11 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
   async function listFull(companyId: string): Promise<CompanySkill[]> {
     if (!visibility().canReadCompany(companyId)) return [];
     await ensureSkillInventoryCurrent(companyId);
+    return readFullSkillRows(companyId);
+  }
+
+  async function readFullSkillRows(companyId: string): Promise<CompanySkill[]> {
+    if (!visibility().canReadCompany(companyId)) return [];
     const rows = await db
       .select(selectCompanySkillColumns())
       .from(companySkills)
@@ -6152,16 +6157,25 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
     companyId: string,
     options: RuntimeSkillEntryOptions = {},
   ): Promise<PaperclipSkillEntry[]> {
+    return withRuntimeSkillEntries(companyId, options, entries => entries);
+  }
+
+  async function withRuntimeSkillEntries<T>(
+    companyId: string,
+    options: RuntimeSkillEntryOptions,
+    consume: (entries: PaperclipSkillEntry[]) => T | Promise<T>,
+  ): Promise<T> {
     if (options.onlyKeys?.size === 0 || (!options.onlyKeys && options.selectionConfig && !options.includeLegacy
-      && readPaperclipSkillSyncPreference(options.selectionConfig).desiredSkills.length === 0)) return [];
-    return withSkillVersionCacheLock(db, companyId, () => listRuntimeSkillEntriesLocked(companyId, options));
+      && readPaperclipSkillSyncPreference(options.selectionConfig).desiredSkills.length === 0)) return consume([]);
+    await ensureSkillInventoryCurrent(companyId);
+    return withSkillVersionCacheLock(db, companyId, async () => consume(await listRuntimeSkillEntriesLocked(companyId, options)));
   }
 
   async function listRuntimeSkillEntriesLocked(
     companyId: string,
     options: RuntimeSkillEntryOptions,
   ): Promise<PaperclipSkillEntry[]> {
-    const skills = await listFull(companyId);
+    const skills = await readFullSkillRows(companyId);
     const selectedKeys = options.onlyKeys ?? (options.selectionConfig ? new Set(
       (options.includeLegacy ? resolveLegacyPaperclipDesiredSkillNames : resolvePaperclipDesiredSkillNames)(
         options.selectionConfig, skills.map(skill => ({ key: skill.key, runtimeName: buildSkillRuntimeName(skill.key, skill.slug) }))),
@@ -7442,5 +7456,6 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
     installUpdate,
     resetSkill,
     listRuntimeSkillEntries,
+    withRuntimeSkillEntries,
   };
 }

server/src/services/heartbeat/execution-configuration.ts

@@ -1,5 +1,4 @@
 import { withSkillActor } from "../skill-access.js";
-import { withSkillVersionCacheLock } from "../skill-version-cache.js";
 import type { AuthorizationActor } from "../authorization.js";
 import { eq, sql } from "drizzle-orm";
 import { heartbeatRuns, type Db, type agents, type projects } from "@paperclipai/db";
@@ -70,7 +69,7 @@ export interface HeartbeatExecutionConfigurationInput {
     "selectedEnvironmentId" | "selectedEnvironmentForConfig" | "resolvedInstanceSettings" | "persistedNativeExecutionInput">;
   services: {
     secretsSvc: Parameters<typeof resolveExecutionRunAdapterConfig>[0]["secretsSvc"];
-    companySkills: Pick<ReturnType<typeof companySkillService>, "listRuntimeSkillEntries">;
+    companySkills: Pick<ReturnType<typeof companySkillService>, "withRuntimeSkillEntries">;
     appendRunEvent: ReturnType<typeof createHeartbeatLifecycle>["appendRunEvent"];
     finalizeAiConnectionBusyDeferral: ReturnType<typeof createHeartbeatRetries>["finalizeAiConnectionBusyDeferral"];
   };
@@ -224,9 +223,6 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
   let skillDeliveryConfig = effectiveResolvedConfig;
   const storedVersions = parseObject(run.contextSnapshot?.skillVersions);
   const hasStoredVersions = Object.prototype.hasOwnProperty.call(run.contextSnapshot ?? {}, "skillVersions");
-  const emptySkillSelection = hasStoredVersions
-    ? run.contextSnapshot?.skillSelectionSnapshot === true && Object.keys(storedVersions).length === 0
-    : agent.adapterType === "paperclip_runner" && runtimeSkillPreference.desiredSkills.length === 0;
   const prepareSkills = async () => {
 
     try {
@@ -235,40 +231,41 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
         : skillVersionSelectionMap(runtimeSkillPreference.desiredSkillEntries, {
           versionPinsEnabled: resolvedInstanceSettings.experimental.enableBetaSkills === true,
         });
-      const entries = await withSkillActor(skillActor, () => companySkills.listRuntimeSkillEntries(agent.companyId, {
+      return await withSkillActor(skillActor, () => companySkills.withRuntimeSkillEntries(agent.companyId, {
         versionSelections, snapshotLatest: true,
         ...(hasStoredVersions && run.contextSnapshot?.skillSelectionSnapshot === true
           ? { onlyKeys: new Set(Object.keys(storedVersions)) }
           : { selectionConfig: effectiveResolvedConfig, includeLegacy: agent.adapterType !== "paperclip_runner" }),
-      }));
-      const available = hasStoredVersions ? entries.filter(entry => Object.prototype.hasOwnProperty.call(storedVersions, entry.key)) : entries;
-      const availableKeys = new Set(available.map(entry => entry.key));
-      const desiredKeys = hasStoredVersions && run.contextSnapshot?.skillSelectionSnapshot === true
-        ? Object.keys(storedVersions)
-        : agent.adapterType === "paperclip_runner"
-          ? resolvePaperclipDesiredSkillNames(effectiveResolvedConfig, available)
-          : resolveLegacyPaperclipDesiredSkillNames(effectiveResolvedConfig, available);
-      const selectedKeys = new Set(desiredKeys.filter(key => availableKeys.has(key)));
-      const resolved = available.filter(entry => selectedKeys.has(entry.key));
-      // Optional assignments can outlive project membership. Drop inaccessible
-      // assignments from delivery; explicit task mentions remain errors below.
-      skillDeliveryConfig = writePaperclipSkillSyncPreference(effectiveResolvedConfig,
-        resolved.map(entry => ({ key: entry.key, versionId: entry.versionId ?? null })));
-      for (const key of runScopedMentionedSkillKeys) {
-        const entry = resolved.find(candidate => candidate.key === key);
-        if (!entry || entry.sourceStatus === "missing") {
-          throw new ConfigurationIncompleteFailure("A referenced skill is unavailable. Restore its installed version or update the task's skill reference.", {
-            configurationIncomplete: { reason: "skill_unavailable", skillKey: key, actionUrl: "/skills" },
-          });
+      }, async entries => {
+        const available = hasStoredVersions ? entries.filter(entry => Object.prototype.hasOwnProperty.call(storedVersions, entry.key)) : entries;
+        const availableKeys = new Set(available.map(entry => entry.key));
+        const desiredKeys = hasStoredVersions && run.contextSnapshot?.skillSelectionSnapshot === true
+          ? Object.keys(storedVersions)
+          : agent.adapterType === "paperclip_runner"
+            ? resolvePaperclipDesiredSkillNames(effectiveResolvedConfig, available)
+            : resolveLegacyPaperclipDesiredSkillNames(effectiveResolvedConfig, available);
+        const selectedKeys = new Set(desiredKeys.filter(key => availableKeys.has(key)));
+        const resolved = available.filter(entry => selectedKeys.has(entry.key));
+        // Optional assignments can outlive project membership. Drop inaccessible
+        // assignments from delivery; explicit task mentions remain errors below.
+        skillDeliveryConfig = writePaperclipSkillSyncPreference(effectiveResolvedConfig,
+          resolved.map(entry => ({ key: entry.key, versionId: entry.versionId ?? null })));
+        for (const key of runScopedMentionedSkillKeys) {
+          const entry = resolved.find(candidate => candidate.key === key);
+          if (!entry || entry.sourceStatus === "missing") {
+            throw new ConfigurationIncompleteFailure("A referenced skill is unavailable. Restore its installed version or update the task's skill reference.", {
+              configurationIncomplete: { reason: "skill_unavailable", skillKey: key, actionUrl: "/skills" },
+            });
+          }
         }
-      }
-      if (!hasStoredVersions) {
-        const skillVersions = Object.fromEntries(resolved.map(entry => [entry.key, entry.versionId ?? entry.currentVersionId ?? null]));
-        context.skillVersions = skillVersions;
-        context.skillSelectionSnapshot = true;
-        await db.update(heartbeatRuns).set({ contextSnapshot: sql`coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) || ${JSON.stringify({ skillVersions, skillSelectionSnapshot: true })}::jsonb` }).where(eq(heartbeatRuns.id, run.id));
-      }
-      return resolved;
+        if (!hasStoredVersions) {
+          const skillVersions = Object.fromEntries(resolved.map(entry => [entry.key, entry.versionId ?? entry.currentVersionId ?? null]));
+          context.skillVersions = skillVersions;
+          context.skillSelectionSnapshot = true;
+          await db.update(heartbeatRuns).set({ contextSnapshot: sql`coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) || ${JSON.stringify({ skillVersions, skillSelectionSnapshot: true })}::jsonb` }).where(eq(heartbeatRuns.id, run.id));
+        }
+        return resolved;
+      }));
     } catch (error) {
       if (agent.adapterType === "paperclip_runner") {
         await recordFailedSkillPreparation({
@@ -282,9 +279,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
       throw error;
     }
   };
-  const runtimeSkillEntries = preconfiguredHome ? [] : await time("configuration_skills", () => emptySkillSelection
-    ? prepareSkills()
-    : withSkillVersionCacheLock(db, agent.companyId, prepareSkills));
+  const runtimeSkillEntries = preconfiguredHome ? [] : await time("configuration_skills", prepareSkills);
   nativeRunnerPreparationSpans.push({
     name: "skills.prepare",
     parentName: "task.prepare",