From d7f02c51b48500aa48bdf683dd153949ba3afa62 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:14:59 -0600 Subject: [PATCH] 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 --- doc/recoverable-disk-cleanup.md | 10 +++ .../src/__tests__/skill-version-cache.test.ts | 79 ++++++++++++++++++- server/src/services/company-skills.ts | 2 +- server/src/services/skill-version-cache.ts | 22 ++++-- 4 files changed, 106 insertions(+), 7 deletions(-) diff --git a/doc/recoverable-disk-cleanup.md b/doc/recoverable-disk-cleanup.md index 0a26bfc799a..c65a6e5f831 100644 --- a/doc/recoverable-disk-cleanup.md +++ b/doc/recoverable-disk-cleanup.md @@ -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 diff --git a/server/src/__tests__/skill-version-cache.test.ts b/server/src/__tests__/skill-version-cache.test.ts index a2e8f9b729c..5c31f420f8f 100644 --- a/server/src/__tests__/skill-version-cache.test.ts +++ b/server/src/__tests__/skill-version-cache.test.ts @@ -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(); const release = Promise.withResolvers(); const preparing = companySkillService(db).withRuntimeSkillEntries(f.companyId, { diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index c23aae8ae34..ea18c85e475 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -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))) { diff --git a/server/src/services/skill-version-cache.ts b/server/src/services/skill-version-cache.ts index f2f08ff774f..9494ff29f62 100644 --- a/server/src/services/skill-version-cache.ts +++ b/server/src/services/skill-version-cache.ts @@ -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 { } 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 | null; leaseDeadlines: Map }> { +export async function protectedVersions(db: Db, companyId: string, targetSkill?: { createdAt: Date }): Promise<{ versions: Set | null; leaseDeadlines: Map }> { const rows = await db.select({ id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, nativeIssueId: heartbeatRuns.nativeIssueId, + status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt, leaseDeadline: sql`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(); const leaseDeadlines = new Map(); 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); }