mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
DottaandPaperclip
06689c1b34
Share each active company skill-inventory refresh before taking the version-cache lock. Read actor visibility under the lock and keep snapshot publication protected through durable run pinning. Preserve empty-selection behavior and the list API. Validation: 176 affected cases, full repository typecheck/build, fresh full CI/E2E, security and exact-head Apex 5/5 pass. The local six-reader latency experiment improves from 50.4s to 18.3s; live response latency is a separate acceptance check. Co-Authored-By: Paperclip <noreply@paperclip.ing>
285 lines
17 KiB
TypeScript
285 lines
17 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { promises as fs } from "node:fs";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import { eq } from "drizzle-orm";
|
|
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
|
import { agents, heartbeatRuns, issues, nativeRunFinalizations, companies, companySkills, companySkillVersions, companySkillRepositorySnapshots, createDb } from "@paperclipai/db";
|
|
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
|
import { companySkillService } from "../services/company-skills.js";
|
|
import { skillVersionCachePaths, touchSkillVersionCache, withSkillVersionCacheLock } from "../services/skill-version-cache.js";
|
|
|
|
import { skillVersionRetentionService } from "../services/skill-version-retention.js";
|
|
async function sweepSkillVersionCache(db: ReturnType<typeof createDb>, options: { companyId: string; now?: number; dryRun?: boolean }) {
|
|
const result = await skillVersionRetentionService(db).pruneCompany(options.companyId, { dryRun: options.dryRun, now: options.now === undefined ? undefined : new Date(options.now) });
|
|
return { ...result, evicted: result.removed };
|
|
}
|
|
|
|
const support = await getEmbeddedPostgresTestSupport();
|
|
const describePostgres = support.supported ? describe : describe.skip;
|
|
if (!support.supported) console.warn(`Skill cache tests require Postgres: ${support.reason}`);
|
|
|
|
describePostgres("immutable skill version cache", () => {
|
|
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
|
let db: ReturnType<typeof createDb>;
|
|
let home: string;
|
|
const previousHome = process.env.PAPERCLIP_HOME;
|
|
|
|
beforeAll(async () => {
|
|
database = await startEmbeddedPostgresTestDatabase("skill-version-cache-");
|
|
db = createDb(database.connectionString);
|
|
home = await fs.mkdtemp(path.join(os.tmpdir(), "skill-version-cache-"));
|
|
process.env.PAPERCLIP_HOME = home;
|
|
});
|
|
afterEach(() => { vi.unstubAllGlobals(); vi.restoreAllMocks(); });
|
|
afterAll(async () => {
|
|
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME;
|
|
else process.env.PAPERCLIP_HOME = previousHome;
|
|
if (home) await fs.rm(home, { recursive: true, force: true });
|
|
await database?.cleanup();
|
|
});
|
|
|
|
async function fixture() {
|
|
const companyId = randomUUID();
|
|
const skillId = randomUUID();
|
|
const key = `company/${companyId}/saved-repository`;
|
|
await db.insert(companies).values({ id: companyId, name: "Skill cache", issuePrefix: `S${companyId.slice(0, 6)}` });
|
|
await db.insert(companySkills).values({ id: skillId, companyId, key, slug: "saved-repository", name: "Saved repository",
|
|
markdown: "# Saved skill", sourceType: "github", sourceLocator: "https://github.com/example/unavailable",
|
|
metadata: { skillSourceId: randomUUID(), snapshotHash: "saved" } });
|
|
const versions = [];
|
|
for (const revision of [1, 2]) {
|
|
const [snapshot] = await db.insert(companySkillRepositorySnapshots).values({ companyId, contentHash: `revision-${revision}`,
|
|
fileInventory: [
|
|
{ path: "skills/example/SKILL.md", kind: "skill", content: `---\nname: saved-repository\ndescription: Saved revision ${revision}\n---\nRun ../../scripts/example.sh.\n` },
|
|
{ path: "scripts/example.sh", kind: "script", content: `#!/bin/sh\necho revision-${revision}\n`, executable: true },
|
|
{ path: "marker.bin", kind: "reference", content: Buffer.from([0, revision, 255]).toString("base64"), encoding: "base64" },
|
|
] }).returning();
|
|
const [version] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skillId, revisionNumber: revision,
|
|
repositorySnapshotId: snapshot!.id, repositorySkillPath: "skills/example/SKILL.md" }).returning();
|
|
versions.push(version!);
|
|
}
|
|
await db.update(companySkills).set({ currentVersionId: versions[1]!.id }).where(eq(companySkills.id, skillId));
|
|
const service = companySkillService(db);
|
|
async function resolve(versionId: string) {
|
|
const entries = await service.listRuntimeSkillEntries(companyId, { onlyKeys: new Set([key]), versionSelections: new Map([[key, versionId]]) });
|
|
expect(entries).toHaveLength(1);
|
|
expect(entries[0]!.sourceStatus, entries[0]!.missingDetail ?? undefined).toBe("available");
|
|
return entries[0]!;
|
|
}
|
|
return { companyId, skillId, key, versions, resolve };
|
|
}
|
|
|
|
it("rebuilds an old repository-backed version from database bytes with upstream unavailable", async () => {
|
|
const upstream = vi.fn(() => Promise.reject(new Error("Upstream is unavailable")));
|
|
vi.stubGlobal("fetch", upstream);
|
|
const { versions, resolve } = await fixture();
|
|
const old = await resolve(versions[0]!.id);
|
|
const current = await resolve(versions[1]!.id);
|
|
await fs.rm(old.source, { recursive: true });
|
|
const restored = await resolve(versions[0]!.id);
|
|
expect(restored.source).toBe(old.source);
|
|
const repository = path.join(restored.source, ".paperclip-repository");
|
|
expect(await fs.readFile(path.join(repository, "marker.bin"))).toEqual(Buffer.from([0, 1, 255]));
|
|
expect(await fs.readFile(path.join(repository, "scripts/example.sh"), "utf8")).toBe("#!/bin/sh\necho revision-1\n");
|
|
expect((await fs.stat(path.join(repository, "scripts/example.sh"))).mode & 0o111).toBe(0o111);
|
|
expect(await fs.readFile(path.join(current.source, ".paperclip-repository/marker.bin"))).toEqual(Buffer.from([0, 2, 255]));
|
|
expect(upstream).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("expires only idle old versions, reports dry-run candidates, and restores the evicted version", async () => {
|
|
const f = await fixture();
|
|
const old = await f.resolve(f.versions[0]!.id);
|
|
const current = await f.resolve(f.versions[1]!.id);
|
|
const aged = Date.now() - 2 * 60 * 60 * 1000;
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, aged);
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[1]!.id, aged);
|
|
const dry = await sweepSkillVersionCache(db, { companyId: f.companyId, dryRun: true });
|
|
expect(dry).toMatchObject({ candidates: 1, evicted: 0, reclaimedBytes: 0 });
|
|
expect(await fs.stat(old.source)).toBeTruthy();
|
|
const swept = await sweepSkillVersionCache(db, { companyId: f.companyId });
|
|
expect(swept).toMatchObject({ candidates: 1, evicted: 1 });
|
|
expect(swept.reclaimedBytes).toBeGreaterThan(0);
|
|
await expect(fs.stat(old.source)).rejects.toMatchObject({ code: "ENOENT" });
|
|
expect(await fs.stat(current.source)).toBeTruthy();
|
|
vi.stubGlobal("fetch", vi.fn(() => Promise.reject(new Error("Upstream unavailable"))));
|
|
expect(await f.resolve(f.versions[0]!.id)).toMatchObject({ source: old.source });
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(0);
|
|
});
|
|
|
|
it("gives existing unobserved caches a full grace period", async () => {
|
|
const f = await fixture();
|
|
await f.resolve(f.versions[0]!.id);
|
|
const paths = skillVersionCachePaths(f.companyId, f.skillId, f.versions[0]!.id);
|
|
await fs.unlink(paths.access);
|
|
const now = Date.now();
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId, now })).skipped.observation_grace).toBe(1);
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId, now: now + 59 * 60 * 1000 })).evicted).toBe(0);
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId, now: now + 61 * 60 * 1000 })).evicted).toBe(1);
|
|
});
|
|
|
|
async function activeRun(companyId: string, contextSnapshot: Record<string, unknown> | null) {
|
|
const [agent] = await db.insert(agents).values({ companyId, name: "Cache reader", role: "engineer", adapterType: "process" }).returning();
|
|
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running", contextSnapshot }).returning();
|
|
return run!;
|
|
}
|
|
|
|
it("retains pinned versions and skips companies with unrecorded active selections", async () => {
|
|
const f = await fixture();
|
|
await f.resolve(f.versions[0]!.id);
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
|
|
const run = await activeRun(f.companyId, { skillVersions: { [f.key]: f.versions[0]!.id } });
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.protected_version).toBe(1);
|
|
await db.update(heartbeatRuns).set({ contextSnapshot: null }).where(eq(heartbeatRuns.id, run.id));
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1);
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1);
|
|
});
|
|
|
|
it("serializes eviction behind preparation and durable pinning across database clients", async () => {
|
|
const f = await fixture();
|
|
await f.resolve(f.versions[0]!.id);
|
|
const run = await activeRun(f.companyId, null);
|
|
const ready = Promise.withResolvers<void>();
|
|
const release = Promise.withResolvers<void>();
|
|
const preparing = companySkillService(db).withRuntimeSkillEntries(f.companyId, {
|
|
onlyKeys: new Set([f.key]), versionSelections: new Map([[f.key, f.versions[0]!.id]]),
|
|
}, async entries => {
|
|
expect(entries).toMatchObject([{ key: f.key, versionId: f.versions[0]!.id, sourceStatus: "available" }]);
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
|
|
ready.resolve();
|
|
await release.promise;
|
|
await db.update(heartbeatRuns).set({ contextSnapshot: { skillVersions: { [f.key]: f.versions[0]!.id } } })
|
|
.where(eq(heartbeatRuns.id, run.id));
|
|
});
|
|
await ready.promise;
|
|
const other = createDb(database.connectionString);
|
|
try {
|
|
const cleaning = await sweepSkillVersionCache(other, { companyId: f.companyId });
|
|
expect(cleaning.evicted).toBe(0);
|
|
expect(cleaning.skipped.cache_busy).toBeGreaterThan(0);
|
|
} finally { release.resolve(); }
|
|
await preparing;
|
|
expect(await sweepSkillVersionCache(other, { companyId: f.companyId })).toMatchObject({ evicted: 0, skipped: { protected_version: 1 } });
|
|
await other.$client.end({ timeout: 0 });
|
|
});
|
|
|
|
it("releases the cache lock after a runtime selection consumer fails", async () => {
|
|
const f = await fixture();
|
|
await expect(companySkillService(db).withRuntimeSkillEntries(f.companyId, {
|
|
onlyKeys: new Set([f.key]), versionSelections: new Map([[f.key, f.versions[0]!.id]]),
|
|
}, async entries => {
|
|
expect(entries[0]!.sourceStatus).toBe("available");
|
|
throw new Error("durable pin write failed");
|
|
})).rejects.toThrow("durable pin write failed");
|
|
const other = createDb(database.connectionString);
|
|
try {
|
|
expect(await withSkillVersionCacheLock(other, f.companyId, async () => "released", true)).toBe("released");
|
|
} finally { await other.$client.end({ timeout: 0 }); }
|
|
});
|
|
|
|
it("reads company-wide protection twice per batch regardless of version count", async () => {
|
|
const f = await fixture();
|
|
const versionIds = [f.versions[0]!.id];
|
|
for (let revision = 3; revision < 8; revision++) {
|
|
const [version] = await db.insert(companySkillVersions).values({ companyId: f.companyId,
|
|
companySkillId: f.skillId, revisionNumber: revision, repositorySnapshotId: f.versions[0]!.repositorySnapshotId,
|
|
repositorySkillPath: "skills/example/SKILL.md" }).returning();
|
|
versionIds.push(version!.id);
|
|
}
|
|
for (const id of [...versionIds, f.versions[1]!.id]) {
|
|
await f.resolve(id);
|
|
await touchSkillVersionCache(f.companyId, f.skillId, id, Date.now() - 7_200_000);
|
|
}
|
|
const statements: string[] = [];
|
|
const debug = db.$client.options.debug;
|
|
db.$client.options.debug = (_connection, query) => { statements.push(query); };
|
|
try {
|
|
expect(await sweepSkillVersionCache(db, { companyId: f.companyId })).toMatchObject({ evicted: 6 });
|
|
expect(statements.filter(query => /from "agents"/.test(query))).toHaveLength(2);
|
|
expect(statements.filter(query => /from "heartbeat_runs"/.test(query))).toHaveLength(2);
|
|
} finally { db.$client.options.debug = debug; }
|
|
});
|
|
|
|
it("lets preparation finish during slow byte inspection and rechecks its new pin", async () => {
|
|
const f = await fixture();
|
|
const old = await f.resolve(f.versions[0]!.id);
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
|
|
const inspecting = Promise.withResolvers<void>();
|
|
const release = Promise.withResolvers<void>();
|
|
const lstat = fs.lstat.bind(fs);
|
|
let held = false;
|
|
vi.spyOn(fs, "lstat").mockImplementation((async (...args: Parameters<typeof fs.lstat>) => {
|
|
if (!held && String(args[0]) === path.join(old.source, ".paperclip-repository", "marker.bin")) {
|
|
held = true;
|
|
inspecting.resolve();
|
|
await release.promise;
|
|
}
|
|
return lstat(...args);
|
|
}) as typeof fs.lstat);
|
|
const sweep = sweepSkillVersionCache(db, { companyId: f.companyId });
|
|
await inspecting.promise;
|
|
const other = createDb(database.connectionString);
|
|
let timeout: ReturnType<typeof setTimeout> | undefined;
|
|
const preparation = withSkillVersionCacheLock(other, f.companyId, async () => {
|
|
await companySkillService(other).listRuntimeSkillEntries(f.companyId, {
|
|
onlyKeys: new Set([f.key]), versionSelections: new Map([[f.key, f.versions[0]!.id]]),
|
|
});
|
|
// Remove grace as an alternative explanation: the newly persisted run pin
|
|
// must prevent eviction after the original candidate snapshot was read.
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
|
|
await activeRun(f.companyId, { skillVersions: { [f.key]: f.versions[0]!.id } });
|
|
return true;
|
|
});
|
|
try {
|
|
expect(await Promise.race([preparation, new Promise<boolean>(resolve => {
|
|
timeout = setTimeout(() => resolve(false), 10_000);
|
|
})])).toBe(true);
|
|
} finally { clearTimeout(timeout); release.resolve(); }
|
|
await preparation;
|
|
expect(await sweep).toMatchObject({ evicted: 0, skipped: { protected_version: 1 } });
|
|
expect(await fs.stat(old.source)).toBeTruthy();
|
|
await other.$client.end({ timeout: 1 });
|
|
});
|
|
|
|
it("keeps terminal run pins until native finalization has settled", async () => {
|
|
const f = await fixture();
|
|
await f.resolve(f.versions[0]!.id);
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
|
|
const run = await activeRun(f.companyId, { skillVersions: { [f.key]: f.versions[0]!.id } });
|
|
const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Pending recovery" }).returning();
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue!.id, status: "failed", finishedAt: new Date() })
|
|
.where(eq(heartbeatRuns.id, run.id));
|
|
await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: issue!.id, runId: run.id, phase: "terminal_failure" });
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.protected_version).toBe(1);
|
|
await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, run.id));
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1);
|
|
});
|
|
|
|
it("finishes interrupted trash removal without deleting a rehydrated copy", async () => {
|
|
const f = await fixture();
|
|
const old = await f.resolve(f.versions[0]!.id);
|
|
const paths = skillVersionCachePaths(f.companyId, f.skillId, f.versions[0]!.id);
|
|
const trash = path.join(path.dirname(paths.directory), `.trash-${randomUUID()}`);
|
|
await fs.rename(old.source, trash);
|
|
await f.resolve(f.versions[0]!.id);
|
|
const result = await sweepSkillVersionCache(db, { companyId: f.companyId });
|
|
expect(result.evicted).toBe(0);
|
|
expect(result.reclaimedBytes).toBeGreaterThan(0);
|
|
await expect(fs.stat(trash)).rejects.toMatchObject({ code: "ENOENT" });
|
|
expect(await fs.stat(old.source)).toBeTruthy();
|
|
});
|
|
|
|
it("does not traverse symlinked cache roots", async () => {
|
|
const f = await fixture();
|
|
const old = await f.resolve(f.versions[0]!.id);
|
|
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
|
|
const outside = await fs.mkdtemp(path.join(home, "outside-"));
|
|
await fs.writeFile(path.join(outside, "unique.txt"), "keep");
|
|
await fs.rm(old.source, { recursive: true });
|
|
await fs.symlink(outside, old.source);
|
|
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unsafe_or_unavailable).toBe(1);
|
|
expect(await fs.readFile(path.join(outside, "unique.txt"), "utf8")).toBe("keep");
|
|
});
|
|
});
|