diff --git a/doc/observability.md b/doc/observability.md index e730ad2120..3d45b420fd 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -148,6 +148,7 @@ task.run │ ├── environment.startup │ │ ├── environment.acquire │ │ └── environment.workspace.realize +│ ├── skills.prepare │ ├── heartbeat.prepare_before_environment │ ├── heartbeat.prepare_after_environment │ └── native.coordinator.claim @@ -766,3 +767,49 @@ ledger across routes: a per-route bound stops one busy route from starving another route's own budget, but the host enforces no smaller ceiling on the sum across every route. Keep every dimension low-cardinality and free of user content. + +### Shared skill preparation + +`skills.prepare` measures the shared inventory listing and runtime materialization +inside `task.prepare`. It is also contained in the broader +`heartbeat.prepare_before_environment` interval; do not add those two durations. +Preparation failures emit a failed span even when no native session starts. +It carries no skill contents, identifiers, locations, or credentials. It uses the +existing run performance events and operator-configured OpenTelemetry endpoint; +no first-party Telemetry event is added. + +Runtime preparation refreshes the company inventory once per listing. Local and +catalog directories remain direct sources, so edits are visible on the next +preparation. Explicit version selections still use their stored snapshots. + +Reconstructed skills use `__runtime_cache_v1__///files` +beneath company skill storage, with a sibling manifest of paths, sizes, and SHA-256 +content digests. Every warm hit validates the manifest and exact file contents; +it does not fetch upstream, rewrite files, or remove directories. The fingerprint +includes installed source identity, revision, file inventory, and stored Markdown, +and excludes display names, stars, and general update timestamps. Manifests stay +outside the directory delivered to agents. + +GitHub and skills.sh imports are cached only when pinned to a full commit SHA. +Remote freshness is explicit: update or reimport selects a new revision, including +supporting-file-only changes. A branch advancing upstream does not change an +installed revision. Legacy mutable refs retain uncached behavior until updated. +URL-only skills use stored Markdown. An unavailable new revision reports missing; +it never silently reuses an older revision. Stored `SKILL.md` remains a fallback, +but missing supporting files prevent publication of a reusable partial cache. + +Builds publish read-only files and directories from unique staging directories. +A skill-scoped lock serializes builds and cleanup across processes. Cold builders +recheck that the skill still exists under its original key before reading files +and before atomic publication. Existing valid +revisions stay readable during updates. Invalid entries are quarantined in the +same skill cache root for inspection; rename/removal cleans up that skill's cache. +Read-only listings validate caches without downloading or repairing them. A +publication lock left by an abruptly terminated process is reported for operator +cleanup; remove it only after confirming its recorded PID is no longer running. + +Run `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-skill-preparation.ts` for an isolated embedded +PostgreSQL benchmark with 114 mixed skills and at least 400 remote files. It +reports one cold sample and ten warm samples (one in a new process), refresh and +fetch counts, rebuilds, missing entries, and content checks. Upstream responses are +deterministic fixtures; use real deployed run spans for user-facing latency. diff --git a/scripts/benchmark-skill-preparation.ts b/scripts/benchmark-skill-preparation.ts new file mode 100644 index 0000000000..068c33ccfa --- /dev/null +++ b/scripts/benchmark-skill-preparation.ts @@ -0,0 +1,115 @@ +/** Isolated, repeatable preparation benchmark. Run with pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-skill-preparation.ts. */ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { companies, companySkills, createDb, startEmbeddedPostgresTestDatabase } from "../packages/db/src/index.js"; +import { removeRuntimeSkillCache } from "../server/src/services/runtime-skill-cache.js"; +import { companySkillService } from "../server/src/services/company-skills.js"; + +const child = process.argv.includes("--warm-child"); +const home = child ? process.env.PAPERCLIP_HOME! : await fs.mkdtemp(path.join(os.tmpdir(), "skill-preparation-benchmark-")); +process.env.PAPERCLIP_HOME = home; +process.env.PAPERCLIP_INSTANCE_ID = "default"; +const database = child ? null : await startEmbeddedPostgresTestDatabase("skill-preparation-benchmark-"); +const db = createDb(database?.connectionString ?? process.env.SKILL_BENCH_DATABASE_URL!); +const companyId = child ? process.env.SKILL_BENCH_COMPANY_ID! : randomUUID(); +const svc = companySkillService(db); +let upstreamFetches = 0; +let inventoryRefreshes = 0; +let rebuilds = 0; +const originalFetch = globalThis.fetch; +globalThis.fetch = async (input) => { upstreamFetches++; return new Response(`# Fixture\n${String(input)}\n`); }; +const originalSelect = db.select.bind(db); +db.select = ((fields?: Record) => { + if (fields && Object.keys(fields).length === 1 && fields.id === companies.id) inventoryRefreshes++; + return originalSelect(fields as never); +}) as typeof db.select; +const originalMkdtemp = fs.mkdtemp; +fs.mkdtemp = ((prefix: string, ...args: unknown[]) => { + if (String(prefix).includes("__runtime_cache_v1__") && String(prefix).includes(".staging-")) rebuilds++; + return Reflect.apply(originalMkdtemp, fs, [prefix, ...args]); +}) as typeof fs.mkdtemp; + +async function measure(label: string) { + upstreamFetches = inventoryRefreshes = rebuilds = 0; + const start = performance.now(); + const entries = await svc.listRuntimeSkillEntries(companyId); + const durationMs = performance.now() - start; + const missingEntries = entries.filter((entry) => entry.sourceStatus === "missing").length; + assert.equal(entries.length, 114); + assert.equal(missingEntries, 0); + assert.equal(inventoryRefreshes, 1); + if (label !== "cold") { assert.equal(upstreamFetches, 0); assert.equal(rebuilds, 0); } + // Check every synthetic supporting file against its installed revision, outside the timed section. + const installed = await db.select().from(companySkills); + for (const skill of installed.filter((skill) => skill.sourceType === "github" || skill.sourceType === "skills_sh")) { + const entry = entries.find((entry) => entry.key === skill.key)!; + for (const file of skill.fileInventory) { + const expected = `# Fixture\nhttps://raw.githubusercontent.com/fixture/skills/${skill.sourceRef}/${skill.slug}/${file.path}\n`; + assert.equal(await fs.readFile(path.join(entry.source, file.path), "utf8"), expected); + } + } + return { label, durationMs: Math.round(durationMs * 100) / 100, inventoryRefreshes, upstreamFetches, rebuilds, missingEntries, entries: entries.length }; +} + +try { + if (child) { + console.log(JSON.stringify(await measure("warm-new-process"))); + } else { + await db.insert(companies).values({ id: companyId, name: "Skill benchmark", issuePrefix: "BENCH" }); + const bundled = await svc.listFull(companyId); + const additional = 114 - bundled.length; + assert.ok(additional >= 6, "Bundled inventory leaves insufficient room for mixed benchmark fixtures"); + const remoteCount = 33; + const catalogCount = 21; + assert.ok(additional > remoteCount + catalogCount, "Bundled inventory leaves no room for local fixtures"); + const filesPerRemote = Math.max(9, Math.ceil(400 / remoteCount)); + for (let index = 0; index < additional; index++) { + const slug = `fixture-${index}`; + const sourceType = index < remoteCount ? (index < 28 ? "github" : "skills_sh") + : index === remoteCount ? "url" : index <= remoteCount + catalogCount ? "catalog" : "local_path"; + const localDir = path.join(home, "instances", "default", "skills", companyId, slug); + if (sourceType === "local_path" || sourceType === "catalog") { + await fs.mkdir(localDir, { recursive: true }); + await fs.writeFile(path.join(localDir, "SKILL.md"), `---\nname: ${slug}\ndescription: Benchmark\n---\n# Local\n`); + } + await db.insert(companySkills).values({ + id: randomUUID(), companyId, key: `company/${companyId}/${slug}`, slug, name: slug, + markdown: `# ${slug}`, sourceType, sourceLocator: sourceType === "local_path" || sourceType === "catalog" ? localDir : `https://example.com/${slug}`, + sourceRef: index < remoteCount ? "a".repeat(40) : null, + trustLevel: "markdown_only", compatibility: "compatible", + metadata: { owner: "fixture", repo: "skills", repoSkillDir: slug }, + fileInventory: [{ path: "SKILL.md", kind: "skill" }, ...Array.from({ length: index < remoteCount ? filesPerRemote - 1 : 0 }, (_, file) => ({ path: `references/${file}.md`, kind: "reference" as const }))], + }); + } + const samples = [await measure("cold")]; + for (let index = 0; index < 9; index++) samples.push(await measure(`warm-${index + 1}`)); + const childOutput = await new Promise((resolve, reject) => { + const proc = spawn(process.execPath, ["--import", fileURLToPath(new URL("../server/node_modules/tsx/dist/loader.mjs", import.meta.url)), fileURLToPath(import.meta.url), "--warm-child"], { + env: { ...process.env, SKILL_BENCH_DATABASE_URL: database!.connectionString, SKILL_BENCH_COMPANY_ID: companyId }, + stdio: ["ignore", "pipe", "inherit"], + }); + let output = ""; + proc.stdout.on("data", (chunk) => { output += chunk; }); + proc.on("error", reject); + proc.on("exit", (code) => code === 0 ? resolve(output) : reject(new Error(`Warm subprocess failed: ${code}`))); + }); + samples.push(JSON.parse(childOutput.trim().split("\n").at(-1)!)); + const warm = samples.slice(1).map((sample) => sample.durationMs).sort((a, b) => a - b); + console.log(JSON.stringify({ inventory: 114, materializedRemoteFiles: remoteCount * filesPerRemote, + warmMedianMs: (warm[4] + warm[5]) / 2, warmMaxMs: warm.at(-1), samples }, null, 2)); + } +} finally { + globalThis.fetch = originalFetch; + fs.mkdtemp = originalMkdtemp; + if (!child) for (const skill of await db.select().from(companySkills)) { + await removeRuntimeSkillCache(path.join(home, "instances", "default", "skills", companyId), skill.id); + } + await db.$client.end(); + await database?.cleanup(); + if (!child) await fs.rm(home, { recursive: true, force: true }); +} diff --git a/server/src/__tests__/company-skills-service.test.ts b/server/src/__tests__/company-skills-service.test.ts index a62b6cd607..7532ca13e5 100644 --- a/server/src/__tests__/company-skills-service.test.ts +++ b/server/src/__tests__/company-skills-service.test.ts @@ -21,6 +21,7 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { companySkillService } from "../services/company-skills.ts"; +import { removeRuntimeSkillCache } from "../services/runtime-skill-cache.js"; import { folderService } from "../services/folders.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -62,6 +63,9 @@ describeEmbeddedPostgres("companySkillService.list", () => { }, 20_000); afterEach(async () => { + for (const skill of await db.select().from(companySkills)) { + await removeRuntimeSkillCache(path.join(paperclipHome!, "instances", "default", "skills", skill.companyId), skill.id); + } await db.delete(agents); await db.delete(companySkills); await db.delete(projectWorkspaces); @@ -84,6 +88,125 @@ describeEmbeddedPostgres("companySkillService.list", () => { await tempDb?.cleanup(); }); + async function createPinnedRuntimeFixture() { + const companyId = randomUUID(); + const skillId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Cache tests", issuePrefix: `T${companyId.slice(0, 6)}` }); + await db.insert(companySkills).values({ + id: skillId, companyId, key: `company/${companyId}/cached`, slug: "cached", name: "Cached", + markdown: "# Installed skill", sourceType: "github", sourceLocator: "https://github.com/acme/cache", + sourceRef: "a".repeat(40), trustLevel: "markdown_only", compatibility: "compatible", + metadata: { owner: "acme", repo: "cache", repoSkillDir: ".", trackingRef: "main" }, + fileInventory: [{ path: "SKILL.md", kind: "skill" }, + ...Array.from({ length: 20 }, (_, index) => ({ path: `references/${index}.md`, kind: "reference" as const }))], + }); + return { companyId, skillId, key: `company/${companyId}/cached` }; + } + + it("refreshes once, reuses pinned runtime contents across service instances, and ignores cosmetic changes", async () => { + const { companyId, skillId, key } = await createPinnedRuntimeFixture(); + const upstream = vi.fn(async (url: string | URL) => new Response(String(url))); + vi.stubGlobal("fetch", upstream); + const select = vi.spyOn(db, "select"); + try { + const cold = (await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === key)!; + expect(cold.sourceStatus).toBe("available"); + expect(select.mock.calls.filter(([fields]) => fields && Object.keys(fields).length === 1 && fields.id === companies.id)).toHaveLength(1); + expect(upstream).toHaveBeenCalledTimes(21); + expect(new Set(upstream.mock.calls.map(([url]) => String(url))).size).toBe(21); + const before = await fs.stat(path.join(cold.source, "SKILL.md")); + upstream.mockClear(); + upstream.mockRejectedValue(new Error("Upstream outage")); + await db.update(companySkills).set({ name: "New display label", updatedAt: new Date(), metadata: { owner: "acme", repo: "cache", repoSkillDir: ".", trackingRef: "main", starred: true } }).where(eq(companySkills.id, skillId)); + const warm = (await companySkillService(db).listRuntimeSkillEntries(companyId)).find((entry) => entry.key === key)!; + expect(warm).toEqual(cold); + expect(upstream).not.toHaveBeenCalled(); + expect((await fs.stat(path.join(warm.source, "SKILL.md"))).mtimeMs).toBe(before.mtimeMs); + const readOnly = (await svc.listRuntimeSkillEntries(companyId, { materializeMissing: false })).find((entry) => entry.key === key)!; + expect(readOnly).toEqual(warm); + await db.update(companySkills).set({ sourceRef: "b".repeat(40) }).where(eq(companySkills.id, skillId)); + expect((await svc.listRuntimeSkillEntries(companyId, { materializeMissing: false })).find((entry) => entry.key === key)!.sourceStatus).toBe("missing"); + expect(upstream).not.toHaveBeenCalled(); + // An unavailable newly installed revision must never use the older cache. + expect((await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === key)!.sourceStatus).toBe("missing"); + expect(await fs.readFile(path.join(cold.source, "references/0.md"), "utf8")).toContain("a".repeat(40)); + upstream.mockImplementation(async (url) => new Response(String(url))); + const next = (await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === key)!; + expect(next.sourceStatus).toBe("available"); + expect(next.source).not.toBe(cold.source); + expect(await fs.readFile(path.join(next.source, "references/0.md"), "utf8")).toContain("b".repeat(40)); + expect(await fs.readFile(path.join(cold.source, "references/0.md"), "utf8")).toContain("a".repeat(40)); + } finally { select.mockRestore(); vi.unstubAllGlobals(); } + }); + + it("deduplicates runtime downloads for twenty concurrent service callers and isolates companies", async () => { + const first = await createPinnedRuntimeFixture(); + const second = await createPinnedRuntimeFixture(); + const upstream = vi.fn(async (url: string | URL) => new Response(String(url))); + vi.stubGlobal("fetch", upstream); + try { + const batches = await Promise.all(Array.from({ length: 20 }, () => companySkillService(db).listRuntimeSkillEntries(first.companyId))); + const sources = batches.map((entries) => entries.find((entry) => entry.key === first.key)!); + expect(sources.every((entry) => entry.sourceStatus === "available")).toBe(true); + expect(new Set(sources.map((entry) => entry.source)).size).toBe(1); + expect(upstream).toHaveBeenCalledTimes(21); + const other = (await svc.listRuntimeSkillEntries(second.companyId)).find((entry) => entry.key === second.key)!; + expect(other.source).not.toBe(sources[0].source); + expect(upstream).toHaveBeenCalledTimes(42); + } finally { vi.unstubAllGlobals(); } + }); + + it("keeps upstream changes dormant until explicit update and refreshes supporting-only changes", async () => { + const companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Update cache", issuePrefix: `T${companyId.slice(0, 6)}` }); + let revision = "a".repeat(40); + const markdown = "---\nname: cached\ndescription: A fixture\n---\n# Unchanged skill\n"; + const upstream = vi.fn(async (input: string | URL) => { + const url = String(input); + if (url.includes("/commits/")) return Response.json({ sha: revision }); + if (url.includes("/git/trees/")) return Response.json({ tree: [{ path: "cached/SKILL.md", type: "blob" }, { path: "cached/reference.md", type: "blob" }] }); + if (url.endsWith("/SKILL.md")) return new Response(markdown); + if (url.endsWith("/reference.md")) return new Response(url.includes("a".repeat(40)) ? "old supporting file" : "new supporting file"); + return Response.json({ default_branch: "main" }); + }); + vi.stubGlobal("fetch", upstream); + try { + const imported = await svc.importFromSource(companyId, "https://github.com/acme/cache"); + const skill = imported.imported[0]; + expect(skill.sourceRef).toBe(revision); + const old = (await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === skill.key)!; + revision = "b".repeat(40); + upstream.mockClear(); + expect((await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === skill.key)).toEqual(old); + expect(upstream).not.toHaveBeenCalled(); + const updated = await svc.installUpdate(companyId, skill.id); + expect(updated?.sourceRef).toBe(revision); + expect(updated?.markdown).toBe(markdown); + const next = (await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === skill.key)!; + expect(next.source).not.toBe(old.source); + expect(await fs.readFile(path.join(next.source, "reference.md"), "utf8")).toBe("new supporting file"); + expect(await fs.readFile(path.join(old.source, "reference.md"), "utf8")).toBe("old supporting file"); + const lockFailure = vi.spyOn(fs, "link").mockRejectedValueOnce(new Error("Publication lock unavailable")); + try { + await expect(svc.deleteSkill(companyId, skill.id)).rejects.toThrow("Publication lock unavailable"); + expect(await svc.getById(companyId, skill.id)).not.toBeNull(); + } finally { lockFailure.mockRestore(); } + await svc.deleteSkill(companyId, skill.id); + await expect(fs.stat(path.dirname(path.dirname(next.source)))).rejects.toMatchObject({ code: "ENOENT" }); + } finally { vi.unstubAllGlobals(); } + }); + + it("observes edits to a direct local source on the next preparation", async () => { + const companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Local cache", issuePrefix: `T${companyId.slice(0, 6)}` }); + const skill = await svc.createLocalSkill(companyId, { name: "Local source", slug: "local-source" }); + const first = (await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === skill.key)!; + await fs.appendFile(path.join(first.source, "SKILL.md"), "\nNew local instructions\n"); + const next = (await svc.listRuntimeSkillEntries(companyId)).find((entry) => entry.key === skill.key)!; + expect(next.source).toBe(first.source); + expect(await fs.readFile(path.join(next.source, "SKILL.md"), "utf8")).toContain("New local instructions"); + }); + it("lists skills without exposing markdown content", async () => { const companyId = randomUUID(); const skillId = randomUUID(); @@ -3070,7 +3193,11 @@ describeEmbeddedPostgres("companySkillService.list", () => { await fs.mkdir(oldRuntimeDir, { recursive: true }); await fs.writeFile(path.join(oldRuntimeDir, "SKILL.md"), "# stale\n", "utf8"); + const revisionCacheRoot = path.join(managedRoot, "__runtime_cache_v1__", skill.id); + await fs.mkdir(revisionCacheRoot, { recursive: true }); + await fs.writeFile(path.join(revisionCacheRoot, "old-cache"), "stale"); await svc.renameSkill(companyId, skill.id, { name: "Runtime Skill", slug: "runtime-renamed" }); + await expect(fs.stat(revisionCacheRoot)).rejects.toMatchObject({ code: "ENOENT" }); await expect(fs.stat(oldRuntimeDir)).rejects.toMatchObject({ code: "ENOENT" }); }); diff --git a/server/src/__tests__/runtime-skill-cache.test.ts b/server/src/__tests__/runtime-skill-cache.test.ts new file mode 100644 index 0000000000..3ee8e97f1b --- /dev/null +++ b/server/src/__tests__/runtime-skill-cache.test.ts @@ -0,0 +1,174 @@ +import { randomUUID } from "node:crypto"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { CompanySkill } from "@paperclipai/shared"; +import { removeRuntimeSkillCache, resolveRuntimeSkillCache, runtimeSkillCacheSpec } from "../services/runtime-skill-cache.js"; + +async function makeWritable(root: string): Promise { + const stat = await fs.lstat(root); + if (!stat.isDirectory() || stat.isSymbolicLink()) return; + await fs.chmod(root, 0o700); + for (const item of await fs.readdir(root)) await makeWritable(path.join(root, item)); +} + +async function makeReadonly(root: string): Promise { + const stat = await fs.lstat(root); + if (stat.isSymbolicLink()) return; + if (stat.isDirectory()) { + for (const item of await fs.readdir(root)) await makeReadonly(path.join(root, item)); + } + await fs.chmod(root, stat.isDirectory() ? 0o555 : 0o444); +} + +describe("runtime skill revision cache", () => { + let root: string; + let skill: CompanySkill; + const contents: Record = { "SKILL.md": "# Test", "references/a.md": "Supporting content" }; + beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "runtime-cache-")); + skill = { id: randomUUID(), companyId: randomUUID(), sourceType: "github", sourceLocator: "https://github.com/org/repo", + sourceRef: "a".repeat(40), slug: "test", markdown: contents["SKILL.md"], metadata: { owner: "org", repo: "repo", repoSkillDir: "skills/test" }, + fileInventory: [{ path: "SKILL.md", kind: "skill" }, { path: "references/a.md", kind: "reference" }], + } as CompanySkill; + }); + afterEach(async () => { await makeWritable(root); await fs.rm(root, { recursive: true, force: true }); }); + const reader = () => vi.fn(async (file: string) => contents[file]); + + it("publishes complete contents once for twenty callers and leaves warm files untouched", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + const read = reader(); + const sources = await Promise.all(Array.from({ length: 20 }, () => resolveRuntimeSkillCache(spec, read))); + expect(new Set(sources).size).toBe(1); + expect(read).toHaveBeenCalledTimes(2); + expect((await fs.stat(sources[0]!)).mode & 0o222).toBe(0); + expect((await fs.stat(path.join(sources[0]!, "SKILL.md"))).mode & 0o222).toBe(0); + const before = await fs.stat(path.join(sources[0]!, "SKILL.md")); + read.mockRejectedValue(new Error("Upstream offline")); + expect(await resolveRuntimeSkillCache(runtimeSkillCacheSpec(root, { ...skill })!, read)).toBe(sources[0]); + expect(await resolveRuntimeSkillCache(spec, read, false)).toBe(sources[0]); + expect((await fs.stat(path.join(sources[0]!, "SKILL.md"))).mtimeMs).toBe(before.mtimeMs); + expect(await fs.readdir(sources[0]!)).toEqual(["SKILL.md", "references"]); + expect(read).toHaveBeenCalledTimes(2); + }); + + it("reuses the winning complete directory across concurrent processes and a process restart", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + const launch = () => new Promise<{ source: string; reads: number }>((resolve, reject) => { + const code = ` + import { resolveRuntimeSkillCache } from ${JSON.stringify(new URL("../services/runtime-skill-cache.ts", import.meta.url).href)}; + let reads = 0; + const source = await resolveRuntimeSkillCache(${JSON.stringify(spec)}, async (file) => { + reads++; + await new Promise(resolve => setTimeout(resolve, 30)); + return ${JSON.stringify(contents)}[file]; + }); + console.log(JSON.stringify({ source, reads })); + `; + const child = spawn(process.execPath, ["--import", fileURLToPath(new URL("../../node_modules/tsx/dist/loader.mjs", import.meta.url)), "--input-type=module", "-e", code], { stdio: ["ignore", "pipe", "pipe"] }); + let output = "", errors = ""; + child.stdout.on("data", (chunk) => { output += chunk; }); + child.stderr.on("data", (chunk) => { errors += chunk; }); + child.on("error", reject); + child.on("exit", (exitCode) => exitCode === 0 ? resolve(JSON.parse(output)) : reject(new Error(errors))); + }); + const [first, second] = await Promise.all([launch(), launch()]); + expect(first.source).toBe(second.source); + expect(await fs.readdir(spec.root)).toEqual([spec.fingerprint]); + expect(await launch()).toEqual({ source: first.source, reads: 0 }); + }); + + it("fingerprints installed content and ownership, excluding cosmetic metadata", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + expect(runtimeSkillCacheSpec(root, { ...skill, name: "Renamed", updatedAt: new Date(), metadata: { ...skill.metadata, starred: true, ref: "new-branch" } })!.fingerprint).toBe(spec.fingerprint); + expect(runtimeSkillCacheSpec(root, { ...skill, fileInventory: [...skill.fileInventory].reverse() })!.fingerprint).toBe(spec.fingerprint); + for (const update of [{ companyId: randomUUID() }, { sourceRef: "b".repeat(40) }, { markdown: "changed" }, + { metadata: { ...skill.metadata, repoSkillDir: "other" } }]) { + expect(runtimeSkillCacheSpec(root, { ...skill, ...update })!.fingerprint).not.toBe(spec.fingerprint); + } + expect(runtimeSkillCacheSpec(root, { ...skill, sourceRef: "main" })).toBeNull(); + }); + + it("retains the old revision when publishing an explicit update", async () => { + const old = await resolveRuntimeSkillCache(runtimeSkillCacheSpec(root, skill)!, reader()); + const next = await resolveRuntimeSkillCache(runtimeSkillCacheSpec(root, { ...skill, sourceRef: "b".repeat(40) })!, async () => "updated"); + expect(next).not.toBe(old); + expect(await fs.readFile(path.join(old!, "references/a.md"), "utf8")).toBe(contents["references/a.md"]); + expect(await fs.readFile(path.join(next!, "references/a.md"), "utf8")).toBe("updated"); + }); + + it.each(["manifest-missing", "manifest-malformed", "changed", "deleted", "extra", "symlink"])("rejects %s without read-only repair, then rebuilds", async (corruption) => { + const spec = runtimeSkillCacheSpec(root, skill)!; + const source = (await resolveRuntimeSkillCache(spec, reader()))!; + await makeWritable(spec.entry); + await fs.chmod(path.join(source, "SKILL.md"), 0o600); + await fs.chmod(path.join(spec.entry, "manifest.json"), 0o600); + const manifest = path.join(spec.entry, "manifest.json"); + if (corruption === "manifest-missing") await fs.unlink(manifest); + if (corruption === "manifest-malformed") await fs.writeFile(manifest, "{bad"); + if (corruption === "changed") await fs.writeFile(path.join(source, "SKILL.md"), "tampered"); + if (corruption === "deleted") await fs.unlink(path.join(source, "SKILL.md")); + if (corruption === "extra") await fs.writeFile(path.join(source, "extra.txt"), "extra"); + if (corruption === "symlink") { + await fs.unlink(path.join(source, "SKILL.md")); + await fs.symlink(path.join(root, "outside.txt"), path.join(source, "SKILL.md")); + await fs.writeFile(path.join(root, "outside.txt"), "outside"); + } + await makeReadonly(spec.entry); + const read = reader(); + expect(await resolveRuntimeSkillCache(spec, read, false)).toBeNull(); + expect(read).not.toHaveBeenCalled(); + expect(await resolveRuntimeSkillCache(spec, read)).toBe(source); + expect(read).toHaveBeenCalledTimes(2); + expect(await fs.readFile(path.join(source, "SKILL.md"), "utf8")).toBe(contents["SKILL.md"]); + if (corruption === "symlink") expect(await fs.readFile(path.join(root, "outside.txt"), "utf8")).toBe("outside"); + }); + + it("does not publish partial builds and retries after a failed upstream read", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + const read = reader().mockRejectedValueOnce(new Error("offline")); + await expect(resolveRuntimeSkillCache(spec, read)).rejects.toThrow("offline"); + expect(await fs.readdir(spec.root)).toEqual([]); + expect(await resolveRuntimeSkillCache(spec, read, false)).toBeNull(); + expect(await resolveRuntimeSkillCache(spec, read)).toBe(path.join(spec.entry, "files")); + }); + + it("coordinates cleanup with an active build and prevents a removed skill from being republished", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + let installed = true; + let release!: () => void; + let entered!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + const started = new Promise((resolve) => { entered = resolve; }); + const build = resolveRuntimeSkillCache(spec, async (file) => { + entered(); await gate; return contents[file]; + }, true, async () => installed); + const rejected = expect(build).rejects.toThrow("renamed or removed"); + await started; + installed = false; + const removal = removeRuntimeSkillCache(root, skill.id); + release(); + await rejected; + await removal; + await expect(fs.stat(spec.root)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(resolveRuntimeSkillCache(spec, reader(), true, async () => installed)).rejects.toThrow("renamed or removed"); + await expect(fs.stat(spec.root)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it.each(["../escape", "/absolute", "a/../../escape", "a/../b", "C:\\escape", "a\\..\\escape"])("rejects traversal %s", (file) => { + expect(() => runtimeSkillCacheSpec(root, { ...skill, fileInventory: [...skill.fileInventory, { path: file, kind: "reference" }] })).toThrow("Invalid runtime skill file path"); + }); + + it("rejects a symlink cache ancestor before invoking the source reader", async () => { + const spec = runtimeSkillCacheSpec(root, skill)!; + const outside = await fs.mkdtemp(path.join(root, "outside-")); + await fs.symlink(outside, path.dirname(spec.root)); + const read = reader(); + await expect(resolveRuntimeSkillCache(spec, read)).rejects.toThrow("Unsafe"); + expect(read).not.toHaveBeenCalled(); + expect(await fs.readdir(outside)).toEqual([]); + }); +}); diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index 4b432dcd4f..25ecb3bc1f 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -1,3 +1,5 @@ +import { logger } from "../middleware/logger.js"; +import { removeRuntimeSkillCache, resolveRuntimeSkillCache, runtimeSkillCacheSpec } from "./runtime-skill-cache.js"; import { createHash, randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; import path from "node:path"; @@ -3149,10 +3151,10 @@ export function companySkillService(db: Db) { continue; } - await db - .delete(companySkills) - .where(eq(companySkills.id, skill.id)); - await fs.rm(resolveRuntimeSkillMaterializedPath(companyId, skill), { recursive: true, force: true }); + await removeRuntimeSkillCache(resolveManagedSkillsRoot(companyId), skill.id, async () => { + await fs.rm(resolveRuntimeSkillMaterializedPath(companyId, skill), { recursive: true, force: true }); + await db.delete(companySkills).where(eq(companySkills.id, skill.id)); + }); } } @@ -4152,12 +4154,14 @@ export function companySkillService(db: Db) { throw error; } - // Remove the stale runtime materialization so runtime sync recreates it - // under the new key/slug. - await fs.rm( - path.resolve(managedRoot, "__runtime__", buildSkillRuntimeName(previousKey, previousSlug)), - { recursive: true, force: true }, - ); + // The rename has committed. Cache cleanup must not make a successful rename appear to fail. + try { + await fs.rm(path.resolve(managedRoot, "__runtime__", buildSkillRuntimeName(previousKey, previousSlug)), + { recursive: true, force: true }); + await removeRuntimeSkillCache(managedRoot, skill.id); + } catch (error) { + logger.warn({ err: error, companyId, skillId: skill.id }, "Skill renamed; obsolete runtime cache cleanup failed"); + } const renamed = await getById(companyId, skill.id); if (!renamed) throw notFound("Renamed skill not found"); @@ -4273,6 +4277,10 @@ export function companySkillService(db: Db) { const skill = await getById(companyId, skillId); if (!skill) return null; + return readLoadedSkillFile(skill, relativePath); + } + + async function readLoadedSkillFile(skill: CompanySkill, relativePath: string): Promise { const normalizedPath = normalizePortablePath(relativePath || "SKILL.md"); const fileEntry = skill.fileInventory.find((entry) => entry.path === normalizedPath); if (!fileEntry) { @@ -5672,10 +5680,12 @@ export function companySkillService(db: Db) { let wroteSkillFile = false; for (const entry of skill.fileInventory) { const normalizedPath = normalizePortablePath(entry.path); - const detail = await readFile(companyId, skill.id, normalizedPath).catch(() => null); + const detail = await readLoadedSkillFile(skill, normalizedPath); const content = detail?.content ?? (normalizedPath === "SKILL.md" ? skill.markdown : null); - if (content === null) continue; - const targetPath = path.resolve(skillDir, entry.path); + if (content === null) throw unprocessable("Declared skill file is unavailable"); + const resolved = resolveVersionSnapshotPath(skillDir, entry.path); + if (!resolved) throw unprocessable("Invalid skill file path"); + const targetPath = resolved.targetPath; await fs.mkdir(path.dirname(targetPath), { recursive: true }); await fs.writeFile(targetPath, content, "utf8"); if (normalizedPath === "SKILL.md") wroteSkillFile = true; @@ -5817,6 +5827,24 @@ export function companySkillService(db: Db) { const source = await resolveExistingSkillDirectory(normalizeSkillDirectory(skill)); if (source) return { status: "available", source }; + try { + const cache = runtimeSkillCacheSpec(resolveManagedSkillsRoot(companyId), skill); + if (cache) { + const cachedSource = await resolveRuntimeSkillCache(cache, + async (relativePath) => (await readLoadedSkillFile(skill, relativePath)).content, + options.materializeMissing !== false, + async () => (await getById(companyId, skill.id))?.key === skill.key); + return cachedSource + ? { status: "available", source: cachedSource } + : { status: "missing", source: path.join(cache.entry, "files"), detail: buildMissingRuntimeSourceDetail(skill) }; + } + } catch (error) { + return { + status: "missing", source: resolveRuntimeSkillMaterializedPath(companyId, skill), + detail: `Failed to materialize skill files: ${error instanceof Error ? error.message : String(error)}`, + }; + } + if (options.materializeMissing === false) { const materializedPath = resolveRuntimeSkillMaterializedPath(companyId, skill); const materializedSource = await resolveExistingSkillDirectory(materializedPath); @@ -6937,13 +6965,12 @@ export function companySkillService(db: Db) { ); } - // Delete DB row - await db - .delete(companySkills) - .where(eq(companySkills.id, skillId)); - - // Clean up materialized runtime files - await fs.rm(resolveRuntimeSkillMaterializedPath(companyId, skill), { recursive: true, force: true }); + // Take the cache lifecycle lock before deleting the row. A busy publisher must not + // turn a committed deletion into an apparent API failure, nor recreate its cache. + await removeRuntimeSkillCache(resolveManagedSkillsRoot(companyId), skill.id, async () => { + await fs.rm(resolveRuntimeSkillMaterializedPath(companyId, skill), { recursive: true, force: true }); + await db.delete(companySkills).where(eq(companySkills.id, skillId)); + }); return skill; } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6d85f99e8c..350877a00e 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -158,7 +158,7 @@ import { assertManagedProfileRecoveryBinding, resolvePaperclipRunnerNativeProviderInput, } from "./native-runtime/provider-profile.js"; -import type { NativeRunHistoricalSpan } from "./native-runtime/native-run-trace.js"; +import { recordFailedSkillPreparation, type NativeRunHistoricalSpan } from "./native-runtime/native-run-trace.js"; import { parseNativeExecutionInput, type NativeExecutionInput, @@ -18916,18 +18916,39 @@ export function heartbeatService( const runtimeSkillPreference = readPaperclipSkillSyncPreference( effectiveResolvedConfig, ); - const runtimeSkillEntries = await companySkills.listRuntimeSkillEntries( - agent.companyId, - { - versionSelections: skillVersionSelectionMap( - runtimeSkillPreference.desiredSkillEntries, + const nativeRunnerPreparationSpans: NativeRunHistoricalSpan[] = []; + const skillsPrepareStartedAtMs = Date.now(); + const runtimeSkillEntries = await (async () => { + try { + return await companySkills.listRuntimeSkillEntries( + agent.companyId, { - versionPinsEnabled: - resolvedInstanceSettings.experimental.enableBetaSkills === true, + versionSelections: skillVersionSelectionMap( + runtimeSkillPreference.desiredSkillEntries, + { + versionPinsEnabled: + resolvedInstanceSettings.experimental.enableBetaSkills === true, + }, + ), }, - ), - }, - ); + ); + } catch (error) { + if (agent.adapterType === "paperclip_runner") { + await recordFailedSkillPreparation({ + runId: run.id, + startedAtMs: skillsPrepareStartedAtMs, + onEvent: async (event) => { await appendRunEvent(run, event); }, + }); + } + throw error; + } + })(); + nativeRunnerPreparationSpans.push({ + name: "skills.prepare", + parentName: "task.prepare", + startedAtMs: skillsPrepareStartedAtMs, + endedAtMs: Date.now(), + }); let runtimeConfig: Record = { ...effectiveResolvedConfig, paperclipRuntimeSkills: runtimeSkillEntries, @@ -19611,7 +19632,6 @@ export function heartbeatService( }) .where(eq(heartbeatRuns.id, run.id)); } - const nativeRunnerPreparationSpans: NativeRunHistoricalSpan[] = []; const environmentAcquireStartedAtMs = Date.now(); let acquiredEnvironment: Awaited< ReturnType diff --git a/server/src/services/native-runtime/native-run-trace.test.ts b/server/src/services/native-runtime/native-run-trace.test.ts index 88c96b58be..9f4ad5efad 100644 --- a/server/src/services/native-runtime/native-run-trace.test.ts +++ b/server/src/services/native-runtime/native-run-trace.test.ts @@ -5,6 +5,7 @@ import { getActiveStepContext } from "@paperclipai/adapter-utils/acpx-engine/sta import type { StartupTraceContextHandle } from "../../instrumentation.js"; import { createNativeRunTrace, + recordFailedSkillPreparation, NATIVE_RUN_SPAN_EVENT_TYPE, NATIVE_RUN_TRACE_SCHEMA_VERSION, } from "./native-run-trace.js"; @@ -128,6 +129,29 @@ describe("native runner performance trace", () => { }); }); + it("records skills.prepare beneath preparation with no attributes and tolerates a failed log sink", async () => { + const { traceContext, spans } = createRecordingTraceContext(); + const trace = createNativeRunTrace({ runId: "skills-run", startedAtMs: 100, + traceContext, onEvent: async () => { throw new Error("run log unavailable"); } }); + const preparation = trace.start("task.prepare", { parentName: "task.run", startedAtMs: 100 }); + await expect(trace.record({ name: "skills.prepare", parentName: "task.prepare", startedAtMs: 120, endedAtMs: 170 })).resolves.toBeUndefined(); + await trace.end(preparation, { endedAtMs: 200 }); + expect(spans.find((span) => span.name === "skills.prepare")).toMatchObject({ name: "skills.prepare", parentName: "task.prepare", endedAtMs: 170 }); + await expect(trace.finish("ok")).resolves.toBeUndefined(); + }); + + it("emits failed skill preparation without starting execution, even when the log sink fails", async () => { + const events: AdapterRuntimeEvent[] = []; + const { traceContext, spans } = createRecordingTraceContext(); + await recordFailedSkillPreparation({ runId: "failed-skills", startedAtMs: Date.now() - 10, + traceContext, onEvent: async (event) => { events.push(event); throw new Error("log unavailable"); } }); + expect(events.find((event) => event.payload?.span === "skills.prepare")?.payload).toMatchObject({ + span: "skills.prepare", parentSpan: "task.prepare", outcome: "failed", + }); + expect(spans.find((span) => span.name === "skills.prepare")?.parentName).toBe("task.prepare"); + expect(spans.some((span) => span.name === "native.session.execute")).toBe(false); + }); + it("never fails runner control flow when its event sink fails", async () => { const trace = createNativeRunTrace({ runId: "run-1", diff --git a/server/src/services/native-runtime/native-run-trace.ts b/server/src/services/native-runtime/native-run-trace.ts index 796533ee7e..e923bdd5b0 100644 --- a/server/src/services/native-runtime/native-run-trace.ts +++ b/server/src/services/native-runtime/native-run-trace.ts @@ -431,3 +431,22 @@ export function createNativeRunTrace(input: { } export type NativeRunTrace = ReturnType; + +/** Emit preparation failure even when execution aborts before a native session exists. */ +export async function recordFailedSkillPreparation(input: { + runId: string; + startedAtMs: number; + onEvent?: NativeRunTraceSink; + traceContext?: StartupTraceContextHandle; +}): Promise { + try { + const trace = createNativeRunTrace(input); + const preparation = trace.start("task.prepare", { parentName: "task.run", startedAtMs: input.startedAtMs }); + const endedAtMs = Date.now(); + await trace.record({ name: "skills.prepare", parentName: "task.prepare", startedAtMs: input.startedAtMs, endedAtMs, outcome: "failed" }); + await trace.end(preparation, { endedAtMs, outcome: "failed" }); + await trace.finish("failed"); + } catch { + // Diagnostics must not replace the original preparation error. + } +} diff --git a/server/src/services/native-runtime/runtime-context.test.ts b/server/src/services/native-runtime/runtime-context.test.ts index 85e6f5df29..c5a051c949 100644 --- a/server/src/services/native-runtime/runtime-context.test.ts +++ b/server/src/services/native-runtime/runtime-context.test.ts @@ -225,6 +225,8 @@ describe("buildNativeRuntimeContext", () => { .toBe("Follow the agent instructions.\n"); expect(await readFile(path.join(context.instructions.bundle.rootPath, "references", "policy.md"), "utf8")) .toBe("Company policy sibling.\n"); + const unselected = await buildNativeRuntimeContext({ ...input, runtimeConfig: {} }); + expect(unselected.skills).toEqual([]); expect(context.skills).toHaveLength(1); expect(context.skills[0]).toMatchObject({ key: "company-1/reviewer", diff --git a/server/src/services/runtime-skill-cache.ts b/server/src/services/runtime-skill-cache.ts new file mode 100644 index 0000000000..76503a6565 --- /dev/null +++ b/server/src/services/runtime-skill-cache.ts @@ -0,0 +1,227 @@ +import { createHash, randomUUID } from "node:crypto"; +import { constants, promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import type { CompanySkill } from "@paperclipai/shared"; + +const FORMAT = 1; +const inFlight = new Map>(); +const digest = (value: string | Buffer) => createHash("sha256").update(value).digest("hex"); + +type FileRecord = { path: string; size: number; digest: string }; +type CacheSpec = { root: string; entry: string; fingerprint: string; paths: string[] }; + +function filePath(value: string): string { + const normalized = value.replace(/\\/g, "/"); + if (!normalized || normalized.startsWith("/") || /^[a-z]:/i.test(normalized) + || normalized.split("/").some((part) => !part || part === "." || part === "..") + || normalized.includes("\0")) throw new Error("Invalid runtime skill file path"); + return normalized; +} + +export function runtimeSkillCacheRoot(managedRoot: string, skillId: string): string { + if (!/^[a-zA-Z0-9_-]+$/.test(skillId)) throw new Error("Invalid runtime skill ID"); + // A sibling of __runtime__: old runtime cleanup cannot remove published revisions. + return path.resolve(managedRoot, `__runtime_cache_v${FORMAT}__`, skillId); +} + +export function runtimeSkillCacheSpec(managedRoot: string, skill: CompanySkill): CacheSpec | null { + if ((skill.sourceType === "github" || skill.sourceType === "skills_sh") + && !/^[a-f0-9]{40}$/i.test(skill.sourceRef ?? "")) return null; + const metadata = skill.metadata ?? {}; + const inventory = skill.fileInventory.map((entry) => ({ path: filePath(entry.path), kind: entry.kind })) + .sort((a, b) => a.path.localeCompare(b.path)); + const paths = inventory.map((entry) => entry.path); + if (!paths.includes("SKILL.md")) throw new Error("Company skill could not be materialized because its stored SKILL.md copy is missing."); + if (new Set(paths).size !== paths.length) throw new Error("Invalid runtime skill file inventory"); + const fingerprint = digest(JSON.stringify({ + format: FORMAT, companyId: skill.companyId, skillId: skill.id, + sourceType: skill.sourceType, sourceLocator: skill.sourceLocator, sourceRef: skill.sourceRef, + // These are the only metadata fields used by the source reader. slug is its fallback directory. + source: { owner: metadata.owner, repo: metadata.repo, hostname: metadata.hostname, + ref: skill.sourceRef ? undefined : metadata.ref, repoSkillDir: metadata.repoSkillDir, + fallbackDirectory: typeof metadata.repoSkillDir === "string" ? undefined : skill.slug }, + markdown: digest(skill.markdown), inventory, + })); + const root = runtimeSkillCacheRoot(managedRoot, skill.id); + return { root, entry: path.join(root, fingerprint), fingerprint, paths }; +} + +// Check every ancestor before traversing it, including the configured cache root. +async function assertDirectories(directory: string, trustedRoot: string, create = false): Promise { + const absolute = path.resolve(directory); + let cursor = path.resolve(trustedRoot); + if (!absolute.startsWith(`${cursor}${path.sep}`)) throw new Error("Runtime cache escaped its root"); + // Ancestors of the configured storage root may be system aliases (e.g. macOS /var). + if (create) await fs.mkdir(cursor, { recursive: true }); + const rootStat = await fs.lstat(cursor); + if (!rootStat.isDirectory() || rootStat.isSymbolicLink()) throw new Error("Unsafe runtime skill storage root"); + for (const part of absolute.slice(cursor.length).split(path.sep).filter(Boolean)) { + cursor = path.join(cursor, part); + if (create) await fs.mkdir(cursor).catch((error: NodeJS.ErrnoException) => { + if (error.code !== "EEXIST") throw error; + }); + const stat = await fs.lstat(cursor); + if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("Unsafe runtime skill cache directory"); + } +} + +async function readRegularFile(filename: string): Promise { + const handle = await fs.open(filename, constants.O_RDONLY | constants.O_NOFOLLOW); + try { + if (!(await handle.stat()).isFile()) throw new Error("Unsafe runtime skill cache file"); + return await handle.readFile(); + } finally { await handle.close(); } +} + +async function inventory(directory: string, base = ""): Promise { + const out: string[] = []; + if ((await fs.lstat(directory)).mode & 0o222) throw new Error("Writable runtime skill cache directory"); + for (const entry of await fs.readdir(directory, { withFileTypes: true })) { + const relative = base ? `${base}/${entry.name}` : entry.name; + if (entry.isDirectory()) out.push(...await inventory(path.join(directory, entry.name), relative)); + else if (entry.isFile()) out.push(relative); + else throw new Error("Symlink or special file in runtime skill cache"); + } + return out.sort(); +} + +async function matches(spec: CacheSpec, entry = spec.entry): Promise { + try { + await assertDirectories(path.join(entry, "files"), path.dirname(path.dirname(spec.root))); + const manifest = JSON.parse((await readRegularFile(path.join(entry, "manifest.json"))).toString("utf8")); + if (manifest.format !== FORMAT || manifest.fingerprint !== spec.fingerprint || !Array.isArray(manifest.files) + || manifest.files.length !== spec.paths.length) return false; + const actual = await inventory(path.join(entry, "files")); + if (JSON.stringify(actual) !== JSON.stringify([...spec.paths].sort())) return false; + const seen = new Set(); + for (const record of manifest.files as FileRecord[]) { + if (!record || typeof record.path !== "string" || filePath(record.path) !== record.path + || !spec.paths.includes(record.path) || seen.has(record.path) + || !Number.isSafeInteger(record.size) || record.size < 0 || !/^[a-f0-9]{64}$/.test(record.digest)) return false; + seen.add(record.path); + const content = await readRegularFile(path.join(entry, "files", record.path)); + if ((await fs.lstat(path.join(entry, "files", record.path))).mode & 0o222) return false; + if (content.length !== record.size || digest(content) !== record.digest) return false; + } + return true; + } catch { return false; } +} + +// Serialize builds and cleanup for one skill across processes as well as callers. +// A hard link publishes complete lock ownership atomically; crashed owners are reported without stealing another publisher’s lock. +async function publishLocked(root: string, fingerprint: string, action: () => Promise): Promise { + const lock = path.join(root, `${fingerprint}.lock`); + const owner = path.join(root, `.owner-${randomUUID()}`); + await fs.writeFile(owner, JSON.stringify({ pid: process.pid, host: os.hostname() }), { flag: "wx" }); + let acquired = false; + try { + const deadline = Date.now() + 60_000; + while (!acquired) { + try { await fs.link(owner, lock); acquired = true; } + catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + const lockContent = await readRegularFile(lock).catch((readError: NodeJS.ErrnoException) => { + if (readError.code === "ENOENT") return null; + throw readError; + }); + if (!lockContent) continue; + const holder = JSON.parse(lockContent.toString("utf8")); + if (holder.host === os.hostname() && Number.isSafeInteger(holder.pid) && holder.pid > 0) { + try { process.kill(holder.pid, 0); } + catch (probeError) { + if ((probeError as NodeJS.ErrnoException).code === "ESRCH") { + throw new Error("Runtime skill cache publisher exited; remove its stale publication lock before retrying"); + } + } + } + if (Date.now() >= deadline) throw new Error("Runtime skill cache publisher is busy; retry preparation"); + await new Promise((resolve) => setTimeout(resolve, 20)); + } + } + return await action(); + } finally { + if (acquired) await fs.unlink(lock).catch(() => {}); + await fs.unlink(owner).catch(() => {}); + } +} + +async function setTreeMode(directory: string, readonly: boolean): Promise { + const stat = await fs.lstat(directory).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (!stat || stat.isSymbolicLink()) return; + if (!stat.isDirectory()) { + if (readonly && stat.isFile()) await fs.chmod(directory, 0o444); + return; + } + if (!readonly) await fs.chmod(directory, 0o700); + for (const entry of await fs.readdir(directory)) await setTreeMode(path.join(directory, entry), readonly); + if (readonly) await fs.chmod(directory, 0o555); +} + +async function removeTree(directory: string): Promise { + await setTreeMode(directory, false); + await fs.rm(directory, { recursive: true, force: true }); +} + +export async function resolveRuntimeSkillCache( + spec: CacheSpec, read: (relativePath: string) => Promise, materialize = true, + stillInstalled: () => Promise = async () => true, +): Promise { + if (await matches(spec)) return path.join(spec.entry, "files"); + if (!materialize) return null; + const active = inFlight.get(spec.entry); + if (active) return active; + const build = (async () => { + const namespace = path.dirname(spec.root); + await assertDirectories(namespace, path.dirname(namespace), true); + // The lock lives outside the skill directory, so cleanup cannot unlink an active lock. + return publishLocked(namespace, path.basename(spec.root), async () => { + if (!await stillInstalled()) throw new Error("Skill was renamed or removed during preparation"); + await assertDirectories(spec.root, path.dirname(namespace), true); + if (await matches(spec)) return path.join(spec.entry, "files"); + const staging = await fs.mkdtemp(path.join(spec.root, ".staging-")); + try { + await fs.mkdir(path.join(staging, "files")); + const files: FileRecord[] = []; + for (const relative of spec.paths) { + const content = Buffer.from(await read(relative), "utf8"); + const target = path.join(staging, "files", relative); + await fs.mkdir(path.dirname(target), { recursive: true }); + await fs.writeFile(target, content, { flag: "wx" }); + files.push({ path: relative, size: content.length, digest: digest(content) }); + } + await fs.writeFile(path.join(staging, "manifest.json"), JSON.stringify({ format: FORMAT, fingerprint: spec.fingerprint, files })); + await setTreeMode(staging, true); + if (!await matches(spec, staging)) throw new Error("Runtime skill cache validation failed"); + // Lifecycle mutations can update the DB while this builder owns the filesystem lock. + if (!await stillInstalled()) throw new Error("Skill was renamed or removed during preparation"); + await fs.rename(spec.entry, path.join(spec.root, `.invalid-${spec.fingerprint}-${randomUUID()}`)) + .catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") throw error; }); + await fs.rename(staging, spec.entry); + return path.join(spec.entry, "files"); + } finally { await removeTree(staging); } + }); + })(); + inFlight.set(spec.entry, build); + try { return await build; } finally { if (inFlight.get(spec.entry) === build) inFlight.delete(spec.entry); } +} + +export async function removeRuntimeSkillCache( + managedRoot: string, skillId: string, afterRemove?: () => Promise, +): Promise { + const root = runtimeSkillCacheRoot(managedRoot, skillId); + const namespace = path.dirname(root); + try { await assertDirectories(namespace, managedRoot, Boolean(afterRemove)); } + catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT" && !afterRemove) return; throw error; } + await publishLocked(namespace, skillId, async () => { + try { + await assertDirectories(root, managedRoot); + await removeTree(root); + } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } + // Commit deletion while builders remain excluded. Lock/cleanup failures leave the row intact. + await afterRemove?.(); + }); +}