From 06689c1b3410000bcd65ff40cb469e600209c4db Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:39:02 -0500 Subject: [PATCH] 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 --- doc/DEVELOPING.md | 6 ++ maps/agent-access-map/skills.md | 7 ++ maps/feature-map/skills.md | 6 ++ ...company-skills-runtime-concurrency.test.ts | 80 +++++++++++++++++++ .../heartbeat-execution-configuration.test.ts | 8 +- .../src/__tests__/skill-version-cache.test.ts | 21 ++++- server/src/services/company-skills.ts | 21 ++++- .../heartbeat/execution-configuration.ts | 69 ++++++++-------- 8 files changed, 173 insertions(+), 45 deletions(-) create mode 100644 server/src/__tests__/company-skills-runtime-concurrency.test.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 7edd08fde99..bd7d1acdfb5 100644 --- a/doc/DEVELOPING.md +++ b/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. diff --git a/maps/agent-access-map/skills.md b/maps/agent-access-map/skills.md index e0b19430907..6106b0eb541 100644 --- a/maps/agent-access-map/skills.md +++ b/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`. diff --git a/maps/feature-map/skills.md b/maps/feature-map/skills.md index 7c7e19a5640..1b5a206d2bc 100644 --- a/maps/feature-map/skills.md +++ b/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 diff --git a/server/src/__tests__/company-skills-runtime-concurrency.test.ts b/server/src/__tests__/company-skills-runtime-concurrency.test.ts new file mode 100644 index 00000000000..db6b464e4f8 --- /dev/null +++ b/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>; + let db: ReturnType; + 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([]); + }); +}); diff --git a/server/src/__tests__/heartbeat-execution-configuration.test.ts b/server/src/__tests__/heartbeat-execution-configuration.test.ts index 644532e8307..0a040d07870 100644 --- a/server/src/__tests__/heartbeat-execution-configuration.test.ts +++ b/server/src/__tests__/heartbeat-execution-configuration.test.ts @@ -49,10 +49,12 @@ vi.mock("../services/company-skills.js", async importOriginal => { const actual = await importOriginal(); return { ...actual, companySkillService: (...args: Parameters) => { const service = actual.companySkillService(...args); - return { ...service, listRuntimeSkillEntries: async (...input: Parameters) => { - const failure = skillFailures.get(input[0]); + return { ...service, withRuntimeSkillEntries: async (companyId: string, + options: Parameters[1], + consume: (entries: Awaited>) => T | Promise) => { + const failure = skillFailures.get(companyId); if (failure) throw new Error(failure); - return service.listRuntimeSkillEntries(...input); + return service.withRuntimeSkillEntries(companyId, options, consume); } }; } }; }); diff --git a/server/src/__tests__/skill-version-cache.test.ts b/server/src/__tests__/skill-version-cache.test.ts index bd6b1f6cd49..67e19206e32 100644 --- a/server/src/__tests__/skill-version-cache.test.ts +++ b/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(); const release = Promise.withResolvers(); - 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 () => { diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index 2637721cf6e..37d3b14d785 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -3383,6 +3383,11 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo async function listFull(companyId: string): Promise { if (!visibility().canReadCompany(companyId)) return []; await ensureSkillInventoryCurrent(companyId); + return readFullSkillRows(companyId); + } + + async function readFullSkillRows(companyId: string): Promise { + 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 { + return withRuntimeSkillEntries(companyId, options, entries => entries); + } + + async function withRuntimeSkillEntries( + companyId: string, + options: RuntimeSkillEntryOptions, + consume: (entries: PaperclipSkillEntry[]) => T | Promise, + ): Promise { 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 { - 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, }; } diff --git a/server/src/services/heartbeat/execution-configuration.ts b/server/src/services/heartbeat/execution-configuration.ts index 1f92076b29e..f10acfda751 100644 --- a/server/src/services/heartbeat/execution-configuration.ts +++ b/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[0]["secretsSvc"]; - companySkills: Pick, "listRuntimeSkillEntries">; + companySkills: Pick, "withRuntimeSkillEntries">; appendRunEvent: ReturnType["appendRunEvent"]; finalizeAiConnectionBusyDeferral: ReturnType["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",