mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(skills): scope file protection to admitted selections (#19904)
Allow unused skill deletion when unstarted admission entries have no retained execution. Limit unknown legacy selections to skills that could have existed when the validated immutable run finished. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
c9959814ba
commit
d7f02c51b4
4 files changed
+106
-7
No files matched your search
@@ -20,6 +20,16 @@ unsettled recovery pins are retained. If an active run's version selection is
|
||||
unknown, that company's caches remain. Non-versioned skill directories and
|
||||
database snapshots are never removed by this worker.
|
||||
|
||||
Queued turns and scheduled retries without any preparation, execution, or saved
|
||||
skill selection do not pin files. They select current skills when preparation
|
||||
starts. Recorded selections and uncertain execution still protect their versions,
|
||||
including requeued runs and unfinished recovery. The same distinction applies to
|
||||
deleting an unused skill.
|
||||
For deletion only, a valid immutable legacy selection with an unknown version
|
||||
does not protect a skill created after that run finished. Uncertain or malformed
|
||||
selections still block deletion. Global cache eviction keeps its conservative
|
||||
protection for these unknown versions.
|
||||
|
||||
Last-use records live separately under `__version_access__`. Older installations
|
||||
without those records receive a full one-hour observation period. Preparation,
|
||||
pin persistence and eviction share a company-scoped database lock across server
|
||||
|
||||
@@ -6,7 +6,7 @@ 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, environmentLeases, companies, companySkills, companySkillVersions, companySkillRepositorySnapshots, createDb } from "@paperclipai/db";
|
||||
import { agents, heartbeatRuns, heartbeatRunEvents, issues, nativeRunFinalizations, environmentLeases, 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";
|
||||
@@ -187,6 +187,53 @@ describePostgres("immutable skill version cache", () => {
|
||||
return run!;
|
||||
}
|
||||
|
||||
it.each(["queued", "scheduled_retry"])("allows cache eviction and unused skill deletion before %s admission", async status => {
|
||||
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, null);
|
||||
await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, run.id));
|
||||
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1);
|
||||
await expect(companySkillService(db).deleteSkill(f.companyId, f.skillId)).resolves.toMatchObject({ id: f.skillId });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toMatchObject({
|
||||
status, startedAt: null, contextSnapshot: null,
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ startedAt: new Date() },
|
||||
{ runtimeModeResolvedAt: new Date() },
|
||||
{ runnerInstanceId: randomUUID() },
|
||||
{ nativePhase: "preparing" },
|
||||
{ executionStage: "preparing" },
|
||||
{ runnerProfileJson: { nativeExecutionInput: {} } },
|
||||
{ contextSnapshot: { skillVersions: null } },
|
||||
{ contextSnapshot: { skillSelectionSnapshot: true } },
|
||||
])("preserves unknown skill selections for queued execution evidence %j", async evidence => {
|
||||
const f = await fixture();
|
||||
const run = await activeRun(f.companyId, null);
|
||||
await db.update(heartbeatRuns).set({ status: "queued", ...evidence }).where(eq(heartbeatRuns.id, run.id));
|
||||
await expect(companySkillService(db).deleteSkill(f.companyId, f.skillId)).rejects.toThrow("active or recovering run");
|
||||
expect(await companySkillService(db).getById(f.companyId, f.skillId)).not.toBeNull();
|
||||
});
|
||||
|
||||
it.each(["queued", "scheduled_retry"])("retains recorded skill selections before %s admission", async status => {
|
||||
const f = await fixture();
|
||||
const run = await activeRun(f.companyId, { skillVersions: { [f.key]: f.versions[0]!.id } });
|
||||
await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, run.id));
|
||||
await expect(companySkillService(db).deleteSkill(f.companyId, f.skillId)).rejects.toThrow("active or recovering run");
|
||||
});
|
||||
|
||||
it("preserves queued execution with a retained provider event", async () => {
|
||||
const f = await fixture();
|
||||
const run = await activeRun(f.companyId, null);
|
||||
await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, run.id));
|
||||
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: run.id, agentId: run.agentId,
|
||||
seq: 1, eventType: "session.started" });
|
||||
await expect(companySkillService(db).deleteSkill(f.companyId, f.skillId)).rejects.toThrow("active or recovering run");
|
||||
});
|
||||
|
||||
it("retains pinned versions and skips companies with unrecorded active selections", async () => {
|
||||
const f = await fixture();
|
||||
await f.resolve(f.versions[0]!.id);
|
||||
@@ -257,6 +304,35 @@ describePostgres("immutable skill version cache", () => {
|
||||
expect(await f.resolve(f.versions[0]!.id)).toMatchObject({ source: old.source });
|
||||
});
|
||||
|
||||
it("deletes a later skill while keeping unknown legacy recovery versions protected", async () => {
|
||||
const f = await fixture();
|
||||
const { run } = await legacyNativeRun(f.companyId, [{ key: "paperclipai/paperclip/slack", versionId: null }]);
|
||||
const [skill] = await db.select().from(companySkills).where(eq(companySkills.id, f.skillId));
|
||||
await db.update(heartbeatRuns).set({ finishedAt: new Date(skill!.createdAt.getTime() - 1_000) })
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
await f.resolve(f.versions[0]!.id);
|
||||
await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000);
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1);
|
||||
await expect(companySkillService(db).deleteSkill(f.companyId, f.skillId)).resolves.toMatchObject({ id: f.skillId });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toMatchObject({ status: "failed" });
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0])
|
||||
.toMatchObject({ phase: "terminal_failure", recoveryState: "blocked" });
|
||||
});
|
||||
|
||||
it.each(["renamed", "running", "malformed"])("keeps uncertain legacy selections for %s execution", async mode => {
|
||||
const f = await fixture();
|
||||
const { run, execution } = await legacyNativeRun(f.companyId, [{ key: f.key, versionId: null }]);
|
||||
const [skill] = await db.select().from(companySkills).where(eq(companySkills.id, f.skillId));
|
||||
await db.update(heartbeatRuns).set({
|
||||
finishedAt: new Date(skill!.createdAt.getTime() + (mode === "renamed" ? 1_000 : -1_000)),
|
||||
...(mode === "running" ? { status: "running" } : {}),
|
||||
...(mode === "malformed" ? { runnerProfileJson: { nativeExecutionInput: { ...execution, binding: { ...execution.binding, runId: randomUUID() } } } } : {}),
|
||||
}).where(eq(heartbeatRuns.id, run.id));
|
||||
if (mode === "renamed") await db.update(companySkills).set({ key: `company/${f.companyId}/renamed-skill` })
|
||||
.where(eq(companySkills.id, f.skillId));
|
||||
await expect(companySkillService(db).deleteSkill(f.companyId, f.skillId)).rejects.toThrow("active or recovering run");
|
||||
});
|
||||
|
||||
it("preserves exact legacy manifest pins", async () => {
|
||||
const f = await fixture();
|
||||
await f.resolve(f.versions[0]!.id);
|
||||
@@ -287,6 +363,7 @@ describePostgres("immutable skill version cache", () => {
|
||||
const f = await fixture();
|
||||
await f.resolve(f.versions[0]!.id);
|
||||
const run = await activeRun(f.companyId, null);
|
||||
await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, run.id));
|
||||
const ready = Promise.withResolvers<void>();
|
||||
const release = Promise.withResolvers<void>();
|
||||
const preparing = companySkillService(db).withRuntimeSkillEntries(f.companyId, {
|
||||
|
||||
@@ -7403,7 +7403,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
.filter((skill): skill is CompanySkill => Boolean(skill));
|
||||
async function assertUnused(skill: CompanySkill, database: DbOrTransaction = db) {
|
||||
const usedByAgents = await usage(companyId, skill.key, database);
|
||||
const pins = await protectedVersions(database as Db, companyId);
|
||||
const pins = await protectedVersions(database as Db, companyId, { createdAt: skill.createdAt });
|
||||
const versions = await database.select({ id: companySkillVersions.id }).from(companySkillVersions).where(and(
|
||||
eq(companySkillVersions.companyId, companyId), eq(companySkillVersions.companySkillId, skill.id)));
|
||||
if (pins.versions === null || versions.some(version => pins.versions!.has(version.id))) {
|
||||
|
||||
@@ -3,7 +3,7 @@ import { AsyncLocalStorage } from "node:async_hooks";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { constants, promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { and, eq, inArray, isNull, sql } from "drizzle-orm";
|
||||
import { companySkillVersions, heartbeatRuns,
|
||||
nativeRunFinalizations, environmentLeases, withDedicatedDbConnection, type Db } from "@paperclipai/db";
|
||||
import { resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
@@ -13,6 +13,7 @@ import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared"
|
||||
import { noteSkillVersionCacheChange } from "./skill-version-retention-notifications.js";
|
||||
import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js";
|
||||
import { beginRunPhase } from "./execution-preparation-timing.js";
|
||||
import { noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js";
|
||||
import { withWorkspaceLockKeepalive } from "./native-runtime/native-workspace-lock-keepalive.js";
|
||||
|
||||
export const SKILL_VERSION_CACHE_IDLE_MS = 60 * 60 * 1000;
|
||||
@@ -101,7 +102,13 @@ export async function lastUsedAt(filename: string): Promise<number | null> {
|
||||
}
|
||||
|
||||
export function skillRunNeedsFilesSql() {
|
||||
return sql`(
|
||||
const unstartedAdmission = and(
|
||||
inArray(heartbeatRuns.status, ["queued", "scheduled_retry"]), noRetainedExecutionPredicate(),
|
||||
isNull(heartbeatRuns.finishedAt), isNull(heartbeatRuns.runnerInstanceId),
|
||||
isNull(heartbeatRuns.nativePhase), isNull(heartbeatRuns.nativePhaseUpdatedAt),
|
||||
sql`not (coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) ?| array['skillVersions', 'skillSelectionSnapshot'])`,
|
||||
);
|
||||
return sql`not (${unstartedAdmission}) and (
|
||||
${heartbeatRuns.status} not in ('succeeded', 'failed', 'timed_out', 'cancelled', 'interrupted')
|
||||
or ${heartbeatRuns.finishedAt} is null
|
||||
or ${heartbeatRuns.controllerLeaseExpiresAt} > now()
|
||||
@@ -119,9 +126,10 @@ export function skillRunNeedsFilesSql() {
|
||||
|
||||
// A terminal run can still own provider execution or recovery. Only fully settled
|
||||
// runs release their pins; resume preparation reacquires the same cache lock.
|
||||
export async function protectedVersions(db: Db, companyId: string): Promise<{ versions: Set<string> | null; leaseDeadlines: Map<string, number> }> {
|
||||
export async function protectedVersions(db: Db, companyId: string, targetSkill?: { createdAt: Date }): Promise<{ versions: Set<string> | null; leaseDeadlines: Map<string, number> }> {
|
||||
const rows = await db.select({
|
||||
id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt,
|
||||
leaseDeadline: sql<number | null>`case when ${heartbeatRuns.status} in ('succeeded', 'failed', 'timed_out', 'cancelled', 'interrupted')
|
||||
and ${heartbeatRuns.finishedAt} is not null and ${heartbeatRuns.controllerLeaseExpiresAt} > now()
|
||||
then extract(epoch from ${heartbeatRuns.controllerLeaseExpiresAt}) * 1000 end`,
|
||||
@@ -135,7 +143,7 @@ export async function protectedVersions(db: Db, companyId: string): Promise<{ ve
|
||||
const versions = new Set<string>();
|
||||
const leaseDeadlines = new Map<string, number>();
|
||||
const unknown = () => ({ versions: null, leaseDeadlines });
|
||||
for (const { pins, legacyInput, id, agentId, nativeIssueId, leaseDeadline } of rows) {
|
||||
for (const { pins, legacyInput, id, agentId, nativeIssueId, leaseDeadline, status, finishedAt } of rows) {
|
||||
const add = (version: string) => {
|
||||
versions.add(version);
|
||||
if (leaseDeadline !== null) leaseDeadlines.set(version, Math.min(leaseDeadlines.get(version) ?? Infinity, Number(leaseDeadline)));
|
||||
@@ -151,7 +159,11 @@ export async function protectedVersions(db: Db, companyId: string): Promise<{ ve
|
||||
|| input.binding.runId !== id || input.binding.agentId !== agentId
|
||||
|| input.binding.issueId !== nativeIssueId) return unknown();
|
||||
for (const skill of input.runtimeContext.skills) {
|
||||
if (skill.versionId === null) return unknown();
|
||||
if (skill.versionId === null) {
|
||||
if (targetSkill && finishedAt && targetSkill.createdAt > finishedAt
|
||||
&& ["succeeded", "failed", "timed_out", "cancelled", "interrupted"].includes(status)) continue;
|
||||
return unknown();
|
||||
}
|
||||
if (!uuid.test(skill.versionId)) return unknown();
|
||||
add(skill.versionId);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user