mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 08:22:20 +02:00
fix(skills): unblock retention for legacy run selections (#16852)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agent runs can select saved skill versions. > - The server keeps those files in a cache that must protect active and unresolved runs. > - Older runs can lack the newer pin map even when their saved execution input records the full selection. > - Finished interrupted runs can also keep a company blocked after all execution and cleanup have settled. > - This pull request uses existing selection evidence and recognizes settled interruptions so safe cache expiration can resume. ## Linked Issues or Issue Description Related: #16433. **What happened?** Skill cache retention reports `unknown_run_pins` for a whole company when it finds a finished interrupted run without pins, or an older native run whose complete selection is stored only in its execution input. Old cache directories cannot expire. **Expected behavior** Release pins from fully settled interruptions. Preserve exact versions from valid saved native selections. Continue to block cleanup when selection evidence is missing, invalid, or belongs to another run. **Steps to reproduce** 1. Materialize an old, restorable skill version and let its idle grace period expire. 2. Add a finished interrupted run with settled ownership and cleanup, or an unresolved native run with a valid saved execution input but no modern pin map. 3. Run the retention service. Before this change, it skips the company with `unknown_run_pins`. ## What Changed - Treat `interrupted` as terminal only when the existing finish, controller, finalization, and environment cleanup checks allow it. - For native runs with no modern pin-map field, parse the saved execution input with the existing recovery parser. Validate its digests and company, run, agent, and issue binding. - Keep captured version IDs protected. Unversioned skills use separate captured asset bundles. Do not replace a malformed modern map or write any recovery records. - Add database regressions and document the fallback. Remove an unused status list that no longer represented the actual eligibility query. ## Verification - Passed 41 focused tests: `skill-version-cache.test.ts`, `skill-version-retention.test.ts`, and `skill-version-retention-config.test.ts`. - The tests perform real PostgreSQL reads, directory eviction, and restoration from saved bytes with upstream unavailable. They cover leases, unfinished cleanup, invalid digests, invalid versions, and mismatched identities. - Passed `node scripts/check-module-boundaries.mjs` and `git diff --check`. - A read-only evaluation of the candidate selector against existing legacy data cleared the unknown-pin blocker while retaining exact version pins. - Passed `pnpm -r --workspace-concurrency=2 typecheck` and `pnpm build`. The full remote CI suite passed, including all server, workspace, Runner, and end-to-end shards. The full local test command remains in progress; its focused cache suites passed as recorded above. - Greptile reviewed this head at 5/5 with no actionable comments or unresolved threads. - Initial local database attempts did not run the cases: sandbox access prevented a listener, then an install with scripts disabled lacked PostgreSQL library symlinks. After running the package's symlink setup, the binary started and all focused tests passed. ## Risks The main risk is releasing a cache still in use. The existing ownership and cleanup checks remain in force. The native fallback applies only when the modern field is absent and the saved input passes the same parser used for recovery. Invalid or missing evidence still blocks the company. Current versions and explicit agent pins stay protected. Saved version rows and runtime asset bundles are not deleted. There is no migration. The selector reads execution inputs only for legacy native runs that need the fallback. Normal runs still load only their pin maps and binding columns. A representative read-only selector check completed in 98 ms; this is a sample, not a latency guarantee. ## Model Used OpenAI Codex, GPT-6 family, with tool-assisted coding and verification. ## 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 and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
9966f31bee
commit
06c5bc471d
4 files changed
+124
-7
No files matched your search
@@ -2196,6 +2196,14 @@ with an active or unsettled run whose pins are unknown is skipped. Optional
|
||||
(default `60`). Separate `__version_access__` records track use; existing caches
|
||||
without records receive a full observation period before expiration.
|
||||
|
||||
Finished interrupted runs release their pins only after controller ownership,
|
||||
native finalization, and environment cleanup have settled. Older native runs
|
||||
without a pin map can supply their complete selection from the saved execution
|
||||
input. The recovery parser validates its digests, and its company, run, agent,
|
||||
and issue binding must match. Captured version IDs remain protected; unversioned
|
||||
skills use separate runtime asset bundles. Missing or invalid evidence still
|
||||
blocks cleanup. This fallback does not write pins or change recovery records.
|
||||
|
||||
Preparation, durable run pins, audits, and eviction share a company-scoped
|
||||
PostgreSQL lock across processes. Eligible directories are atomically renamed
|
||||
to unique managed `.trash-<UUID>` names. Deletion runs outside the preparation
|
||||
|
||||
@@ -4,12 +4,14 @@ 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 { agents, heartbeatRuns, 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";
|
||||
|
||||
import { skillVersionRetentionService } from "../services/skill-version-retention.js";
|
||||
import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js";
|
||||
import { canonicalNativeRuntimeContextDigest, parseNativeExecutionInput } from "../vendor/paperclip-runner/index.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 };
|
||||
@@ -136,6 +138,88 @@ describePostgres("immutable skill version cache", () => {
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1);
|
||||
});
|
||||
|
||||
it("releases settled interrupted runs but preserves unfinished execution and cleanup", 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, null);
|
||||
await db.update(heartbeatRuns).set({ status: "interrupted" }).where(eq(heartbeatRuns.id, run.id));
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1);
|
||||
await db.update(heartbeatRuns).set({ finishedAt: new Date(), controllerLeaseExpiresAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1);
|
||||
await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(0) }).where(eq(heartbeatRuns.id, run.id));
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id,
|
||||
status: "pending_cleanup", cleanupStatus: "failed" }).returning();
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1);
|
||||
await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() })
|
||||
.where(eq(environmentLeases.id, lease!.id));
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1);
|
||||
});
|
||||
|
||||
async function legacyNativeRun(companyId: string, skills: Array<{ key: string; versionId: string | null }> = []) {
|
||||
const run = await activeRun(companyId, null);
|
||||
const [issue] = await db.insert(issues).values({ companyId, title: "Saved native selection" }).returning();
|
||||
const runtimeContext = nativeRuntimeContextFixture();
|
||||
runtimeContext.skills = skills.map((skill, index) => ({ ...skill, runtimeName: `skill-${index}`, bundle: runtimeContext.instructions.bundle }));
|
||||
runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext);
|
||||
const execution = parseNativeExecutionInput({
|
||||
schema: "paperclip.native-execution-input.v3",
|
||||
binding: { companyId, runId: run.id, agentId: run.agentId, issueId: issue!.id, executionWorkspaceId: run.id },
|
||||
task: { identifier: "CACHE-1", title: "Saved selection", description: null, prompt: "Read saved skills", workMode: "standard" },
|
||||
workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: null, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
provider: { kind: "codex", model: null },
|
||||
completionContract: { id: "contract", sha256: "saved", schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: { revision: "1", objective: "Read saved skills", criteria: [{ id: "read", requirement: "Read skills" }] } },
|
||||
interactionResponses: [], credentialBindings: [], executionMode: "default", planningContext: null, runtimeContext,
|
||||
});
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue!.id, status: "failed", finishedAt: new Date(),
|
||||
runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, run.id));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue!.id, runId: run.id,
|
||||
phase: "terminal_failure", recoveryState: "blocked" });
|
||||
return { run, execution };
|
||||
}
|
||||
|
||||
it.each([false, true])("uses the legacy native manifest without changing unresolved recovery (bundled skill: %s)", async bundled => {
|
||||
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 { run } = await legacyNativeRun(f.companyId, bundled ? [{ key: "paperclipai/paperclip/slack", versionId: null }] : []);
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1);
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0])
|
||||
.toMatchObject({ phase: "terminal_failure", recoveryState: "blocked" });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]!.contextSnapshot).toBeNull();
|
||||
vi.stubGlobal("fetch", vi.fn(() => Promise.reject(new Error("Upstream unavailable"))));
|
||||
expect(await f.resolve(f.versions[0]!.id)).toMatchObject({ source: old.source });
|
||||
});
|
||||
|
||||
it("preserves exact legacy manifest pins", 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);
|
||||
await legacyNativeRun(f.companyId, [{ key: f.key, versionId: f.versions[0]!.id }]);
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.protected_version).toBe(1);
|
||||
});
|
||||
|
||||
it.each(["companyId", "runId", "agentId", "issueId", "digest", "missing", "invalid_version", "invalid_pins", "legacy_adapter"])(
|
||||
"fails closed for unusable native selection evidence (%s)", async invalid => {
|
||||
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, execution } = await legacyNativeRun(f.companyId,
|
||||
invalid === "invalid_version" ? [{ key: f.key, versionId: "not-a-version" }] : []);
|
||||
if (["companyId", "runId", "agentId", "issueId"].includes(invalid)) Object.assign(execution.binding, { [invalid]: randomUUID() });
|
||||
if (invalid === "digest" && "runtimeContext" in execution) execution.runtimeContext.aggregateDigest = "f".repeat(64);
|
||||
await db.update(heartbeatRuns).set({
|
||||
runnerProfileJson: invalid === "missing" ? {} : { nativeExecutionInput: execution },
|
||||
...(invalid === "invalid_pins" ? { contextSnapshot: { skillVersions: null } } : {}),
|
||||
...(invalid === "legacy_adapter" ? { runtimeMode: "legacy" } : {}),
|
||||
}).where(eq(heartbeatRuns.id, run.id));
|
||||
expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).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);
|
||||
|
||||
@@ -9,6 +9,7 @@ import { resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
import { assertSkillSnapshotPath } from "./skill-snapshot.js";
|
||||
import { repositorySkillFrontmatter } from "./skill-repository-package.js";
|
||||
import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared";
|
||||
import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js";
|
||||
|
||||
export const SKILL_VERSION_CACHE_IDLE_MS = 60 * 60 * 1000;
|
||||
const locks = new AsyncLocalStorage<Array<{ db: Db; companyId: string }>>();
|
||||
@@ -91,9 +92,16 @@ export async function lastUsedAt(filename: string): Promise<number | null> {
|
||||
// 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<Set<string> | null> {
|
||||
const rows = await db.select({ pins: sql<unknown>`${heartbeatRuns.contextSnapshot}->'skillVersions'` })
|
||||
const rows = await db.select({
|
||||
id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
pins: sql<unknown>`${heartbeatRuns.contextSnapshot}->'skillVersions'`,
|
||||
// Avoid loading large execution inputs for runs with the modern pin map.
|
||||
legacyInput: sql<unknown>`case when ${heartbeatRuns.runtimeMode} = 'native'
|
||||
and not coalesce(${heartbeatRuns.contextSnapshot} ? 'skillVersions', false)
|
||||
then ${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput' end`,
|
||||
})
|
||||
.from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), sql`(
|
||||
${heartbeatRuns.status} not in ('succeeded', 'failed', 'timed_out', 'cancelled')
|
||||
${heartbeatRuns.status} not in ('succeeded', 'failed', 'timed_out', 'cancelled', 'interrupted')
|
||||
or ${heartbeatRuns.finishedAt} is null
|
||||
or ${heartbeatRuns.controllerLeaseExpiresAt} > now()
|
||||
or exists (select 1 from ${nativeRunFinalizations} f
|
||||
@@ -107,7 +115,27 @@ export async function protectedVersions(db: Db, companyId: string): Promise<Set<
|
||||
and l.lease_policy = 'reuse_by_environment' and l.cleanup_status = 'success', false))))
|
||||
)`));
|
||||
const versions = new Set<string>();
|
||||
for (const { pins } of rows) {
|
||||
for (const { pins, legacyInput, id, agentId, nativeIssueId } of rows) {
|
||||
if (legacyInput != null) {
|
||||
// Older native runs captured the complete selection in their immutable
|
||||
// execution input before contextSnapshot.skillVersions was introduced.
|
||||
// Use the recovery parser (including digest validation) and the exact run
|
||||
// binding. Never repair a malformed modern pin map with a different source.
|
||||
try {
|
||||
const input = parseNativeExecutionInput(legacyInput);
|
||||
if (!("runtimeContext" in input) || input.binding.companyId !== companyId
|
||||
|| input.binding.runId !== id || input.binding.agentId !== agentId
|
||||
|| input.binding.issueId !== nativeIssueId) return null;
|
||||
for (const skill of input.runtimeContext.skills) {
|
||||
// Unversioned skills have their own captured asset bundle; they do
|
||||
// not reference a directory under __versions__.
|
||||
if (skill.versionId === null) continue;
|
||||
if (!uuid.test(skill.versionId)) return null;
|
||||
versions.add(skill.versionId);
|
||||
}
|
||||
continue;
|
||||
} catch { return null; }
|
||||
}
|
||||
// Legacy/incomplete preparation is not proof that no skill is in use.
|
||||
if (!pins || typeof pins !== "object" || Array.isArray(pins)) return null;
|
||||
for (const version of Object.values(pins)) {
|
||||
|
||||
@@ -33,9 +33,6 @@ export const SKILL_VERSION_RETENTION_DEFAULT_MAX_REMOVALS = 50;
|
||||
export const SKILL_VERSION_RETENTION_INTERVAL_MS = 60 * 60 * 1000;
|
||||
export const SKILL_VERSION_RETENTION_STARTUP_DELAY_MS = 60 * 1000;
|
||||
|
||||
/** A run in one of these states can still read (or re-read) its captured skill versions. */
|
||||
export const SKILL_VERSION_LIVE_RUN_STATUSES = ["queued", "scheduled_retry", "running", "interrupted"] as const;
|
||||
|
||||
const VERSIONS_DIR = "__versions__";
|
||||
const TRASH_PREFIX = ".trash-";
|
||||
const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
|
||||
|
||||
Reference in new issue
Block a user