mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 13:24:55 +01:00
fix(skills): expire unused version caches and restore saved contents (#16433)
Expire idle skill-version caches with durable runtime pins, bounded discovery, conservative retention, and database-backed restoration. Co-Authored-By: Dotta <cryppadotta@users.noreply.github.com> Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
18 files changed
+1641
-24
No files matched your search
@@ -1989,6 +1989,41 @@ CI runs `pnpm --filter @paperclipai/skills-catalog validate` and the package's
|
||||
vitest suite, so always regenerate the manifest in the same commit as the
|
||||
catalog change.
|
||||
|
||||
## Skill Version Snapshot Retention
|
||||
|
||||
A run uses a skill version from a directory on disk:
|
||||
`<instance>/skills/<companyId>/__versions__/<skillId>/<versionId>/`. The server
|
||||
writes this directory from the stored files of the `company_skill_versions` row.
|
||||
The directory is a cache. When the server removes it and the row stays, the next
|
||||
run that selects the version writes it again.
|
||||
|
||||
The server starts scanning one minute after startup, then hourly, in batches of
|
||||
50 inspected directories with one-second continuations. Current versions,
|
||||
explicit agent pins, active runs, and unresolved recovery pins stay. A company
|
||||
with an active or unsettled run whose pins are unknown is skipped. Optional
|
||||
`PAPERCLIP_SKILL_VERSION_RETENTION_KEEP` retains additional newest versions
|
||||
(default `0`). Idle versions expire after `PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES`
|
||||
(default `60`). Separate `__version_access__` records track use; existing caches
|
||||
without records receive a full observation period before expiration.
|
||||
|
||||
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
|
||||
lock and resumes after a crash. Saved version and repository snapshot rows are
|
||||
retained, and normal runtime use restores missing caches without upstream access.
|
||||
|
||||
The sweep also handles orphan version directories after explicit skill deletion,
|
||||
preserving captured run pins until they settle. Only recognized staging/trash
|
||||
names are eligible; unknown directories and symlinked paths are retained.
|
||||
|
||||
Set `PAPERCLIP_SKILL_VERSION_RETENTION_DRY_RUN=true` before rollout to inspect
|
||||
candidate bytes and skip reasons without deleting or changing access records.
|
||||
Set `PAPERCLIP_SKILL_VERSION_RETENTION_ENABLED=false` to disable sweeps while
|
||||
leaving restoration available. Internal `skillVersionRetentionService` methods
|
||||
support company-scoped diagnosis and bounded instance-wide continuations.
|
||||
Server logs report candidates, allocated bytes reclaimed, and skip reasons.
|
||||
Compare volume usage separately because concurrent writes affect net savings.
|
||||
|
||||
## App-Shipped Teams Catalog
|
||||
|
||||
The team catalog package mirrors the skills catalog workflow for
|
||||
@@ -2110,6 +2145,13 @@ Environment overrides:
|
||||
non-numeric value falls back to the default.
|
||||
- `PAPERCLIP_WORKSPACE_COMPACTION_ENABLED=false` turns compaction off. Any other
|
||||
value keeps it on.
|
||||
- `PAPERCLIP_SKILL_VERSION_RETENTION_KEEP=<count>` sets how many of the newest
|
||||
materialized versions of each skill stay on disk (see "Skill Version Snapshot
|
||||
Retention" above). The default is `0`. A negative, empty, or non-numeric value
|
||||
falls back to the default.
|
||||
- `PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES=<minutes>` sets how long a
|
||||
materialized skill version stays after its last use. The default is `60`. A
|
||||
negative, empty, or non-numeric value falls back to the default.
|
||||
|
||||
Without `PAPERCLIP_DB_BACKUP_ALERT_FILE`, health checks look for
|
||||
`db-backup-to-s3.failure` in the backup directory, beside the backup directory,
|
||||
|
||||
@@ -130,6 +130,10 @@ Each execution records the immutable versions it uses; resuming keeps those vers
|
||||
while a new execution resolves latest again. Upstream failures and removals preserve
|
||||
installed snapshots. GitHub refresh does not delay task execution.
|
||||
|
||||
Old, unused skill-version disk caches expire automatically while current and in-use
|
||||
versions remain. Saved database snapshots rebuild an expired copy on demand, including
|
||||
when its upstream repository is unavailable. Recovery data has separate retention.
|
||||
|
||||
Disconnecting removes a repository from Sources after an explicit choice: keep
|
||||
independent editable copies in My Skills, or delete its imported skills. Kept
|
||||
copies and derived forks retain their project access restrictions. Copies preserve
|
||||
|
||||
@@ -2084,6 +2084,12 @@ unavailable. Preserve current ownership and newer-work fences. See
|
||||
and history. New key collisions cannot overwrite unrelated skills. Only changed package
|
||||
bytes or executable modes create versions. Library reads, tests, and subsequent unpinned
|
||||
runs use local installed snapshots; pinned/active runs retain their versions.
|
||||
- Runtime version directories are disposable caches. An hourly, bounded sweep may
|
||||
evict non-current copies unused for one hour, preserving active and unresolved
|
||||
execution pins. Selection, materialization, and pin persistence serialize with
|
||||
eviction across processes. Missing copies rebuild from immutable database bytes
|
||||
without upstream access. See [recoverable disk cleanup](recoverable-disk-cleanup.md)
|
||||
for rollout controls, legacy grace periods, and interrupted cleanup recovery.
|
||||
- Manually selected sources retain reviewed selection, with new candidates checked by default and
|
||||
explicit folder exclusions retained. Upstream deletion keeps installed content with
|
||||
removed status; moves are removal plus discovery. Deselect retains content
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
# Recoverable disk cleanup — delivery state
|
||||
|
||||
## Outcome and boundaries
|
||||
|
||||
Automatically expire reproducible skill caches and finished isolated Git workspaces,
|
||||
then restore them through normal use. Deliver two reviewed slices and verify actual
|
||||
reclamation plus restoration on `paperclip.staging.paperclip.app`. Keep the current
|
||||
storage architecture, unique work, and recovery data. Git restoration may depend
|
||||
on upstream availability; skill restoration must use saved database bytes alone.
|
||||
|
||||
Apply dotta-kickoff, goal-poteto-style-definition, shorten-feedback-loop,
|
||||
paperclip-feature-to-production, and test-it-for-real. Retain one deployment owner.
|
||||
|
||||
## Acceptance matrix
|
||||
|
||||
| Case | Required outcome |
|
||||
| --- | --- |
|
||||
| Old skill version, upstream unavailable | Identical restored bytes and executable modes |
|
||||
| Current/active/unsettled skill version | Preserved through concurrent preparation and cleanup |
|
||||
| Finished Git workspace/base, upstream advanced | Reopen at recorded commit and perform subsequent work |
|
||||
| Unique files, live consumers, pending recovery | Skip cleanup |
|
||||
| Restart/concurrent reopen/interrupted deletion | Recover safely and preserve lifecycle fencing |
|
||||
| Upstream failure/cross-company access | Explicit failure or skip; no fallback or access widening |
|
||||
|
||||
## Delivery slices
|
||||
|
||||
1. Skill `__versions__` cache: one-hour inactivity, current/in-use protection,
|
||||
persistent access records, cross-process exclusion, hourly batches of 50.
|
||||
2. Owned isolated Git workspace/base caches: 24-hour terminal-tree cooldown,
|
||||
verified exact-commit recipe before deletion, safe on-demand restoration.
|
||||
|
||||
Each slice gets focused tests, required repository checks, review, single-workspace
|
||||
deployment at an exact SHA, dry-run inspection, measured reclamation, and ordinary
|
||||
product acceptance. Rollback disables deletion and retains restoration support.
|
||||
|
||||
## Current state
|
||||
|
||||
- Checkout: `/Users/dotta/.codex/worktrees/2da2/paperclip`
|
||||
- Skill PR branch: `fix/skill-version-dir-retention`, based on Devin Foley's
|
||||
[PR #16433](https://github.com/paperclipai/paperclip/pull/16433). User approved
|
||||
building on that PR. Added safety commits to its existing branch, preserving
|
||||
Devin’s original commit; latest published revision before this fix is `6385a28ce7e0ee24bf0bc5e1dc088e7f4abb9508`.
|
||||
User also authorized merging every PR from this effort after checks pass.
|
||||
- Earlier proof/implementation preserved at `da754ff744` on
|
||||
`codex/recoverable-skill-cache`; do not ship its competing worker.
|
||||
- Staging: `paperclip.staging.paperclip.app`, stack `stack-pool-50531a9d17dd`.
|
||||
- Staging read-only recheck at 16:07 UTC: 71% disk used, 103,676,784,640 bytes
|
||||
used, 43,621,777,408 available. Serving `7ec6fd09725fdcd771425e31ce900f8db9aae07d`.
|
||||
This task has not deployed or deleted staging data.
|
||||
- A temporary Railway key named `paperclip-cleanup-acceptance-2026-10-15` is
|
||||
registered under the prior explicit approval. Revoke it and remove both local
|
||||
copies at acceptance completion (paths retained in private task evidence).
|
||||
- Existing unresolved staging runs have missing skill-version pins. The approved
|
||||
conservative policy skips that company; do not clear recovery or weaken the
|
||||
guard to obtain reclamation. Qualify the skill slice with a disposable pinned
|
||||
fixture and report the legacy-data gap separately.
|
||||
- Integrated one worker, cross-process preparation/pin locks, separate last-use
|
||||
metadata, legacy observation grace, unresolved-recovery protection, dry run,
|
||||
bounded scans/continuations, and deletion outside the preparation lock.
|
||||
- First integrated run: 22 focused tests passed; broader controls passed 100/101.
|
||||
Removed an unnecessary explicit-deletion lock exposed by the rollback fixture;
|
||||
that control passed on rerun (12.21s). The original failure remains in
|
||||
`/tmp/paperclip-skill-integration-controls.log`.
|
||||
- Full typecheck initially found a plugin dependency API mismatch from the
|
||||
unlocked bootstrap install. Current master supplied the refreshed lockfile;
|
||||
frozen install completed in 18.7s and full typecheck passed in 59.82s.
|
||||
- Review at `5eb505ff5d` was 3/5: unsafe skill roots blocked the company sweep,
|
||||
and recursive byte counts held the preparation lock. Both fixed locally;
|
||||
18 affected tests passed in 24.94s, including preparation during a paused file
|
||||
walk and fresh pin revalidation. Server typecheck passed after those fixes.
|
||||
- Full tests were stopped after approximately four minutes to address review
|
||||
findings before final qualification; no full-suite pass claimed. Full build at
|
||||
`d153929d70` passed. CI at that SHA passed all but the empty-runtime suite and
|
||||
Greptile (4/5). The remaining review finding was repeated company-wide pin
|
||||
queries; batching reads twice per sweep phase fixes it and has a query-count
|
||||
regression. The empty-selection fast path now returns before locking.
|
||||
- Final affected skill/heartbeat controls: 35 tests passed in 29.17s, log
|
||||
`/tmp/paperclip-skill-final-controls.log`. A prior local attempt could not start
|
||||
Postgres because macOS shared-memory slots were full; retry succeeded after a
|
||||
slot became available, without changing global limits or unrelated services.
|
||||
Full typecheck on `666c60c859` passed in 32.98s. Full tests are still running;
|
||||
initial native-session cases hit the same infrastructure limit. Removed 19
|
||||
abandoned 56-byte IPC headers only after verifying zero attachments and dead
|
||||
creator/last-user PIDs. Live databases and their files were untouched.
|
||||
- Review round four accepted batched reads but found unbounded per-directory
|
||||
discovery stats. Discovery now uses directory-entry types and applies the cursor
|
||||
and budget before per-path inspection, including empty-root budget accounting.
|
||||
Twenty affected tests passed in 22.05s, including filesystem-call counts across
|
||||
continuation pages (`/tmp/paperclip-skill-discovery-budget.log`). Final review
|
||||
and full checks remain. Review round five found an exhausted-root continuation
|
||||
edge case; the cursor now skips completed roots before budgeting discovery.
|
||||
Twenty-one affected tests passed in 27.98s, including a page-boundary regression
|
||||
(`/tmp/paperclip-skill-continuation.log`). The server CI shard stopped with
|
||||
"The operation was canceled" and no failing assertion; investigate its timeout.
|
||||
- Concurrent Git work is on `codex/recoverable-git-cache` and remains uncommitted.
|
||||
Actual runtime/service reopening now restores A after deleting the worktree,
|
||||
branch, and base while upstream is at B. A subsequent commit survives another
|
||||
reopen. The real repository/DB experiment took 6.831s; evidence is
|
||||
`/tmp/paperclip-git-recovery-candidate.json`. Expiration is not implemented yet.
|
||||
- First Git baseline used a real bare remote, runtime workspace creation, a
|
||||
database row, and ordinary reopening: missing worktree restores A while its
|
||||
branch exists; missing branch restores upstream B; missing base fails. Fixture
|
||||
script `/tmp/paperclip-git-recovery-baseline.mts`, inspected output
|
||||
`/tmp/paperclip-git-recovery-baseline.json`, retained Git fixture
|
||||
`/tmp/paperclip-git-baseline-sbJWbo`. The test DB was cleaned up.
|
||||
- Greptile cleared `0da71b2ca2` (all its findings resolved). Security review then
|
||||
identified repeated full-directory enumeration. Discovery now retains bounded
|
||||
directory iterators across continuations, counts root entries in its budget,
|
||||
closes abandoned iterators, and bounds company discovery. Twenty-one focused
|
||||
cases passed in 29.99s (`/tmp/paperclip-skill-streaming-final.log`); server
|
||||
typecheck passed. A new regression inspects actual directory reads/opener reuse.
|
||||
- No deployed artifact or staging acceptance yet. Next: publish review fixes,
|
||||
qualify final skill SHA, then perform dry-run rollout and measured restoration.
|
||||
|
||||
## Feedback-loop evidence
|
||||
|
||||
| Question | Environment / loop | Elapsed / main wait | Result / next decision |
|
||||
| --- | --- | --- | --- |
|
||||
| Can this checkout run focused tests? | Local dependency setup | 43.2s install; later one-time Runner build took several minutes | Baseline manifest/lock drift required install without changing CI-owned lockfile. Warm builds are now reusable. |
|
||||
| Can an older skill restore without upstream? | Existing service with real Postgres snapshots, pre-cleanup proof at baseline `57fa6b358d` | 4.83s warm test attempt; setup excluded as one-time above | Normal runtime path restored saved binary/script bytes and executable bits with fetch unavailable; proceed with cleanup. |
|
||||
| Do the integrated safety cases hold? | Devin's `0f3adda980` plus local follow-up; real Postgres and filesystem | 49s from launch through inspected result, 47.23s runner (62% tests, 21% import) | 22 tests passed, server typecheck passed. Added deleted-skill and failure-budget review cases plus heartbeat controls. No cloud proof claimed. |
|
||||
|
||||
| Does Git reopening preserve A after upstream moves? | `5eb505ff5d`, local bare remote plus real runtime/service/database path | 6.398s setup/run/inspection script, about 9s launch through result review | Missing branch returned B; missing base failed. Preserve fixture for exact-commit restoration tests. |
|
||||
| Does cleanup block preparation during a slow file walk? | Same checkout plus review fixes, two DB clients and a deliberately paused filesystem read | 24.94s affected suite, about 28s through inspected result | 18 tests pass; preparation completes while scan is paused and its new pin prevents eviction. |
|
||||
|
||||
## Deferred
|
||||
|
||||
Skill-version creation/deduplication, recovery retention, non-Git workspace expiry,
|
||||
shared/operator checkout expiry, and moving storage off the webserver.
|
||||
@@ -0,0 +1,47 @@
|
||||
# Recoverable disk cleanup
|
||||
|
||||
The server materializes immutable skill versions under
|
||||
`<instance>/skills/<company>/__versions__/<skill>/<version>`. These are disposable
|
||||
disk copies: the company-scoped version and repository snapshot rows retain the
|
||||
contents, including binary files and executable flags. Normal skill resolution
|
||||
rebuilds a missing copy from the database without contacting its upstream source.
|
||||
|
||||
## Skill version caches
|
||||
|
||||
An hourly background sweep considers at most 50 directories per batch, yielding
|
||||
between batches. It starts after server readiness, and respects the existing
|
||||
worker shutdown, standby, and idle-drain controls. A non-current version becomes
|
||||
eligible after one hour without use. Current versions, active-run pins and
|
||||
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.
|
||||
|
||||
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
|
||||
processes. Snapshot validation and byte counting run outside that lock; cleanup
|
||||
rechecks current versions, run pins, last use, and directory identity before rename. Audits hold the lock through their file reads. Eviction atomically
|
||||
moves a cache into unique `.trash-<UUID>` directories; trash deletion uses its own lock so normal
|
||||
preparation can restore files while deletion finishes. A later sweep finishes
|
||||
interrupted trash removal without deleting a newly materialized version.
|
||||
|
||||
Controls:
|
||||
|
||||
| Environment variable | Default | Effect |
|
||||
| --- | --- | --- |
|
||||
| `PAPERCLIP_SKILL_VERSION_RETENTION_ENABLED` | `true` | Set `false` to stop future sweeps; restoration stays available |
|
||||
| `PAPERCLIP_SKILL_VERSION_RETENTION_DRY_RUN` | `false` | Set `true` to report candidates without changing cache or access records |
|
||||
|
||||
The internal `skillVersionRetentionService` supports company-scoped diagnosis,
|
||||
bounded continuations, cancellation between entries, and dry run. Explicit agent
|
||||
pins stay cached. Operators can also retain newest versions with
|
||||
`PAPERCLIP_SKILL_VERSION_RETENTION_KEEP` (default `0`), or change the one-hour
|
||||
window with `PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES` (default `60`). Each batch reports
|
||||
candidate and reclaimed allocated bytes, eviction counts, and skip reasons in
|
||||
the server log. Compare volume usage before and after to verify physical savings;
|
||||
concurrent writes and filesystem allocation can affect the net change. A dry run
|
||||
does not start the legacy observation period.
|
||||
|
||||
Roll out with dry run enabled, inspect candidates/skips, then enable removal.
|
||||
Recovery archives and runner backups have separate ownership and retention;
|
||||
this worker never scans or removes them.
|
||||
@@ -89,6 +89,12 @@ remaining operations before qualifying the whole capability.
|
||||
Gaps: Revision metadata has an API; pinned execution skill versions and installed latest
|
||||
are distinct.
|
||||
|
||||
Runtime disk-cache expiration does not delete version records or change access. Normal
|
||||
execution rematerializes saved bytes after existing skill authorization; cleanup is an
|
||||
internal company-scoped maintenance operation, with no new agent endpoint. Verify pinned
|
||||
restoration and active-run protection with `server/src/__tests__/skill-version-cache.test.ts`
|
||||
and the assembled heartbeat runtime-skill tests; live acceptance remains separate.
|
||||
|
||||
## `policy`
|
||||
|
||||
Review: `baseline`.
|
||||
|
||||
@@ -65,6 +65,10 @@ Manual: Enable the skill for one test agent and leave another without access. Ru
|
||||
- Installed does not mean enabled for every agent.
|
||||
- Read-only GitHub/bundled skills may require a writable copy rather than an in-place edit.
|
||||
- Studio interaction coverage is in [questions and approvals](./questions-and-approvals.md); a passing mock is not a live harness test.
|
||||
- Old version disk caches may expire; saved versions remain available. Verify a pinned
|
||||
version is restored through normal execution with upstream unavailable. Current
|
||||
and in-use versions stay materialized. See [cleanup controls](../../doc/recoverable-disk-cleanup.md)
|
||||
and the [database-backed cache tests](../../server/src/__tests__/skill-version-cache.test.ts).
|
||||
|
||||
### Project repository source behavior
|
||||
|
||||
|
||||
@@ -1,8 +1,15 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { expect, it } from "vitest";
|
||||
import { expect, it, vi } from "vitest";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { companySkillService } from "../services/company-skills.js";
|
||||
|
||||
// Nonempty selection now first acquires a dedicated connection. Route that
|
||||
// connection to the same fail-on-access fixture; empty selection must skip it.
|
||||
vi.mock("@paperclipai/db", async importOriginal => ({
|
||||
...await importOriginal<typeof import("@paperclipai/db")>(),
|
||||
withDedicatedDbConnection: <T>(database: Db, action: (connection: Db) => Promise<T>) => action(database),
|
||||
}));
|
||||
|
||||
const databaseAccess = new Error("Runtime selection touched the inventory database");
|
||||
const db = new Proxy({} as Db, { get() { throw databaseAccess; } });
|
||||
|
||||
|
||||
@@ -0,0 +1,267 @@
|
||||
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 = withSkillVersionCacheLock(db, f.companyId, async () => {
|
||||
await f.resolve(f.versions[0]!.id);
|
||||
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 } });
|
||||
});
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,26 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { loadConfig } from "../config.ts";
|
||||
|
||||
describe("skill version retention config parsing", () => {
|
||||
afterEach(() => {
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
|
||||
it("uses the defaults when the variables are not set", () => {
|
||||
vi.stubEnv("PAPERCLIP_SKILL_VERSION_RETENTION_KEEP", undefined);
|
||||
vi.stubEnv("PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES", undefined);
|
||||
expect(loadConfig()).toMatchObject({ skillVersionRetentionKeepNewest: 0, skillVersionRetentionGraceMinutes: 60 });
|
||||
});
|
||||
|
||||
it("accepts explicit non-negative values, including 0", () => {
|
||||
vi.stubEnv("PAPERCLIP_SKILL_VERSION_RETENTION_KEEP", " 0 ");
|
||||
vi.stubEnv("PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES", "30");
|
||||
expect(loadConfig()).toMatchObject({ skillVersionRetentionKeepNewest: 0, skillVersionRetentionGraceMinutes: 30 });
|
||||
});
|
||||
|
||||
it.each(["", " ", "-1", "many"])("falls back to the defaults for %j", (value) => {
|
||||
vi.stubEnv("PAPERCLIP_SKILL_VERSION_RETENTION_KEEP", value);
|
||||
vi.stubEnv("PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES", value);
|
||||
expect(loadConfig()).toMatchObject({ skillVersionRetentionKeepNewest: 0, skillVersionRetentionGraceMinutes: 60 });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,370 @@
|
||||
import { touchSkillVersionCache } from "../services/skill-version-cache.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promises as fs } from "node:fs";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
companySkills,
|
||||
companySkillVersions,
|
||||
createDb,
|
||||
heartbeatRuns,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { companySkillService } from "../services/company-skills.ts";
|
||||
import {
|
||||
resolveCompanySkillVersionsRoot,
|
||||
runVersionSnapshotFlight,
|
||||
skillVersionRetentionService,
|
||||
} from "../services/skill-version-retention.ts";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
if (!embeddedPostgresSupport.supported) {
|
||||
console.warn(
|
||||
`Skipping embedded Postgres skill version retention tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
|
||||
);
|
||||
}
|
||||
|
||||
const HOUR_MS = 60 * 60 * 1000;
|
||||
|
||||
async function exists(target: string) {
|
||||
return fs.lstat(target).then(() => true, () => false);
|
||||
}
|
||||
|
||||
async function makeWritable(target: string): Promise<void> {
|
||||
const stat = await fs.lstat(target).catch(() => null);
|
||||
if (!stat?.isDirectory()) return;
|
||||
await fs.chmod(target, 0o755);
|
||||
for (const entry of await fs.readdir(target)) await makeWritable(path.join(target, entry));
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("skill version snapshot retention", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let oldPaperclipHome: string | undefined;
|
||||
let oldPaperclipInstanceId: string | undefined;
|
||||
let paperclipHome: string | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-skill-version-retention-");
|
||||
oldPaperclipHome = process.env.PAPERCLIP_HOME;
|
||||
oldPaperclipInstanceId = process.env.PAPERCLIP_INSTANCE_ID;
|
||||
paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-skill-version-retention-home-"));
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "default";
|
||||
db = createDb(tempDb.connectionString);
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agents);
|
||||
await db.delete(companySkills);
|
||||
await db.delete(companies);
|
||||
if (paperclipHome) {
|
||||
const skillsRoot = path.join(paperclipHome, "instances", "default", "skills");
|
||||
await makeWritable(skillsRoot);
|
||||
await fs.rm(skillsRoot, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
if (oldPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME;
|
||||
else process.env.PAPERCLIP_HOME = oldPaperclipHome;
|
||||
if (oldPaperclipInstanceId === undefined) delete process.env.PAPERCLIP_INSTANCE_ID;
|
||||
else process.env.PAPERCLIP_INSTANCE_ID = oldPaperclipInstanceId;
|
||||
if (paperclipHome) await fs.rm(paperclipHome, { recursive: true, force: true });
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
async function createCompany() {
|
||||
const companyId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Retention", issuePrefix: `R${companyId.slice(0, 6)}` });
|
||||
return companyId;
|
||||
}
|
||||
|
||||
async function createSkill(companyId: string, versionCount: number, currentIndex = versionCount - 1) {
|
||||
const skillId = randomUUID();
|
||||
const key = `company/${companyId}/retained-${skillId.slice(0, 8)}`;
|
||||
await db.insert(companySkills).values({
|
||||
id: skillId, companyId, key, slug: `retained-${skillId.slice(0, 8)}`, name: "Retained",
|
||||
markdown: "# Retained", sourceType: "github", sourceLocator: "https://github.com/acme/retained",
|
||||
sourceRef: "a".repeat(40), trustLevel: "markdown_only", compatibility: "compatible",
|
||||
metadata: {}, fileInventory: [{ path: "SKILL.md", kind: "skill" }],
|
||||
});
|
||||
const versionIds: string[] = [];
|
||||
for (let index = 0; index < versionCount; index++) {
|
||||
const [version] = await db.insert(companySkillVersions).values({
|
||||
companyId, companySkillId: skillId, revisionNumber: index + 1, label: `v${index + 1}`,
|
||||
fileInventory: [
|
||||
{ path: "SKILL.md", kind: "skill", content: `---\nname: retained\n---\n\n# Revision ${index + 1}\n` },
|
||||
{ path: "references/notes.md", kind: "reference", content: `notes ${index + 1}` },
|
||||
],
|
||||
}).returning({ id: companySkillVersions.id });
|
||||
versionIds.push(version!.id);
|
||||
}
|
||||
if (versionCount > 0) {
|
||||
await db.update(companySkills).set({ currentVersionId: versionIds[currentIndex]! })
|
||||
.where(eq(companySkills.id, skillId));
|
||||
}
|
||||
return { skillId, key, versionIds };
|
||||
}
|
||||
|
||||
/** Writes a read-only snapshot directory like materializeVersionSnapshot does, aged by `ageMs`. */
|
||||
async function writeSnapshot(companyId: string, skillId: string, name: string, ageMs = 3 * HOUR_MS) {
|
||||
const dir = path.join(resolveCompanySkillVersionsRoot(companyId), skillId, name);
|
||||
await fs.mkdir(path.join(dir, "references"), { recursive: true });
|
||||
await fs.writeFile(path.join(dir, "SKILL.md"), "# Snapshot", { mode: 0o444 });
|
||||
await fs.writeFile(path.join(dir, "references", "notes.md"), "notes", { mode: 0o444 });
|
||||
await fs.chmod(path.join(dir, "references"), 0o555);
|
||||
const when = new Date(Date.now() - ageMs);
|
||||
await fs.utimes(dir, when, when);
|
||||
if (/^[a-f0-9-]{36}$/.test(name)) await touchSkillVersionCache(companyId, skillId, name, when.getTime());
|
||||
return dir;
|
||||
}
|
||||
|
||||
it("keeps current, pinned, live-run, newest, and recent versions and removes superseded ones", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { skillId, key, versionIds: v } = await createSkill(companyId, 7);
|
||||
const [agent] = await db.insert(agents).values({
|
||||
companyId, name: "Pinned agent",
|
||||
adapterConfig: { paperclipSkillSync: { desiredSkills: [{ key, versionId: v[0] }] } },
|
||||
}).returning({ id: agents.id });
|
||||
await db.insert(heartbeatRuns).values([
|
||||
{ companyId, agentId: agent!.id, status: "running", contextSnapshot: { skillVersions: { [key]: v[1] } } },
|
||||
{ companyId, agentId: agent!.id, status: "succeeded", finishedAt: new Date(), contextSnapshot: { skillVersions: { [key]: v[2] } } },
|
||||
]);
|
||||
const dirs = await Promise.all(v.map((id, index) => writeSnapshot(companyId, skillId, id, index === 3 ? 0 : 3 * HOUR_MS)));
|
||||
|
||||
const result = await skillVersionRetentionService(db).pruneCompany(companyId, { keepNewest: 2 });
|
||||
|
||||
// v1 pinned, v2 live run, v4 recent, v6 + v7 newest (v7 current).
|
||||
expect(await Promise.all(dirs.map(exists))).toEqual([true, true, false, true, false, true, true]);
|
||||
expect(result).toMatchObject({ removed: 2, removedOrphans: 0, skippedRecent: 1, failed: 0, truncated: false });
|
||||
// Rows are never deleted.
|
||||
expect(await db.select().from(companySkillVersions).where(eq(companySkillVersions.companySkillId, skillId))).toHaveLength(7);
|
||||
});
|
||||
|
||||
it("removes orphan version folders, orphan skill folders, and stale staging folders", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { skillId, versionIds } = await createSkill(companyId, 1);
|
||||
const root = resolveCompanySkillVersionsRoot(companyId);
|
||||
const current = await writeSnapshot(companyId, skillId, versionIds[0]!);
|
||||
const orphanVersion = await writeSnapshot(companyId, skillId, randomUUID());
|
||||
const orphanSkillId = randomUUID();
|
||||
const orphanSkillVersion = await writeSnapshot(companyId, orphanSkillId, randomUUID());
|
||||
const staleStaging = await writeSnapshot(companyId, skillId, `.${versionIds[0]}-abc123`);
|
||||
const freshStaging = await writeSnapshot(companyId, skillId, `.${versionIds[0]}-def456`, 0);
|
||||
const leftoverTrash = await writeSnapshot(companyId, skillId, `\.trash-${randomUUID()}`, 0);
|
||||
const unknown = await writeSnapshot(companyId, skillId, "operator-notes");
|
||||
const hidden = await writeSnapshot(companyId, skillId, ".notes");
|
||||
|
||||
const result = await skillVersionRetentionService(db).pruneCompany(companyId);
|
||||
|
||||
expect(await exists(current)).toBe(true);
|
||||
expect(await exists(orphanVersion)).toBe(false);
|
||||
expect(await exists(orphanSkillVersion)).toBe(false);
|
||||
expect(await fs.readdir(path.join(root, orphanSkillId))).toEqual([]);
|
||||
expect(await exists(staleStaging)).toBe(false);
|
||||
expect(await exists(freshStaging)).toBe(true);
|
||||
expect(await exists(leftoverTrash)).toBe(false);
|
||||
expect(await exists(unknown)).toBe(true);
|
||||
expect(await exists(hidden)).toBe(true);
|
||||
expect(result).toMatchObject({ removedOrphans: 2, removedStaging: 1, failed: 0 });
|
||||
expect((await fs.readdir(path.join(root, skillId))).filter(name => name.startsWith(".trash-"))).toEqual([]);
|
||||
});
|
||||
|
||||
it("re-materializes a pruned version from its row on the next resolve", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { key, versionIds } = await createSkill(companyId, 5);
|
||||
const svc = companySkillService(db);
|
||||
const selections = new Map([[key, versionIds[0]!]]);
|
||||
const first = (await svc.listRuntimeSkillEntries(companyId, { versionSelections: selections }))
|
||||
.find(entry => entry.key === key)!;
|
||||
expect(first.sourceStatus).toBe("available");
|
||||
const old = new Date(Date.now() - 3 * HOUR_MS);
|
||||
await touchSkillVersionCache(companyId, path.basename(path.dirname(first.source)), path.basename(first.source), old.getTime());
|
||||
|
||||
await skillVersionRetentionService(db).pruneCompany(companyId, { keepNewest: 1 });
|
||||
expect(await exists(first.source)).toBe(false);
|
||||
|
||||
const again = (await svc.listRuntimeSkillEntries(companyId, { versionSelections: selections }))
|
||||
.find(entry => entry.key === key)!;
|
||||
expect(again).toMatchObject({ sourceStatus: "available", source: first.source });
|
||||
expect(await fs.readFile(path.join(again.source, "SKILL.md"), "utf8")).toContain("# Revision 1");
|
||||
});
|
||||
|
||||
it("refreshes the age of a reused snapshot so the grace window protects it", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { key, versionIds } = await createSkill(companyId, 5);
|
||||
const svc = companySkillService(db);
|
||||
const selections = new Map([[key, versionIds[0]!]]);
|
||||
const first = (await svc.listRuntimeSkillEntries(companyId, { versionSelections: selections }))
|
||||
.find(entry => entry.key === key)!;
|
||||
const old = new Date(Date.now() - 3 * HOUR_MS);
|
||||
await touchSkillVersionCache(companyId, path.basename(path.dirname(first.source)), path.basename(first.source), old.getTime());
|
||||
await svc.listRuntimeSkillEntries(companyId, { versionSelections: selections });
|
||||
|
||||
const result = await skillVersionRetentionService(db).pruneCompany(companyId, { keepNewest: 1 });
|
||||
expect(result.skippedRecent).toBe(1);
|
||||
expect(await exists(first.source)).toBe(true);
|
||||
});
|
||||
|
||||
it("skips a version folder while a materialization holds it", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { skillId, versionIds } = await createSkill(companyId, 3);
|
||||
const dir = await writeSnapshot(companyId, skillId, versionIds[0]!);
|
||||
let release!: () => void;
|
||||
const held = runVersionSnapshotFlight(dir, () => new Promise<string>(resolve => { release = () => resolve(dir); }));
|
||||
|
||||
await skillVersionRetentionService(db).pruneCompany(companyId, { keepNewest: 1 });
|
||||
expect(await exists(dir)).toBe(true);
|
||||
release();
|
||||
await held;
|
||||
await skillVersionRetentionService(db).pruneCompany(companyId, { keepNewest: 1 });
|
||||
expect(await exists(dir)).toBe(false);
|
||||
});
|
||||
|
||||
it("sweeps deleted skills after protecting existing runs", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { skillId, versionIds } = await createSkill(companyId, 2);
|
||||
const first = await writeSnapshot(companyId, skillId, versionIds[0]!);
|
||||
const second = await writeSnapshot(companyId, skillId, versionIds[1]!);
|
||||
const [agent] = await db.insert(agents).values({ companyId, name: "Detached agent" }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running",
|
||||
contextSnapshot: { skillVersions: { retained: versionIds[0] } } }).returning();
|
||||
await companySkillService(db).deleteSkill(companyId, skillId);
|
||||
expect(await exists(first)).toBe(true);
|
||||
await skillVersionRetentionService(db).pruneCompany(companyId);
|
||||
expect(await exists(first)).toBe(true);
|
||||
expect(await exists(second)).toBe(false);
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run!.id));
|
||||
await skillVersionRetentionService(db).pruneCompany(companyId);
|
||||
expect(await exists(first)).toBe(false);
|
||||
});
|
||||
|
||||
it("never removes anything outside the company's versions root", async () => {
|
||||
const companyId = await createCompany();
|
||||
const otherCompanyId = await createCompany();
|
||||
const { skillId, versionIds } = await createSkill(companyId, 3);
|
||||
const other = await createSkill(otherCompanyId, 3);
|
||||
const otherDirs = await Promise.all(other.versionIds.map(id => writeSnapshot(otherCompanyId, other.skillId, id)));
|
||||
const outside = await fs.mkdtemp(path.join(paperclipHome!, "outside-"));
|
||||
await fs.writeFile(path.join(outside, "keep.txt"), "keep");
|
||||
const root = resolveCompanySkillVersionsRoot(companyId);
|
||||
const victim = await writeSnapshot(companyId, skillId, versionIds[0]!);
|
||||
const safe = await writeSnapshot(companyId, skillId, versionIds[1]!);
|
||||
await fs.chmod(victim, 0o755);
|
||||
await fs.symlink(outside, path.join(victim, "escape"));
|
||||
const old = new Date(Date.now() - 3 * HOUR_MS);
|
||||
await fs.utimes(victim, old, old);
|
||||
await fs.mkdir(root, { recursive: true });
|
||||
await fs.symlink(outside, path.join(root, randomUUID()));
|
||||
await fs.symlink(outside, path.join(root, skillId, randomUUID()));
|
||||
|
||||
await skillVersionRetentionService(db).pruneCompany(companyId, { keepNewest: 1 });
|
||||
|
||||
expect(await exists(victim)).toBe(true);
|
||||
expect(await exists(safe)).toBe(false);
|
||||
expect(await fs.readFile(path.join(outside, "keep.txt"), "utf8")).toBe("keep");
|
||||
expect(await Promise.all(otherDirs.map(exists))).toEqual([true, true, true]);
|
||||
});
|
||||
|
||||
it("bounds removals per tick and continues on the next tick", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { skillId, versionIds } = await createSkill(companyId, 6);
|
||||
await Promise.all(versionIds.map(id => writeSnapshot(companyId, skillId, id)));
|
||||
const retention = skillVersionRetentionService(db, { keepNewest: 1, maxRemovals: 2 });
|
||||
|
||||
let total = 0;
|
||||
for (let index = 0; index < 5; index++) {
|
||||
const result = await retention.tick();
|
||||
expect(result.checked).toBeLessThanOrEqual(2);
|
||||
total += result.removed;
|
||||
if (!result.truncated) break;
|
||||
}
|
||||
expect(total).toBe(5);
|
||||
expect(await fs.readdir(path.join(resolveCompanySkillVersionsRoot(companyId), skillId))).toEqual([versionIds[5]]);
|
||||
});
|
||||
it("bounds directory reads and retains the discovery cursor between pages", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { skillId, versionIds } = await createSkill(companyId, 100);
|
||||
await Promise.all(versionIds.map(id => writeSnapshot(companyId, skillId, id)));
|
||||
const parent = path.join(resolveCompanySkillVersionsRoot(companyId), skillId);
|
||||
const originalOpen = fs.opendir.bind(fs);
|
||||
let reads = 0;
|
||||
const opened = vi.spyOn(fs, "opendir").mockImplementation(async (...args: Parameters<typeof fs.opendir>) => {
|
||||
const directory = await originalOpen(...args);
|
||||
const read = directory.read.bind(directory);
|
||||
vi.spyOn(directory, "read").mockImplementation(async () => { reads++; return read(); });
|
||||
return directory;
|
||||
});
|
||||
const stat = vi.spyOn(fs, "lstat");
|
||||
const retention = skillVersionRetentionService(db, { maxEntries: 2, dryRun: true });
|
||||
try {
|
||||
const first = await retention.tick();
|
||||
expect(first).toMatchObject({ checked: 2, truncated: true });
|
||||
expect(reads).toBeLessThanOrEqual(2);
|
||||
const inspected = () => new Set(stat.mock.calls.map(call => String(call[0]))
|
||||
.filter(file => path.dirname(file) === parent).map(file => path.basename(file)));
|
||||
const firstPaths = inspected();
|
||||
expect(firstPaths.size).toBe(1); // Opening the skill root also consumes a slot.
|
||||
stat.mockClear(); reads = 0;
|
||||
const second = await retention.tick();
|
||||
expect(second).toMatchObject({ checked: 2, truncated: true });
|
||||
expect(reads).toBe(2);
|
||||
expect(inspected().size).toBe(2);
|
||||
expect([...inspected()].some(name => firstPaths.has(name))).toBe(false);
|
||||
expect(opened.mock.calls.filter(call => call[0] === parent)).toHaveLength(1);
|
||||
} finally { await retention.dispose(); }
|
||||
});
|
||||
|
||||
it("continues beyond a page ending at an exhausted skill root", async () => {
|
||||
const companyId = await createCompany();
|
||||
const roots = [1, 2, 3].map(value => `00000000-0000-4000-8000-${String(value).padStart(12, "0")}`);
|
||||
await Promise.all([randomUUID(), randomUUID()].map(id => writeSnapshot(companyId, roots[0]!, id, 0)));
|
||||
await fs.mkdir(path.join(resolveCompanySkillVersionsRoot(companyId), roots[1]!), { recursive: true });
|
||||
const versions = Array.from({ length: 5 }, () => randomUUID()).sort();
|
||||
const paths = await Promise.all(versions.map((id, index) => writeSnapshot(companyId, roots[2]!, id, index === 4 ? 3 * HOUR_MS : 0)));
|
||||
const retention = skillVersionRetentionService(db, { maxEntries: 3 });
|
||||
let reclaimed = 0;
|
||||
for (let page = 0; page < 5; page++) {
|
||||
const result = await retention.tick();
|
||||
expect(result.checked).toBeLessThanOrEqual(3);
|
||||
reclaimed += result.removed;
|
||||
if (!result.truncated) break;
|
||||
}
|
||||
expect(reclaimed).toBe(1);
|
||||
expect(await exists(paths[4]!)).toBe(false);
|
||||
expect(await Promise.all(paths.slice(0, 4).map(exists))).toEqual([true, true, true, true]);
|
||||
});
|
||||
|
||||
it("bounds failed attempts and continues beyond them", async () => {
|
||||
const companyId = await createCompany();
|
||||
const { skillId, versionIds } = await createSkill(companyId, 6);
|
||||
await Promise.all(versionIds.map(id => writeSnapshot(companyId, skillId, id)));
|
||||
const rename = vi.spyOn(fs, "rename").mockRejectedValue(Object.assign(new Error("read-only fixture"), { code: "EACCES" }));
|
||||
const retention = skillVersionRetentionService(db, { maxRemovals: 2 });
|
||||
let checked = 0;
|
||||
let failures = 0;
|
||||
for (let index = 0; index < 5; index++) {
|
||||
const result = await retention.tick();
|
||||
expect(result.checked).toBeLessThanOrEqual(2);
|
||||
checked += result.checked;
|
||||
failures += result.failed;
|
||||
if (!result.truncated) break;
|
||||
}
|
||||
expect(checked).toBe(7); // Six versions plus the skill-root discovery slot.
|
||||
expect(failures).toBe(5);
|
||||
expect(rename).toHaveBeenCalledTimes(5);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -81,6 +81,10 @@ export interface Config {
|
||||
databaseBackupDir: string;
|
||||
workspaceReaperCooldownDays: number;
|
||||
workspaceCompactionDelayHours: number | null;
|
||||
skillVersionRetentionEnabled: boolean;
|
||||
skillVersionRetentionDryRun: boolean;
|
||||
skillVersionRetentionKeepNewest: number;
|
||||
skillVersionRetentionGraceMinutes: number;
|
||||
serveUi: boolean;
|
||||
uiDevMiddleware: boolean;
|
||||
secretsProvider: SecretProvider;
|
||||
@@ -313,6 +317,19 @@ export function loadConfig(): Config {
|
||||
&& workspaceCompactionDelayHoursRaw >= 0
|
||||
? workspaceCompactionDelayHoursRaw
|
||||
: 1;
|
||||
// Skill version retention removes materialized version directories under
|
||||
// `skills/<companyId>/__versions__/` that no current version, agent pin, or
|
||||
// live run uses. It keeps this many newest versions per skill as a warm
|
||||
// cache. A version directory used within the grace window also stays. A
|
||||
// removed directory is written again from its version row on next use.
|
||||
// Negative, empty, or non-numeric values fall back to the defaults.
|
||||
const nonNegativeIntegerEnv = (name: string, fallback: number) => {
|
||||
const raw = process.env[name]?.trim();
|
||||
const value = Number(raw);
|
||||
return raw && Number.isFinite(value) && value >= 0 ? Math.floor(value) : fallback;
|
||||
};
|
||||
const skillVersionRetentionKeepNewest = nonNegativeIntegerEnv("PAPERCLIP_SKILL_VERSION_RETENTION_KEEP", 0);
|
||||
const skillVersionRetentionGraceMinutes = nonNegativeIntegerEnv("PAPERCLIP_SKILL_VERSION_RETENTION_GRACE_MINUTES", 60);
|
||||
const bindValidationErrors = validateConfiguredBindMode({
|
||||
deploymentMode,
|
||||
deploymentExposure,
|
||||
@@ -360,6 +377,10 @@ export function loadConfig(): Config {
|
||||
databaseBackupDir,
|
||||
workspaceReaperCooldownDays,
|
||||
workspaceCompactionDelayHours,
|
||||
skillVersionRetentionEnabled: process.env.PAPERCLIP_SKILL_VERSION_RETENTION_ENABLED !== "false",
|
||||
skillVersionRetentionDryRun: process.env.PAPERCLIP_SKILL_VERSION_RETENTION_DRY_RUN === "true",
|
||||
skillVersionRetentionKeepNewest,
|
||||
skillVersionRetentionGraceMinutes,
|
||||
serveUi:
|
||||
process.env.SERVE_UI !== undefined
|
||||
? process.env.SERVE_UI === "true"
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
import { timeStartupRecovery } from "./startup-recovery-timing.js";
|
||||
import { skillSourceSyncService } from "./services/skill-source-sync.js";
|
||||
import {
|
||||
SKILL_VERSION_RETENTION_INTERVAL_MS,
|
||||
SKILL_VERSION_RETENTION_STARTUP_DELAY_MS,
|
||||
skillVersionRetentionService,
|
||||
} from "./services/skill-version-retention.js";
|
||||
import { subscribeAllCompanyLiveEvents } from "./services/live-events.js";
|
||||
import { DELIVERY_QUEUES } from "./services/delivery-work-notifications.js";
|
||||
import type { createDeliveryWorkCoordinator } from "./services/delivery-work-coordinator.js";
|
||||
@@ -1225,6 +1230,39 @@ async function startServerWithDatabaseTeardown(
|
||||
const skillSourceSyncInterval = setInterval(sweepSkillSources, 5000);
|
||||
skillSourceSyncInterval.unref?.();
|
||||
sweepSkillSources();
|
||||
// Materialized skill versions are disk caches of version rows. The sweep is
|
||||
// single-flight, bounded per tick, and stops between removals on shutdown or
|
||||
// idle drain. The first sweep runs after startup without delaying readiness.
|
||||
const skillVersionRetention = skillVersionRetentionService(db as any, {
|
||||
dryRun: config.skillVersionRetentionDryRun,
|
||||
keepNewest: config.skillVersionRetentionKeepNewest,
|
||||
graceMs: config.skillVersionRetentionGraceMinutes * 60 * 1000,
|
||||
shouldContinue: () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby(),
|
||||
});
|
||||
let skillVersionRetentionInFlight = false;
|
||||
let skillVersionRetentionContinuation: ReturnType<typeof setTimeout> | undefined;
|
||||
const sweepSkillVersionSnapshots = () => {
|
||||
if (!config.skillVersionRetentionEnabled || skillVersionRetentionInFlight || isIdleTaskDrainActive() || heartbeatSchedulerStopped || isWarmStandby()) return;
|
||||
skillVersionRetentionInFlight = true;
|
||||
trackHeartbeatSchedulerWork(skillVersionRetention.tick()
|
||||
.then((result) => {
|
||||
if (result.checked > 0 || result.failed > 0 || Object.keys(result.skipped).length > 0) {
|
||||
logger.info({ ...result, dryRun: config.skillVersionRetentionDryRun }, "skill version snapshot retention");
|
||||
}
|
||||
if (result.truncated && !heartbeatSchedulerStopped) {
|
||||
skillVersionRetentionContinuation = setTimeout(sweepSkillVersionSnapshots, 1000);
|
||||
skillVersionRetentionContinuation.unref?.();
|
||||
}
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "skill version snapshot retention failed");
|
||||
})
|
||||
.finally(() => { skillVersionRetentionInFlight = false; }));
|
||||
};
|
||||
const skillVersionRetentionInterval = setInterval(sweepSkillVersionSnapshots, SKILL_VERSION_RETENTION_INTERVAL_MS);
|
||||
skillVersionRetentionInterval.unref?.();
|
||||
const skillVersionRetentionStartup = setTimeout(sweepSkillVersionSnapshots, SKILL_VERSION_RETENTION_STARTUP_DELAY_MS);
|
||||
skillVersionRetentionStartup.unref?.();
|
||||
const externalObjects = externalObjectService(db as any, {
|
||||
pluginWorkerManager,
|
||||
enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true,
|
||||
@@ -1977,6 +2015,10 @@ async function startServerWithDatabaseTeardown(
|
||||
await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]);
|
||||
heartbeatSchedulerStopped = true;
|
||||
clearInterval(skillSourceSyncInterval);
|
||||
clearInterval(skillVersionRetentionInterval);
|
||||
clearTimeout(skillVersionRetentionStartup);
|
||||
clearTimeout(skillVersionRetentionContinuation);
|
||||
await skillVersionRetention.dispose();
|
||||
await skillSourceSync.stop();
|
||||
unsubscribeChatCompletions();
|
||||
await deliveryWork.stop();
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ensureSkillCacheDirectory, touchSkillVersionCache, withSkillVersionCacheLock } from "./skill-version-cache.js";
|
||||
import { retainProjectSkillAccess, skillAccessService } from "./skill-access.js";
|
||||
import type { AuthorizationActor } from "./authorization.js";
|
||||
import { updateAgentConfigurationInTransaction } from "./agent-configuration-transaction.js";
|
||||
@@ -34,6 +35,7 @@ import {
|
||||
import { readPaperclipSkillSyncPreference, writePaperclipSkillSyncPreference, resolvePaperclipDesiredSkillNames,
|
||||
resolveLegacyPaperclipDesiredSkillNames } from "@paperclipai/adapter-utils/server-utils";
|
||||
import { REPOSITORY_CONTENT_DIR, repositorySkillProjection } from "./skill-repository-package.js";
|
||||
import { runVersionSnapshotFlight } from "./skill-version-retention.js";
|
||||
import type { PaperclipDesiredSkillEntry, PaperclipSkillEntry } from "@paperclipai/adapter-utils/server-utils";
|
||||
import type {
|
||||
AgentDesiredSkillEntry,
|
||||
@@ -281,8 +283,6 @@ type ParsedSkillImportSource = {
|
||||
warnings: string[];
|
||||
};
|
||||
|
||||
const versionSnapshotFlights = new Map<string, Promise<string>>();
|
||||
|
||||
const EXTERNAL_SKILL_SOURCE_TYPES = new Set<CompanySkillSourceType>(["github", "skills_sh", "url"]);
|
||||
|
||||
function isPinnedCommitRef(value: string | null | undefined) {
|
||||
@@ -3533,17 +3533,20 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
const skill = await getById(companyId, skillId);
|
||||
if (!skill) return null;
|
||||
if (skill.metadata?.skillSourceId && skill.metadata?.snapshotHash && skill.currentVersionId) {
|
||||
const version = await getVersion(companyId, skill.id, skill.currentVersionId);
|
||||
if (!version) throw unprocessable("Installed skill version is unavailable.");
|
||||
const sourceLocator = await materializeVersionSnapshot(companyId, skill, version);
|
||||
const audit = await auditInstalledSkillBytes({
|
||||
...skill,
|
||||
sourceType: "local_path",
|
||||
sourceLocator,
|
||||
fileInventory: await runtimeVersionInventory(companyId, version),
|
||||
const versionId = skill.currentVersionId;
|
||||
return withSkillVersionCacheLock(db, companyId, async () => {
|
||||
const version = await getVersion(companyId, skill.id, versionId);
|
||||
if (!version) throw unprocessable("Installed skill version is unavailable.");
|
||||
const sourceLocator = await materializeVersionSnapshot(companyId, skill, version);
|
||||
const audit = await auditInstalledSkillBytes({
|
||||
...skill,
|
||||
sourceType: "local_path",
|
||||
sourceLocator,
|
||||
fileInventory: await runtimeVersionInventory(companyId, version),
|
||||
});
|
||||
await persistAuditMetadata(skill, audit);
|
||||
return audit;
|
||||
});
|
||||
await persistAuditMetadata(skill, audit);
|
||||
return audit;
|
||||
}
|
||||
if (skill.sourceType !== "catalog" && skill.sourceType !== "local_path") {
|
||||
throw unprocessable("Only local-path and catalog-managed company skills support audit.");
|
||||
@@ -5963,13 +5966,15 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
} else if (entry.isFile()) {
|
||||
out.push(normalizePortablePath(relativePath));
|
||||
} else {
|
||||
out.push(normalizePortablePath(relativePath));
|
||||
throw unprocessable("Unsafe file in materialized skill version.");
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
try {
|
||||
const rootStat = await fs.lstat(root);
|
||||
if (!rootStat.isDirectory() || rootStat.isSymbolicLink()) throw unprocessable("Unsafe materialized skill directory.");
|
||||
return await walk(root, "");
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
|
||||
@@ -6013,14 +6018,20 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
}
|
||||
|
||||
async function materializeVersionSnapshot(companyId: string, skill: CompanySkill, version: CompanySkillVersion): Promise<string> {
|
||||
return withSkillVersionCacheLock(db, companyId, async () => {
|
||||
const directory = await materializeVersionSnapshotLocked(companyId, skill, version);
|
||||
await touchSkillVersionCache(companyId, skill.id, version.id);
|
||||
return directory;
|
||||
});
|
||||
}
|
||||
|
||||
async function materializeVersionSnapshotLocked(companyId: string, skill: CompanySkill, version: CompanySkillVersion): Promise<string> {
|
||||
version = { ...version, fileInventory: await runtimeVersionInventory(companyId, version) };
|
||||
const runtimeRoot = path.resolve(resolveManagedSkillsRoot(companyId), "__versions__", skill.id);
|
||||
const skillDir = path.join(runtimeRoot, version.id);
|
||||
const existing = versionSnapshotFlights.get(skillDir);
|
||||
if (existing) return existing;
|
||||
const work = (async () => {
|
||||
return runVersionSnapshotFlight(skillDir, async () => {
|
||||
await ensureSkillCacheDirectory(runtimeRoot, true);
|
||||
if (await materializedVersionSnapshotMatches(skillDir, version)) return skillDir;
|
||||
await fs.mkdir(runtimeRoot, { recursive: true });
|
||||
const staging = await fs.mkdtemp(path.join(runtimeRoot, `.${version.id}-`));
|
||||
try {
|
||||
let wroteSkillFile = false;
|
||||
@@ -6044,9 +6055,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
}
|
||||
return skillDir;
|
||||
} finally { await fs.rm(staging, { recursive: true, force: true }); }
|
||||
})();
|
||||
versionSnapshotFlights.set(skillDir, work);
|
||||
try { return await work; } finally { versionSnapshotFlights.delete(skillDir); }
|
||||
});
|
||||
}
|
||||
|
||||
function resolveRuntimeSkillMaterializedPath(companyId: string, skill: Pick<CompanySkill, "key" | "slug">) {
|
||||
@@ -6145,6 +6154,13 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
): Promise<PaperclipSkillEntry[]> {
|
||||
if (options.onlyKeys?.size === 0 || (!options.onlyKeys && options.selectionConfig && !options.includeLegacy
|
||||
&& readPaperclipSkillSyncPreference(options.selectionConfig).desiredSkills.length === 0)) return [];
|
||||
return withSkillVersionCacheLock(db, companyId, () => listRuntimeSkillEntriesLocked(companyId, options));
|
||||
}
|
||||
|
||||
async function listRuntimeSkillEntriesLocked(
|
||||
companyId: string,
|
||||
options: RuntimeSkillEntryOptions,
|
||||
): Promise<PaperclipSkillEntry[]> {
|
||||
const skills = await listFull(companyId);
|
||||
const selectedKeys = options.onlyKeys ?? (options.selectionConfig ? new Set(
|
||||
(options.includeLegacy ? resolveLegacyPaperclipDesiredSkillNames : resolvePaperclipDesiredSkillNames)(
|
||||
@@ -7351,6 +7367,8 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
||||
logger.warn({ err: error, companyId }, "Skill deleted; quarantined source cleanup failed");
|
||||
});
|
||||
}
|
||||
// Version caches may still be linked into a captured run. The retention
|
||||
// worker removes orphan versions only after their run pins have settled.
|
||||
return deleted;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { withSkillActor } from "../skill-access.js";
|
||||
import { withSkillVersionCacheLock } from "../skill-version-cache.js";
|
||||
import type { AuthorizationActor } from "../authorization.js";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db, type agents, type projects } from "@paperclipai/db";
|
||||
@@ -208,7 +209,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
const nativeRunnerPreparationSpans: NativeRunHistoricalSpan[] = [];
|
||||
const skillsPrepareStartedAtMs = Date.now();
|
||||
let skillDeliveryConfig = effectiveResolvedConfig;
|
||||
const runtimeSkillEntries = await time("configuration_skills", async () => {
|
||||
const runtimeSkillEntries = await time("configuration_skills", () => withSkillVersionCacheLock(db, agent.companyId, async () => {
|
||||
try {
|
||||
const storedVersions = parseObject(run.contextSnapshot?.skillVersions);
|
||||
const hasStoredVersions = Object.prototype.hasOwnProperty.call(run.contextSnapshot ?? {}, "skillVersions");
|
||||
@@ -263,7 +264,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
}));
|
||||
nativeRunnerPreparationSpans.push({
|
||||
name: "skills.prepare",
|
||||
parentName: "task.prepare",
|
||||
|
||||
@@ -6,6 +6,9 @@ import { assertSkillSnapshotPath, skillFileBytes } from './skill-snapshot.js';
|
||||
export const REPOSITORY_CONTENT_DIR = '.paperclip-repository';
|
||||
export const MAX_REPOSITORY_FILES = 9_990;
|
||||
export const MAX_REPOSITORY_BYTES = 61 * 1024 * 1024;
|
||||
export function repositorySkillFrontmatter(content: string): string | undefined {
|
||||
return content.match(/^---\r?\n[\s\S]*?\r?\n---(?:\r?\n|$)/)?.[0];
|
||||
}
|
||||
const manifestSchema = z.object({
|
||||
version: z.literal(1),
|
||||
skills: z.array(z.string().min(1).max(4096)).min(1).max(1000),
|
||||
@@ -38,7 +41,7 @@ export function repositorySkillProjection(files: CompanySkillVersionFileInventor
|
||||
assertSkillSnapshotPath(skillPath);
|
||||
const skill = files.find(file => file.path === skillPath);
|
||||
if (!skill || skill.encoding === 'base64') throw new Error('Repository skill entrypoint is missing or not text.');
|
||||
const frontmatter = skill.content.match(/^---\r?\n[\s\S]*?\r?\n---(?:\r?\n|$)/)?.[0];
|
||||
const frontmatter = repositorySkillFrontmatter(skill.content);
|
||||
if (!frontmatter) throw new Error('Repository skill entrypoint has no frontmatter.');
|
||||
const target = `${REPOSITORY_CONTENT_DIR}/${skillPath}`;
|
||||
const wrapper = `${frontmatter.trimEnd()}\n\nRead and follow the complete skill instructions in [${skillPath}](${target.split('/').map(encodeURIComponent).join('/')}). Resolve the original skill's relative file references from its directory inside \`${REPOSITORY_CONTENT_DIR}/\`. That directory contains the included pinned repository files, including supporting skills and shared files. Repository-root paths are relative to that directory.\n`;
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
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 { companySkillVersions, companySkillRepositorySnapshots, heartbeatRuns,
|
||||
nativeRunFinalizations, environmentLeases, withDedicatedDbConnection, type Db } from "@paperclipai/db";
|
||||
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";
|
||||
|
||||
export const SKILL_VERSION_CACHE_IDLE_MS = 60 * 60 * 1000;
|
||||
const locks = new AsyncLocalStorage<Array<{ db: Db; companyId: string }>>();
|
||||
const uuid = /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i;
|
||||
export class SkillCacheBusy extends Error {}
|
||||
// Repository snapshots are immutable and referenced with restrictive foreign
|
||||
// keys. Cache only validated entrypoint names, never large file contents, so an
|
||||
// hourly sweep does not transfer the same repository for each skill/version.
|
||||
const verifiedSnapshots = new WeakMap<Db, Map<string, ReadonlySet<string>>>();
|
||||
|
||||
/** Hold through publication AND durable run pinning, not just file creation. */
|
||||
export async function withSkillVersionCacheLock<T>(db: Db, companyId: string, action: () => Promise<T>, tryOnly = false): Promise<T> {
|
||||
const held = locks.getStore() ?? [];
|
||||
if (held.some(lock => lock.db === db && lock.companyId === companyId)) return action();
|
||||
// Preparation uses the ordinary pool for version/pin writes. Borrowing a pool
|
||||
// connection for this lock would deadlock installations with a one-slot pool.
|
||||
return withDedicatedDbConnection(db, connection => connection.transaction(async tx => {
|
||||
const key = `skill_version_cache:${companyId}`;
|
||||
if (tryOnly) {
|
||||
const rows = await tx.execute<{ acquired: boolean }>(sql`select pg_try_advisory_xact_lock(hashtextextended(${key}, 0)) as acquired`);
|
||||
if (!rows[0]?.acquired) throw new SkillCacheBusy();
|
||||
} else await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${key}, 0))`);
|
||||
return locks.run([...held, { db, companyId }], action);
|
||||
}));
|
||||
}
|
||||
|
||||
export function skillVersionCachePaths(companyId: string, skillId: string, versionId: string) {
|
||||
if (![companyId, skillId, versionId].every(value => uuid.test(value))) throw new Error("Invalid skill cache identity");
|
||||
const root = path.join(resolvePaperclipInstanceRoot(), "skills", companyId);
|
||||
return {
|
||||
root,
|
||||
directory: path.join(root, "__versions__", skillId, versionId),
|
||||
access: path.join(root, "__version_access__", skillId, `${versionId}.json`),
|
||||
};
|
||||
}
|
||||
|
||||
/** Never traverse a symlink below the configured instance root. */
|
||||
export async function ensureSkillCacheDirectory(directory: string, create = false) {
|
||||
const root = path.resolve(resolvePaperclipInstanceRoot());
|
||||
const relative = path.relative(root, directory);
|
||||
if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) throw new Error("Skill cache escaped instance root");
|
||||
if (create) await fs.mkdir(root, { recursive: true });
|
||||
let cursor = root;
|
||||
for (const part of relative.split(path.sep)) {
|
||||
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 skill cache directory");
|
||||
}
|
||||
}
|
||||
|
||||
export async function touchSkillVersionCache(companyId: string, skillId: string, versionId: string, now = Date.now()) {
|
||||
const { access } = skillVersionCachePaths(companyId, skillId, versionId);
|
||||
await ensureSkillCacheDirectory(path.dirname(access), true);
|
||||
const temporary = `${access}.${randomUUID()}.tmp`;
|
||||
try {
|
||||
await fs.writeFile(temporary, JSON.stringify({ version: 1, lastUsedAt: now }), { flag: "wx", mode: 0o600 });
|
||||
await fs.rename(temporary, access);
|
||||
} finally { await fs.rm(temporary, { force: true }); }
|
||||
}
|
||||
|
||||
export async function lastUsedAt(filename: string): Promise<number | null> {
|
||||
try {
|
||||
await ensureSkillCacheDirectory(path.dirname(filename));
|
||||
const file = await fs.open(filename, constants.O_RDONLY | constants.O_NOFOLLOW);
|
||||
try {
|
||||
const stat = await file.stat();
|
||||
if (!stat.isFile() || stat.size > 256) return null;
|
||||
const value = JSON.parse(await file.readFile("utf8"));
|
||||
return value.version === 1 && Number.isSafeInteger(value.lastUsedAt) && value.lastUsedAt > 0 ? value.lastUsedAt : null;
|
||||
} finally { await file.close(); }
|
||||
} catch (error) {
|
||||
if (error instanceof SyntaxError || ["ENOENT", "ELOOP"].includes((error as NodeJS.ErrnoException).code ?? "")) return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// 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'` })
|
||||
.from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), sql`(
|
||||
${heartbeatRuns.status} not in ('succeeded', 'failed', 'timed_out', 'cancelled')
|
||||
or ${heartbeatRuns.finishedAt} is null
|
||||
or ${heartbeatRuns.controllerLeaseExpiresAt} > now()
|
||||
or exists (select 1 from ${nativeRunFinalizations} f
|
||||
where f.company_id = ${heartbeatRuns.companyId} and f.run_id = ${heartbeatRuns.id}
|
||||
and (f.phase <> 'committed' or f.lease_owner is not null or f.recovery_state is not null))
|
||||
or exists (select 1 from ${environmentLeases} l
|
||||
where l.company_id = ${heartbeatRuns.companyId} and l.heartbeat_run_id = ${heartbeatRuns.id}
|
||||
and (l.status = 'pending_cleanup' or l.cleanup_status = 'failed'
|
||||
or l.metadata ? 'nativeWorkspaceExportResume'
|
||||
or (l.released_at is null and not coalesce(l.status = 'retained'
|
||||
and l.lease_policy = 'reuse_by_environment' and l.cleanup_status = 'success', false))))
|
||||
)`));
|
||||
const versions = new Set<string>();
|
||||
for (const { pins } of rows) {
|
||||
// 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)) {
|
||||
if (typeof version !== "string" || !uuid.test(version)) return null;
|
||||
versions.add(version);
|
||||
}
|
||||
}
|
||||
return versions;
|
||||
}
|
||||
|
||||
function validateInventory(inventory: CompanySkillVersionFileInventoryEntry[]) {
|
||||
const seen = new Set<string>();
|
||||
for (const file of inventory) {
|
||||
assertSkillSnapshotPath(file.path);
|
||||
if (seen.has(file.path) || typeof file.content !== "string") return false;
|
||||
seen.add(file.path);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
export async function hasRestorableSnapshot(db: Db, companyId: string, versionId: string) {
|
||||
const [version] = await db.select().from(companySkillVersions).where(and(
|
||||
eq(companySkillVersions.companyId, companyId), eq(companySkillVersions.id, versionId)));
|
||||
if (!version) return false;
|
||||
if (version.repositorySnapshotId) {
|
||||
if (!version.repositorySkillPath) return false;
|
||||
let cache = verifiedSnapshots.get(db);
|
||||
if (!cache) { cache = new Map(); verifiedSnapshots.set(db, cache); }
|
||||
const key = `${companyId}:${version.repositorySnapshotId}`;
|
||||
const verified = cache.get(key);
|
||||
if (verified) return verified.has(version.repositorySkillPath);
|
||||
const [snapshot] = await db.select({ inventory: companySkillRepositorySnapshots.fileInventory })
|
||||
.from(companySkillRepositorySnapshots).where(and(eq(companySkillRepositorySnapshots.companyId, companyId),
|
||||
eq(companySkillRepositorySnapshots.id, version.repositorySnapshotId)));
|
||||
if (!snapshot || !validateInventory(snapshot.inventory)) return false;
|
||||
const entrypoints = new Set(snapshot.inventory.filter(file => file.encoding !== "base64"
|
||||
&& repositorySkillFrontmatter(file.content)).map(file => file.path));
|
||||
if (cache.size >= 128) cache.delete(cache.keys().next().value!);
|
||||
cache.set(key, entrypoints);
|
||||
return entrypoints.has(version.repositorySkillPath);
|
||||
}
|
||||
return validateInventory(version.fileInventory) && version.fileInventory.some(file => file.path === "SKILL.md");
|
||||
}
|
||||
|
||||
// Counts allocated bytes without following links, matching what df can reclaim.
|
||||
export async function allocatedBytes(directory: string, shouldContinue: () => boolean = () => true): Promise<number> {
|
||||
if (!shouldContinue()) throw new Error("Skill cache scan stopped");
|
||||
const stat = await fs.lstat(directory);
|
||||
if (stat.isSymbolicLink() || (!stat.isDirectory() && !stat.isFile())) throw new Error("Unsafe skill cache entry");
|
||||
let bytes = stat.blocks * 512;
|
||||
if (stat.isDirectory()) {
|
||||
for (const entry of await fs.readdir(directory)) bytes += await allocatedBytes(path.join(directory, entry), shouldContinue);
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,458 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { and, asc, eq, gt, gte, inArray, lte, sql } from "drizzle-orm";
|
||||
import { agents, companies, companySkills, companySkillVersions, type Db } from "@paperclipai/db";
|
||||
import { readPaperclipSkillSyncPreference } from "@paperclipai/adapter-utils/server-utils";
|
||||
import { resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { allocatedBytes, ensureSkillCacheDirectory, hasRestorableSnapshot, lastUsedAt,
|
||||
protectedVersions, SkillCacheBusy, skillVersionCachePaths, touchSkillVersionCache,
|
||||
withSkillVersionCacheLock } from "./skill-version-cache.js";
|
||||
|
||||
/**
|
||||
* Retention for materialized skill version directories.
|
||||
*
|
||||
* `materializeVersionSnapshot` writes each selected skill version to
|
||||
* `<instance>/skills/<companyId>/__versions__/<skillId>/<versionId>/` from the
|
||||
* version row's stored file inventory. A directory is a cache: when it is
|
||||
* removed while its row survives, the next resolve writes it again. Nothing
|
||||
* removed these directories before, so they grew with every new version.
|
||||
*
|
||||
* The prune keeps a directory when it is the skill's current version, an agent
|
||||
* pins it, a non-terminal heartbeat run captured it, it is among the newest
|
||||
* versions of its skill (when configured), or it was materialized or reused within the grace
|
||||
* window. It removes every other version directory, directories whose version
|
||||
* or skill row is gone, and abandoned staging directories. Version rows are
|
||||
* never deleted here.
|
||||
*/
|
||||
|
||||
export const SKILL_VERSION_RETENTION_DEFAULT_KEEP_NEWEST = 0;
|
||||
export const SKILL_VERSION_RETENTION_DEFAULT_GRACE_MS = 60 * 60 * 1000;
|
||||
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;
|
||||
const QUERY_BATCH = 500;
|
||||
|
||||
// In-process coordination between materialization and pruning, keyed by the
|
||||
// absolute version directory. A materialization waits for a pending removal of
|
||||
// its directory; a prune skips a directory that a materialization is writing.
|
||||
const versionSnapshotFlights = new Map<string, Promise<string>>();
|
||||
const versionSnapshotRemovals = new Map<string, Promise<unknown>>();
|
||||
// Trash directories this process is deleting right now; a concurrent prune skips them.
|
||||
const activeTrash = new Set<string>();
|
||||
|
||||
export function resolveCompanySkillVersionsRoot(companyId: string) {
|
||||
return path.resolve(resolvePaperclipInstanceRoot(), "skills", companyId, VERSIONS_DIR);
|
||||
}
|
||||
|
||||
export function resolveSkillVersionSnapshotDir(companyId: string, skillId: string, versionId: string) {
|
||||
return path.join(resolveCompanySkillVersionsRoot(companyId), skillId, versionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs one materialization per version directory at a time. Concurrent callers
|
||||
* share the in-flight promise. A pending prune of the same directory finishes
|
||||
* first, so a caller never receives a path that a prune is about to remove.
|
||||
*/
|
||||
export async function runVersionSnapshotFlight(skillDir: string, work: () => Promise<string>): Promise<string> {
|
||||
for (;;) {
|
||||
const removal = versionSnapshotRemovals.get(skillDir);
|
||||
if (!removal) break;
|
||||
await removal.catch(() => undefined);
|
||||
}
|
||||
const existing = versionSnapshotFlights.get(skillDir);
|
||||
if (existing) return existing;
|
||||
const flight = work();
|
||||
versionSnapshotFlights.set(skillDir, flight);
|
||||
try { return await flight; } finally {
|
||||
if (versionSnapshotFlights.get(skillDir) === flight) versionSnapshotFlights.delete(skillDir);
|
||||
}
|
||||
}
|
||||
|
||||
function errorCode(error: unknown) {
|
||||
return (error as NodeJS.ErrnoException | null)?.code;
|
||||
}
|
||||
|
||||
async function makeTreeWritable(target: string): Promise<void> {
|
||||
const stat = await fs.lstat(target).catch(() => null);
|
||||
if (!stat || stat.isSymbolicLink()) return;
|
||||
if (stat.isDirectory()) {
|
||||
await fs.chmod(target, 0o700).catch(() => undefined);
|
||||
const entries = await fs.readdir(target).catch(() => [] as string[]);
|
||||
for (const entry of entries) await makeTreeWritable(path.join(target, entry));
|
||||
} else {
|
||||
await fs.chmod(target, 0o600).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes a snapshot tree. Snapshot files are read-only, and some platforms
|
||||
* refuse to unlink read-only entries, so a permission failure makes the tree
|
||||
* writable and retries. Symbolic links are removed, never followed.
|
||||
*/
|
||||
async function removeSnapshotTree(target: string) {
|
||||
if (activeTrash.has(target)) return;
|
||||
activeTrash.add(target);
|
||||
try {
|
||||
await fs.rm(target, { recursive: true, force: true });
|
||||
} catch (error) {
|
||||
const code = errorCode(error);
|
||||
if (code !== "EACCES" && code !== "EPERM" && code !== "ENOTEMPTY") throw error;
|
||||
await makeTreeWritable(target);
|
||||
await fs.rm(target, { recursive: true, force: true });
|
||||
} finally {
|
||||
activeTrash.delete(target);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Moves a directory out of its readable name in one atomic rename under the
|
||||
* shared cache lock. Its unique trash name can be deleted after releasing it.
|
||||
*/
|
||||
async function removeVersionDirectory(parent: string, name: string): Promise<string | null> {
|
||||
const target = path.join(parent, name);
|
||||
if (path.dirname(target) !== parent) return null;
|
||||
if (versionSnapshotFlights.has(target) || versionSnapshotRemovals.has(target)) return null;
|
||||
const trash = path.join(parent, `${TRASH_PREFIX}${randomUUID()}`);
|
||||
const rename = fs.rename(target, trash);
|
||||
versionSnapshotRemovals.set(target, rename);
|
||||
try {
|
||||
await rename;
|
||||
} catch (error) {
|
||||
if (errorCode(error) === "ENOENT") return null;
|
||||
throw error;
|
||||
} finally {
|
||||
versionSnapshotRemovals.delete(target);
|
||||
}
|
||||
return trash;
|
||||
}
|
||||
|
||||
type CompanyCursor = { skillId: string; name: string };
|
||||
export type SkillVersionRetentionCursor = CompanyCursor & { companyId: string };
|
||||
export type SkillVersionPruneOptions = {
|
||||
keepNewest?: number;
|
||||
graceMs?: number;
|
||||
maxRemovals?: number;
|
||||
/** Bound inspection too, including protected directories. */
|
||||
maxEntries?: number;
|
||||
now?: Date;
|
||||
dryRun?: boolean;
|
||||
shouldContinue?: () => boolean;
|
||||
};
|
||||
export type SkillVersionPruneResult = {
|
||||
checked: number; candidates: number; candidateBytes: number; reclaimedBytes: number;
|
||||
removed: number; removedOrphans: number; removedStaging: number; kept: number;
|
||||
skippedRecent: number; failed: number; truncated: boolean;
|
||||
skipped: Record<string, number>;
|
||||
cursor: CompanyCursor | null;
|
||||
};
|
||||
function emptyResult(): SkillVersionPruneResult {
|
||||
return { checked: 0, candidates: 0, candidateBytes: 0, reclaimedBytes: 0,
|
||||
removed: 0, removedOrphans: 0, removedStaging: 0, kept: 0,
|
||||
skippedRecent: 0, failed: 0, truncated: false, skipped: {}, cursor: null };
|
||||
}
|
||||
|
||||
type Candidate = { parent: string; skillId: string; name: string; unsafe?: boolean; discoveryOnly?: boolean };
|
||||
|
||||
// Retain directory handles between pages instead of re-reading/sorting N names
|
||||
// for every 50-entry continuation. Every directory entry consumes a scan slot.
|
||||
async function* discoverDirectories(root: string): AsyncGenerator<Candidate> {
|
||||
await ensureSkillCacheDirectory(root);
|
||||
const directory = await fs.opendir(root, { bufferSize: 1 });
|
||||
try {
|
||||
for (;;) {
|
||||
const entry = await directory.read();
|
||||
if (!entry) break;
|
||||
if (!UUID_PATTERN.test(entry.name)) {
|
||||
yield { parent: root, skillId: "", name: entry.name };
|
||||
continue;
|
||||
}
|
||||
const parent = path.join(root, entry.name);
|
||||
yield { parent, skillId: entry.name, name: "", discoveryOnly: true };
|
||||
let childDirectory;
|
||||
try {
|
||||
await ensureSkillCacheDirectory(parent);
|
||||
childDirectory = await fs.opendir(parent, { bufferSize: 1 });
|
||||
} catch {
|
||||
yield { parent, skillId: entry.name, name: "", unsafe: true };
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
for (;;) {
|
||||
const child = await childDirectory.read();
|
||||
if (!child) break;
|
||||
yield { parent, skillId: entry.name, name: child.name };
|
||||
}
|
||||
} finally { await childDirectory.close(); }
|
||||
}
|
||||
} finally { await directory.close(); }
|
||||
}
|
||||
|
||||
function chunks<T>(values: T[], size: number): T[][] {
|
||||
const out: T[][] = [];
|
||||
for (let index = 0; index < values.length; index += size) out.push(values.slice(index, index + size));
|
||||
return out;
|
||||
}
|
||||
|
||||
export function skillVersionRetentionService(db: Db, defaults: SkillVersionPruneOptions = {}) {
|
||||
const scans = new Map<string, { iterator: AsyncGenerator<Candidate>; cursor: CompanyCursor | null;
|
||||
pending: Candidate[]; expiry: ReturnType<typeof setTimeout> }>();
|
||||
const scanning = new Set<string>();
|
||||
async function closeScan(companyId: string) {
|
||||
const scan = scans.get(companyId);
|
||||
if (!scan) return;
|
||||
scans.delete(companyId);
|
||||
clearTimeout(scan.expiry);
|
||||
await scan.iterator.return(undefined);
|
||||
}
|
||||
async function dispose() { for (const id of [...scans.keys()]) await closeScan(id); }
|
||||
|
||||
async function protectedVersionIds(companyId: string, skillIds: string[], keepNewest: number) {
|
||||
const keep = new Set<string>();
|
||||
if (skillIds.length === 0) return keep;
|
||||
for (const batch of chunks(skillIds, QUERY_BATCH)) {
|
||||
const rows = await db.select({ currentVersionId: companySkills.currentVersionId }).from(companySkills)
|
||||
.where(and(eq(companySkills.companyId, companyId), inArray(companySkills.id, batch)));
|
||||
for (const row of rows) if (row.currentVersionId) keep.add(row.currentVersionId);
|
||||
if (keepNewest > 0) {
|
||||
const ranked = db.select({
|
||||
id: companySkillVersions.id,
|
||||
rank: sql<number>`row_number() over (partition by ${companySkillVersions.companySkillId} order by ${companySkillVersions.revisionNumber} desc)`.as("rank"),
|
||||
}).from(companySkillVersions)
|
||||
.where(and(eq(companySkillVersions.companyId, companyId), inArray(companySkillVersions.companySkillId, batch)))
|
||||
.as("ranked");
|
||||
for (const row of await db.select({ id: ranked.id }).from(ranked).where(lte(ranked.rank, keepNewest))) keep.add(row.id);
|
||||
}
|
||||
}
|
||||
for (const agent of await db.select({ adapterConfig: agents.adapterConfig }).from(agents).where(eq(agents.companyId, companyId))) {
|
||||
for (const entry of readPaperclipSkillSyncPreference(agent.adapterConfig ?? {}).desiredSkillEntries) {
|
||||
if (entry.versionId) keep.add(entry.versionId);
|
||||
}
|
||||
}
|
||||
const runVersions = await protectedVersions(db, companyId);
|
||||
if (!runVersions) return null;
|
||||
for (const version of runVersions) keep.add(version);
|
||||
return keep;
|
||||
}
|
||||
|
||||
async function pruneCompanyPage(companyId: string, options: SkillVersionPruneOptions & { cursor?: CompanyCursor | null } = {}): Promise<SkillVersionPruneResult> {
|
||||
if (!UUID_PATTERN.test(companyId)) throw new Error("Invalid company ID for skill retention");
|
||||
const keepNewest = Math.max(0, Math.floor(options.keepNewest ?? defaults.keepNewest ?? SKILL_VERSION_RETENTION_DEFAULT_KEEP_NEWEST));
|
||||
const graceMs = Math.max(0, options.graceMs ?? defaults.graceMs ?? SKILL_VERSION_RETENTION_DEFAULT_GRACE_MS);
|
||||
const maxRemovals = Math.min(50, Math.max(0, options.maxRemovals ?? defaults.maxRemovals ?? SKILL_VERSION_RETENTION_DEFAULT_MAX_REMOVALS));
|
||||
const maxEntries = Math.min(maxRemovals, Math.max(1, options.maxEntries ?? defaults.maxEntries ?? 50));
|
||||
const shouldContinue = options.shouldContinue ?? defaults.shouldContinue ?? (() => true);
|
||||
const dryRun = options.dryRun ?? defaults.dryRun ?? false;
|
||||
const now = (options.now ?? new Date()).getTime();
|
||||
const result = emptyResult();
|
||||
const skip = (reason: string) => { result.skipped[reason] = (result.skipped[reason] ?? 0) + 1; };
|
||||
const trash: string[] = [];
|
||||
const root = resolveCompanySkillVersionsRoot(companyId);
|
||||
const trashName = (name: string) => name.startsWith(TRASH_PREFIX) && UUID_PATTERN.test(name.slice(TRASH_PREFIX.length));
|
||||
let scan = scans.get(companyId);
|
||||
if (!options.cursor || !scan || scan.cursor?.skillId !== options.cursor.skillId || scan.cursor?.name !== options.cursor.name) {
|
||||
await closeScan(companyId);
|
||||
// A service normally holds two descriptors for one company's continuation.
|
||||
// Bound direct/internal callers too; abandoned cursors expire automatically.
|
||||
if (scans.size >= 16) await closeScan(scans.keys().next().value!);
|
||||
const expiry = setTimeout(() => { if (!scanning.has(companyId)) void closeScan(companyId).catch(() => {}); }, 5 * 60 * 1000);
|
||||
expiry.unref();
|
||||
scan = { iterator: discoverDirectories(root), cursor: null, pending: [], expiry };
|
||||
scans.set(companyId, scan);
|
||||
}
|
||||
scan.expiry.refresh();
|
||||
const entries: Candidate[] = [];
|
||||
let discoveryDone = false;
|
||||
try {
|
||||
while (entries.length < maxEntries && shouldContinue()) {
|
||||
const next = scan.pending.length ? { done: false, value: scan.pending.shift()! } : await scan.iterator.next();
|
||||
if (next.done) { discoveryDone = true; break; }
|
||||
entries.push(next.value);
|
||||
}
|
||||
} catch (error) {
|
||||
await closeScan(companyId);
|
||||
if (errorCode(error) === "ENOENT" || errorCode(error) === "ENOTDIR") return result;
|
||||
result.checked++;
|
||||
result.failed++;
|
||||
skip("unsafe_or_unavailable");
|
||||
logger.warn({ err: error, companyId }, "skill version retention skipped unsafe company root");
|
||||
return result;
|
||||
}
|
||||
result.truncated = !discoveryDone;
|
||||
if (entries.length === 0) { await closeScan(companyId); return result; }
|
||||
const skillIds = [...new Set(entries.filter(entry => !entry.unsafe && entry.skillId).map(entry => entry.skillId))];
|
||||
let initialKeep: Set<string> | null;
|
||||
try { initialKeep = await withSkillVersionCacheLock(db, companyId, () => protectedVersionIds(companyId, skillIds, keepNewest), true); }
|
||||
catch (error) {
|
||||
result.checked = 1;
|
||||
if (error instanceof SkillCacheBusy) skip("cache_busy");
|
||||
else { result.failed++; skip("unavailable_pins"); }
|
||||
await closeScan(companyId);
|
||||
result.truncated = false;
|
||||
return result;
|
||||
}
|
||||
if (!initialKeep) { await closeScan(companyId); result.truncated = false; result.checked = 1; skip("unknown_run_pins"); return result; }
|
||||
const inspected: Array<{
|
||||
entry: Candidate; isStaging: boolean; version: boolean;
|
||||
before: { dev: number; ino: number }; bytes: number;
|
||||
eligible: (keep: Set<string> | null) => Promise<{ dev: number; ino: number } | null>;
|
||||
}> = [];
|
||||
for (const entry of entries) {
|
||||
if (!shouldContinue()) { scan.pending.push(...entries.slice(result.checked)); result.truncated = true; break; }
|
||||
result.checked++;
|
||||
result.cursor = { skillId: entry.skillId, name: entry.name };
|
||||
if (entry.unsafe) { skip("unsafe_or_unavailable"); continue; }
|
||||
if (entry.discoveryOnly) continue;
|
||||
const target = path.join(entry.parent, entry.name);
|
||||
const isTrash = trashName(entry.name);
|
||||
const isVersion = UUID_PATTERN.test(entry.name);
|
||||
const isStaging = entry.name.startsWith(".") && UUID_PATTERN.test(entry.name.slice(1, 37))
|
||||
&& /^-[a-zA-Z0-9]{6}$/.test(entry.name.slice(37));
|
||||
if (!isTrash && !isVersion && !isStaging) { skip("unknown_directory"); continue; }
|
||||
if (isTrash) { if (!dryRun) trash.push(target); continue; }
|
||||
// Called under the shared lock both before inspection and immediately
|
||||
// before rename. A preparation between these calls changes the pins/use.
|
||||
const eligible = async (keep: Set<string> | null) => {
|
||||
if (!shouldContinue()) return null;
|
||||
await ensureSkillCacheDirectory(target);
|
||||
if (versionSnapshotFlights.has(target) || versionSnapshotRemovals.has(target)) { skip("cache_busy"); return null; }
|
||||
if (!keep) { skip("unknown_run_pins"); return null; }
|
||||
if (isVersion && keep.has(entry.name)) { result.kept++; skip("protected_version"); return null; }
|
||||
const stat = await fs.lstat(target);
|
||||
if (isVersion) {
|
||||
const used = await lastUsedAt(skillVersionCachePaths(companyId, entry.skillId, entry.name).access);
|
||||
if (used === null) {
|
||||
if (!dryRun) await touchSkillVersionCache(companyId, entry.skillId, entry.name, now);
|
||||
skip("observation_grace"); return null;
|
||||
}
|
||||
if (now - used < graceMs) { result.skippedRecent++; skip("recent_use"); return null; }
|
||||
} else if (now - stat.mtimeMs < graceMs) { result.skippedRecent++; return null; }
|
||||
return stat;
|
||||
};
|
||||
try {
|
||||
const before = await withSkillVersionCacheLock(db, companyId, () => eligible(initialKeep), true);
|
||||
if (!before) continue;
|
||||
// Inventory transfer/validation and recursive file walks must not hold
|
||||
// the lock needed by agent preparation. Immutable snapshots can be
|
||||
// checked here; mutable current/run pins are checked again below.
|
||||
const [version] = isVersion ? await db.select({ id: companySkillVersions.id }).from(companySkillVersions).where(and(
|
||||
eq(companySkillVersions.companyId, companyId), eq(companySkillVersions.companySkillId, entry.skillId),
|
||||
eq(companySkillVersions.id, entry.name))) : [];
|
||||
if (version && !await hasRestorableSnapshot(db, companyId, version.id)) { skip("snapshot_unavailable"); continue; }
|
||||
const bytes = await allocatedBytes(target, shouldContinue);
|
||||
inspected.push({ entry, isStaging, version: Boolean(version), before, bytes, eligible });
|
||||
} catch (error) {
|
||||
if (!shouldContinue()) { result.truncated = true; break; }
|
||||
if (error instanceof SkillCacheBusy) skip("cache_busy");
|
||||
else if (errorCode(error) === "ENOENT") skip("not_materialized");
|
||||
else {
|
||||
result.failed++;
|
||||
skip("unsafe_or_unavailable");
|
||||
logger.warn({ err: error, companyId }, "skill version cache candidate skipped");
|
||||
}
|
||||
}
|
||||
}
|
||||
if (inspected.length && shouldContinue()) {
|
||||
try {
|
||||
// Read company-wide agent/run pins once per phase, not once per cache.
|
||||
// The lock keeps this final set authoritative through every rename.
|
||||
await withSkillVersionCacheLock(db, companyId, async () => {
|
||||
const keep = await protectedVersionIds(companyId, skillIds, keepNewest);
|
||||
for (const candidate of inspected) {
|
||||
if (!shouldContinue()) { result.truncated = true; break; }
|
||||
try {
|
||||
const current = await candidate.eligible(keep);
|
||||
if (!current) continue;
|
||||
if (current.dev !== candidate.before.dev || current.ino !== candidate.before.ino) { skip("cache_changed"); continue; }
|
||||
result.candidates++;
|
||||
result.candidateBytes += candidate.bytes;
|
||||
if (dryRun) continue;
|
||||
const renamed = await removeVersionDirectory(candidate.entry.parent, candidate.entry.name);
|
||||
if (!renamed) continue;
|
||||
trash.push(renamed);
|
||||
result.removed++;
|
||||
if (candidate.isStaging) result.removedStaging++;
|
||||
else if (!candidate.version) result.removedOrphans++;
|
||||
} catch (error) {
|
||||
if (errorCode(error) === "ENOENT") { skip("not_materialized"); continue; }
|
||||
result.failed++; skip("unsafe_or_unavailable");
|
||||
logger.warn({ err: error, companyId }, "skill version cache rename skipped");
|
||||
}
|
||||
}
|
||||
}, true);
|
||||
} catch (error) {
|
||||
if (error instanceof SkillCacheBusy) skip("cache_busy");
|
||||
else { result.failed++; skip("unavailable_pins"); }
|
||||
}
|
||||
}
|
||||
if (!result.truncated) { result.cursor = null; await closeScan(companyId); }
|
||||
else scan.cursor = result.cursor;
|
||||
// These unique names are never reused by materialization. A crash leaves
|
||||
// them for the next scan; a returning run can rehydrate immediately.
|
||||
for (const target of trash) {
|
||||
if (!shouldContinue()) break;
|
||||
try {
|
||||
await withSkillVersionCacheLock(db, `trash:${companyId}:${target}`, async () => {
|
||||
await ensureSkillCacheDirectory(target);
|
||||
const bytes = await allocatedBytes(target, shouldContinue);
|
||||
if (!shouldContinue()) return;
|
||||
await removeSnapshotTree(target);
|
||||
result.reclaimedBytes += bytes;
|
||||
}, true);
|
||||
} catch (error) {
|
||||
if (!shouldContinue()) break;
|
||||
if (error instanceof SkillCacheBusy || errorCode(error) === "ENOENT") continue;
|
||||
result.failed++;
|
||||
logger.warn({ err: error, companyId }, "skill version snapshot trash removal failed");
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
async function pruneCompany(companyId: string, options: SkillVersionPruneOptions & { cursor?: CompanyCursor | null } = {}) {
|
||||
if (scanning.has(companyId)) return { ...emptyResult(), checked: 1, skipped: { scan_busy: 1 } };
|
||||
scanning.add(companyId);
|
||||
try { return await pruneCompanyPage(companyId, options); }
|
||||
finally { scanning.delete(companyId); }
|
||||
}
|
||||
|
||||
// A continuation visits later companies and protected directories too; a busy
|
||||
// first company or a dry run cannot starve the remainder of the instance.
|
||||
let cursor: SkillVersionRetentionCursor | null = null;
|
||||
async function tick(options: SkillVersionPruneOptions = {}) {
|
||||
const maxRemovals = Math.min(50, options.maxRemovals ?? defaults.maxRemovals ?? SKILL_VERSION_RETENTION_DEFAULT_MAX_REMOVALS);
|
||||
const maxEntries = Math.min(maxRemovals, options.maxEntries ?? defaults.maxEntries ?? 50);
|
||||
const shouldContinue = options.shouldContinue ?? defaults.shouldContinue ?? (() => true);
|
||||
const totals = { ...emptyResult(), companies: 0, cursor: null as SkillVersionRetentionCursor | null };
|
||||
const companyRows = await db.select({ id: companies.id }).from(companies)
|
||||
.where(cursor ? cursor.skillId === "\uffff" ? gt(companies.id, cursor.companyId) : gte(companies.id, cursor.companyId) : undefined)
|
||||
.orderBy(asc(companies.id)).limit(51);
|
||||
for (const company of companyRows) {
|
||||
if (cursor && (company.id < cursor.companyId || (company.id === cursor.companyId && cursor.skillId === "\uffff"))) continue;
|
||||
if (totals.companies >= 50 || totals.checked >= maxEntries || totals.candidates >= maxRemovals || !shouldContinue()) { totals.truncated = true; break; }
|
||||
const result = await pruneCompany(company.id, { ...options, cursor: cursor?.companyId === company.id ? cursor : null,
|
||||
maxEntries: maxEntries - totals.checked, maxRemovals: maxRemovals - totals.candidates, shouldContinue });
|
||||
totals.companies++;
|
||||
for (const key of ["checked", "candidates", "candidateBytes", "reclaimedBytes", "removed", "removedOrphans", "removedStaging", "kept", "skippedRecent", "failed"] as const) totals[key] += result[key];
|
||||
for (const [reason, count] of Object.entries(result.skipped)) totals.skipped[reason] = (totals.skipped[reason] ?? 0) + count;
|
||||
cursor = { companyId: company.id, skillId: "\uffff", name: "" };
|
||||
if (result.truncated) {
|
||||
cursor = { companyId: company.id, ...(result.cursor ?? { skillId: "", name: "" }) };
|
||||
totals.truncated = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!totals.truncated) cursor = null;
|
||||
totals.cursor = cursor;
|
||||
return totals;
|
||||
}
|
||||
return { pruneCompany, tick, dispose };
|
||||
}
|
||||
Reference in new issue
Block a user