From c723bb4dfcef2e6bcd479c7b78392d3506aaa63d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 8 Sep 2026 09:35:32 -0500 Subject: [PATCH] fix(skills): reuse validated runtime revisions during preparation (#13042) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip manages AI agents and prepares their runtime inputs before each turn. > - Shared company skills are part of those inputs for native and legacy adapters. > - Runtime materialization refreshed the full inventory again for every declared file. > - Remote skill directories were also downloaded and rebuilt on every turn. > - Measured preparation took 42–73 seconds while runner execution took 7–9 seconds. > - This change reads the inventory once and reuses validated installed revisions. > - Agents retain their selected skills while repeated preparation avoids upstream work. ## Linked Issues or Issue Description **What happened?** One 114-skill preparation performed 407 inventory refreshes, 48 directory rebuilds, and 388 GitHub file fetches. Reusing existing local copies took 151 ms. **Expected behavior** Each listing refreshes inventory once. Unchanged installed remote revisions reuse complete, validated local copies. Local edits remain visible. Explicit updates select new revisions. **Steps to reproduce** 1. Import GitHub skills with supporting files. 2. Run an agent turn, then run another with the same installed revisions. 3. Observe repeated inventory scans, downloads, and runtime directory replacement before execution. Related prior attempts: #2330 and #9268 (still open; #9268 last updated July 9). Those use a marker compared with `updatedAt`. This patch follows the required content validation, immutable revision, company isolation, atomic publication, and read-only semantics, and removes refresh-per-file multiplication. ## What Changed - Split public file reading from reading an already loaded skill. Runtime listing refreshes inventory once. - Add a company-scoped revision cache with file manifests outside the delivered skill directory. Fingerprints omit cosmetic metadata. - Validate exact file inventory, sizes, and hashes before warm reuse. Reject traversal and symlinks. Stage complete builds and serialize atomic publication across processes. - Preserve local/catalog direct sources, stored Markdown fallback, explicit version snapshots, and legacy mutable-ref compatibility. Report missing supporting files and keep older valid revisions readable. - Clean both runtime layouts on rename/removal and record `skills.prepare` under preparation timing. - Add service/cache regressions and an isolated 114-skill benchmark, including a new-process warm run. ## Verification - Final targeted skill-service/cache/trace validation: 86 tests pass (61 embedded-PostgreSQL service tests, 19 cache tests, 6 trace tests). Database tests executed rather than skipped. Focused skill routes, adapter selection, and native runtime context also pass. - `pnpm -r typecheck` and `pnpm build` pass locally at `22caa1fe4`. - The full `pnpm test:run` matrix passes on supported Linux CI at the final head: [CI run](https://github.com/paperclipai/paperclip/actions/runs/34236097762). Local full-suite execution encountered PostgreSQL startup contention, a random allocated-port boundary, and a socket hang-up; every affected suite passed on an isolated rerun. The interrupted local serialized run is not claimed as a complete local pass. - Repeatable benchmark: `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-skill-preparation.ts`. Mixed 114-skill inventory with 429 remote files on Linux: cold 286 ms, warm median 96 ms / maximum 153 ms including a new process. Every warm sample performs one refresh, zero upstream fetches/rebuilds, and reports no missing entries; content assertions pass. - Controlled deployment against the previously deployed revision completed with zero lost runs. Real inventory: 114 skills, 670 declared files; 402 cached files match the prior installed copies byte-for-byte. Ten post-deployment warm preparations: median 129 ms / maximum 208 ms; new-process warm 194 ms, zero downloads/rebuilds/missing entries. - Five sequential real browser questions persisted in 10.6–20.7 s (median 12.2 s), versus 50–83 s before. Skill preparation median 240 ms, with one 2.37 s outlier. Total preparation median 3.337 s / maximum 8.728 s **does not fully meet** the <3 s / <5 s target. The excluded historical-run redaction query takes about 1.36 s per scan at two preparation call sites; wider application latency coincided with the outlier, without a cache rebuild. These residuals are reported rather than discarded. - Disposable skill reimport verified through actual selected-skill runs: the next run read the changed code. Fixture removed and agent configuration verified unchanged. - Greptile 5/5, zero unresolved review threads, all final-head CI checks green. ## Risks - Cold preparation still requires upstream availability for supporting files. An unavailable revision is reported missing and never falls back to an older revision. - Valid older revisions and quarantined invalid entries consume additive disk space until skill cleanup. An abruptly killed publisher can leave a lock that requires operator cleanup after confirming its PID is dead. - Warm validation reads all cached file bytes. Very large inventories still have proportional local I/O cost. - No HTTP API, schema, agent configuration, or first-party Telemetry changes. OpenTelemetry retains its operator endpoint gate. ## Model Used OpenAI GPT-6 in Codex, with reasoning, repository inspection, code editing, and test execution. The exact serving snapshot and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass (targeted and isolated reruns; full Linux CI matrix passes, local full-run caveats above) - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/observability.md | 47 ++++ scripts/benchmark-skill-preparation.ts | 115 +++++++++ .../__tests__/company-skills-service.test.ts | 127 ++++++++++ .../src/__tests__/runtime-skill-cache.test.ts | 174 ++++++++++++++ server/src/services/company-skills.ts | 67 ++++-- server/src/services/heartbeat.ts | 44 +++- .../native-runtime/native-run-trace.test.ts | 24 ++ .../native-runtime/native-run-trace.ts | 19 ++ .../native-runtime/runtime-context.test.ts | 2 + server/src/services/runtime-skill-cache.ts | 227 ++++++++++++++++++ 10 files changed, 814 insertions(+), 32 deletions(-) create mode 100644 scripts/benchmark-skill-preparation.ts create mode 100644 server/src/__tests__/runtime-skill-cache.test.ts create mode 100644 server/src/services/runtime-skill-cache.ts 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?.(); + }); +}