From 86c965ce7ab1499864f74b8dd6ed0d2d8db15197 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Thu, 15 Oct 2026 10:20:57 -0700 Subject: [PATCH] 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 Co-Authored-By: Paperclip --- doc/DEVELOPING.md | 42 ++ doc/PRODUCT.md | 4 + doc/SPEC-implementation.md | 6 + .../2026-10-15-recoverable-disk-cleanup.md | 129 +++++ doc/recoverable-disk-cleanup.md | 47 ++ maps/agent-access-map/skills.md | 6 + maps/feature-map/skills.md | 4 + .../company-skills-empty-runtime.test.ts | 9 +- .../src/__tests__/skill-version-cache.test.ts | 267 ++++++++++ .../skill-version-retention-config.test.ts | 26 + .../__tests__/skill-version-retention.test.ts | 370 ++++++++++++++ server/src/config.ts | 21 + server/src/index.ts | 42 ++ server/src/services/company-skills.ts | 58 ++- .../heartbeat/execution-configuration.ts | 5 +- .../src/services/skill-repository-package.ts | 5 +- server/src/services/skill-version-cache.ts | 166 +++++++ .../src/services/skill-version-retention.ts | 458 ++++++++++++++++++ 18 files changed, 1641 insertions(+), 24 deletions(-) create mode 100644 doc/plans/2026-10-15-recoverable-disk-cleanup.md create mode 100644 doc/recoverable-disk-cleanup.md create mode 100644 server/src/__tests__/skill-version-cache.test.ts create mode 100644 server/src/__tests__/skill-version-retention-config.test.ts create mode 100644 server/src/__tests__/skill-version-retention.test.ts create mode 100644 server/src/services/skill-version-cache.ts create mode 100644 server/src/services/skill-version-retention.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 65f279c1661..09e8038a57b 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -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: +`/skills//__versions__///`. 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-` 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=` 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=` 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, diff --git a/doc/PRODUCT.md b/doc/PRODUCT.md index 4a5277ddfe4..a3a220d22d9 100644 --- a/doc/PRODUCT.md +++ b/doc/PRODUCT.md @@ -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 diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 658e1859a17..edcdb86eb8c 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -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 diff --git a/doc/plans/2026-10-15-recoverable-disk-cleanup.md b/doc/plans/2026-10-15-recoverable-disk-cleanup.md new file mode 100644 index 00000000000..d8a11ad79fc --- /dev/null +++ b/doc/plans/2026-10-15-recoverable-disk-cleanup.md @@ -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. diff --git a/doc/recoverable-disk-cleanup.md b/doc/recoverable-disk-cleanup.md new file mode 100644 index 00000000000..f27de547ac7 --- /dev/null +++ b/doc/recoverable-disk-cleanup.md @@ -0,0 +1,47 @@ +# Recoverable disk cleanup + +The server materializes immutable skill versions under +`/skills//__versions__//`. 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-` 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. diff --git a/maps/agent-access-map/skills.md b/maps/agent-access-map/skills.md index 8f8cc177d89..7f1051511af 100644 --- a/maps/agent-access-map/skills.md +++ b/maps/agent-access-map/skills.md @@ -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`. diff --git a/maps/feature-map/skills.md b/maps/feature-map/skills.md index 9559dd6d6c4..216bc7203e4 100644 --- a/maps/feature-map/skills.md +++ b/maps/feature-map/skills.md @@ -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 diff --git a/server/src/__tests__/company-skills-empty-runtime.test.ts b/server/src/__tests__/company-skills-empty-runtime.test.ts index 369546e4e40..48195e1e63c 100644 --- a/server/src/__tests__/company-skills-empty-runtime.test.ts +++ b/server/src/__tests__/company-skills-empty-runtime.test.ts @@ -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(), + withDedicatedDbConnection: (database: Db, action: (connection: Db) => Promise) => action(database), +})); + const databaseAccess = new Error("Runtime selection touched the inventory database"); const db = new Proxy({} as Db, { get() { throw databaseAccess; } }); diff --git a/server/src/__tests__/skill-version-cache.test.ts b/server/src/__tests__/skill-version-cache.test.ts new file mode 100644 index 00000000000..bd6b1f6cd49 --- /dev/null +++ b/server/src/__tests__/skill-version-cache.test.ts @@ -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, 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>; + let db: ReturnType; + 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 | 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(); + const release = Promise.withResolvers(); + 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(); + const release = Promise.withResolvers(); + const lstat = fs.lstat.bind(fs); + let held = false; + vi.spyOn(fs, "lstat").mockImplementation((async (...args: Parameters) => { + 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 | 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(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"); + }); +}); diff --git a/server/src/__tests__/skill-version-retention-config.test.ts b/server/src/__tests__/skill-version-retention-config.test.ts new file mode 100644 index 00000000000..6205e17d8f7 --- /dev/null +++ b/server/src/__tests__/skill-version-retention-config.test.ts @@ -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 }); + }); +}); diff --git a/server/src/__tests__/skill-version-retention.test.ts b/server/src/__tests__/skill-version-retention.test.ts new file mode 100644 index 00000000000..05d4d0fbfd7 --- /dev/null +++ b/server/src/__tests__/skill-version-retention.test.ts @@ -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 { + 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; + let tempDb: Awaited> | 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(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) => { + 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); + }); + +}); diff --git a/server/src/config.ts b/server/src/config.ts index 69860d95f96..8ea462b8fbe 100644 --- a/server/src/config.ts +++ b/server/src/config.ts @@ -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//__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" diff --git a/server/src/index.ts b/server/src/index.ts index cb8625e13b6..3221a851ffc 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -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 | 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(); diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index 0c4f5df645d..2637721cf6e 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -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>(); - const EXTERNAL_SKILL_SOURCE_TYPES = new Set(["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 { + 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 { 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) { @@ -6145,6 +6154,13 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo ): Promise { 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 { 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; } diff --git a/server/src/services/heartbeat/execution-configuration.ts b/server/src/services/heartbeat/execution-configuration.ts index e36ee3d5f31..f55f8068ab9 100644 --- a/server/src/services/heartbeat/execution-configuration.ts +++ b/server/src/services/heartbeat/execution-configuration.ts @@ -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", diff --git a/server/src/services/skill-repository-package.ts b/server/src/services/skill-repository-package.ts index 44f8aeb21f1..6db0a8c84e3 100644 --- a/server/src/services/skill-repository-package.ts +++ b/server/src/services/skill-repository-package.ts @@ -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`; diff --git a/server/src/services/skill-version-cache.ts b/server/src/services/skill-version-cache.ts new file mode 100644 index 00000000000..50ee27f8468 --- /dev/null +++ b/server/src/services/skill-version-cache.ts @@ -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>(); +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>>(); + +/** Hold through publication AND durable run pinning, not just file creation. */ +export async function withSkillVersionCacheLock(db: Db, companyId: string, action: () => Promise, tryOnly = false): Promise { + 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 { + 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 | null> { + const rows = await db.select({ pins: sql`${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(); + 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(); + 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 { + 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; +} + diff --git a/server/src/services/skill-version-retention.ts b/server/src/services/skill-version-retention.ts new file mode 100644 index 00000000000..4dc46985d00 --- /dev/null +++ b/server/src/services/skill-version-retention.ts @@ -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 + * `/skills//__versions__///` 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>(); +const versionSnapshotRemovals = new Map>(); +// Trash directories this process is deleting right now; a concurrent prune skips them. +const activeTrash = new Set(); + +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): Promise { + 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 { + 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 { + 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; + 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 { + 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(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; cursor: CompanyCursor | null; + pending: Candidate[]; expiry: ReturnType }>(); + const scanning = new Set(); + 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(); + 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`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 { + 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 | 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 | 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 | 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 }; +}