diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 376f551f490..a55eed6b7d5 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -3509,3 +3509,19 @@ cluster ran the same acceptance suites. No further IPC cleanup was performed. Remote checks/review and full-stack acceptance remain pending. See the [routine scheduling audit](2026-10-27-routine-schedule-sleep.md) for producer coverage, lifecycle evidence, harness corrections and the supported single-process writer model. + + +## Slice 64 — skill-version cache retention (2026-10-27) + +The hourly cache scan now reconstructs once and wakes from cache changes, +transaction-owned version/pin changes, and saved grace or terminal-controller +lease deadlines. The existing prune service retains all deletion safety checks. +One active company iterator prevents the 16-cursor cap from restarting a large +multi-company backlog. Unknown pins and failed inspections retain scoped retries; +permanent pins and fully settled history become quiet. This reason is separate +from the still-pending native-harness backup retention duty. See +[the slice audit](2026-10-27-skill-version-retention-sleep-audit.md) for the full +writer inventory, module contracts, local acceptance, and limitations. Remote +review and checks remain pending. The user's October 27 instruction now authorizes +merging qualifying sleep PRs in dependency order after Greptile 5/5 and any +present Ponytail review passes, with all required CI checks passing. diff --git a/doc/plans/2026-10-27-skill-version-retention-sleep-audit.md b/doc/plans/2026-10-27-skill-version-retention-sleep-audit.md new file mode 100644 index 00000000000..bd77107dcff --- /dev/null +++ b/doc/plans/2026-10-27-skill-version-retention-sleep-audit.md @@ -0,0 +1,459 @@ +# Skill-version snapshot retention + +Parent: slice 63, PR #17876, at +`31c12f1d68ff4a51e06fc2be2abf2c2aee9f9d3a`. Pinned integrated master: +`8548173fc9efdc0fa07644fb92dcc0b6851a95e5`. + +This slice is implemented and locally verified. Remote review and CI are pending. +It does not qualify deployed host or database suspension. + +## Existing obligation + +`skill-version-retention.ts` removes materialized filesystem caches, never the +version rows. Startup waits 60 seconds, then an independent hourly interval +selects companies and inspects their version directories. One-second timers +continue a truncated pass. The code protects current versions, configured newest +versions, agent pins, captured run pins, and recent access. It validates restorable +bytes and rechecks mutable pins under the existing cache advisory lock before +atomic rename. Trash cleanup occurs after releasing that lock. + +The cache directories, access-marker timestamps, version inventories, agent +configuration, run context, native finalizations, and environment leases already +persist the necessary facts. No per-worker table or persistent queue is needed. + +## Implemented lifecycle + +- Reconstruct once at startup/resume using bounded company and directory pages. +- Wake the affected company when cache publication/access changes, skill versions + or permanent pins change, or run/finalization/environment ownership changes. +- Schedule access or staging grace expiry from its saved timestamp. Reconstruct + terminal controller-lease expiry from its saved deadline. +- Retry concrete failed inspection, lock contention, incomplete pins, and failed + trash removal in the affected company. Unknown directories and permanently + unrestorable immutable snapshots remain protected without a blanket scan. +- Leave current/newest/agent-pinned and fully settled retained history quiet when + no timed or uncertain obligation remains. +- Preserve dirty requests through drain and changes during a page. Close cache + directory iterators only after admitted work has stopped. + +The existing scanner holds directory iterators between pages. It caps each page +at 50 inspected entries/removals, limits live scans to 16, and expires abandoned +iterators after five minutes. Starting every company's cursor in a fair round-robin +worker could exceed that cap, evict iterators, and restart large directories from +the beginning. Preserve the existing single active company continuation and bounded +company page before introducing cross-company cursor concurrency. A due-company +queue and per-company deadlines can remain in memory; startup reconstruction owns +recovery of that scheduling state. + +## Producer audit + +| State change | Owning boundary to use | +| --- | --- | +| Materialization success or interrupted staging | `runVersionSnapshotFlight` completion, including failure | +| Reuse of a materialized version | `touchSkillVersionCache` atomic marker replacement | +| New version/current/newest selection | `companySkillService.createVersion` inside its caller-owned or local transaction | +| Failed fork cleanup and skill deletion | Existing company-skill deletion transactions | +| Agent pin edit, hire approval replay, setup completion | Existing agent-configuration and hire transaction hooks | +| Agent/company deletion | Existing instruction-collection invalidation before cascades | +| Legacy/native run terminalization | Existing instruction-collection producer fanout; see slice 42 audit | +| Native finalization ownership/phase | Existing finalization notification boundary | +| Environment cleanup/export settlement | Existing native-environment-settlement and retained-session notification boundaries; reused-task prior-owner retirement uses `writeEnvironmentCleanup` | +| Terminal controller lease expiry | Saved lease deadline; no write is required at expiration | + +Before retiring the interval, verify direct terminal, finalization, lease, and pin +writers against these boundaries. In particular, a terminal run can still hold +pins while finalization or cleanup is unfinished. Terminal presentation is not +permission to delete a cache. Pin protection and immutable-byte validation remain +with the existing cache/retention service. Notification helpers only carry a +company scope; startup only wires the coordinator. + +Supported writers are the existing service/API owners in one server process. +Filesystem writes from unrelated processes and raw SQL from independent replicas +need an explicit ingress contract. Startup reconstruction does not establish live +cross-process delivery. Keep Cloud sleep-safety and alarm coverage unchanged. + +## Acceptance required + +- Existing cache rehydration, pin, lock, symlink, staging, orphan, and bounded-page + cases pass against isolated real PostgreSQL databases. +- No repeated worker SQL or filesystem scan after empty reconstruction or a pass + containing only permanent pins/completed history. +- Ordinary materialization and version/pin edits wake a quiet worker. Outer + commit wakes it; rollback does not strand an intent. +- Saved access/staging/lease deadlines are honored. Fresh access replaces an old + deadline; a pin added before rename prevents removal. +- Terminal ownership and environment settlement release caches only after their + existing durable predicates permit deletion. +- Unknown pins/contention/I/O failure retain scope and recovery. Other companies + do not require an instance-wide recurring scan. +- Bounded continuation crosses company pages and directory pages without losing + earlier grace deadlines or repeatedly evicting live iterators. +- Drain/resume, a write during a page, held startup work, disabled configuration, + and shutdown preserve ownership and readiness. + +No host suspension, Cloud wake, or fleet-cost claim follows from these local tests. + +## Implementation and writer coverage + +`skill-version-retention-work.ts` owns company discovery, one active directory +cursor, dirty company hints, scoped failures, and the earliest observed deadline. +It registers through `registerDeadlineWork`. The helper now forwards committed +scope hints through an optional `onSettled` callback. Existing consumers omit +that callback and retain their behavior. The coordinator still owns transaction +reconciliation, admission, overlap prevention, timers, and shutdown joining. +The worker stores no provider protocol and performs no SQL in `next`. + +The cache owner returns per-page deadlines with its existing prune result. The +worker retains their minimum across directory pages. Only terminal, finished +runs with a database-live controller lease contribute a lease deadline. Current, +newest, and agent-pinned versions suppress that deadline. PostgreSQL still owns +the lease predicate. If the application clock reaches the saved timestamp before +the database clock, the same concrete lease is rechecked after one second; the +worker never treats application time as proof of release. + +The producer audit covered every production writer of `currentVersionId`, agent +adapter configuration, `nativeRunFinalizations`, and `environmentLeases`: + +- `createVersion`, `cleanupFailedFork`, and `deleteSkills` signal in the transaction + that changes version/current protection. Caller-owned transactions remain owners. +- Agent configuration and hire-approval replay compare normalized desired skill + entries. Setup completion also signals because it replaces a draft agent row. + Plugin-managed edits use the canonical agent configuration path. Other agent + writers change lifecycle, budgets, identity, or metadata and do not remove pins. +- `notifyInstructionCollection` covers legacy/native terminalization and agent or + company deletion. `notifyNativeEnvironmentSettlement` includes the finalization + fanout and environment cleanup writes. `notifyRetainedNativeSessionCleanup` + covers stop/absence receipts that can release a retained run without an ordinary + environment release. Each carries only company scope to this worker. +- The direct finalization writers outside those fanouts create or maintain an + uncommitted/terminal-failure row, change failure detail, or clear an owner only + without a committed result. They do not satisfy the committed finalization + predicate that releases a captured pin. A later committed transition uses the + existing finalization notification boundary. +- Environment runtime release/destroy and lease cleanup use `environmentService` + or `writeEnvironmentCleanup`. Reused provider-task completion also releases prior + retained owners; that update now uses the same transaction boundary. Direct + credential/project-auth, cleanup-attempt, identity, and recovery-evidence metadata + updates do not change this service's ownership predicate. Removing the completed + workspace-export marker from an active, unreleased lease still leaves its lease + pin; its later release uses the ordinary cleanup boundary. +- Filesystem publication/reuse signals are local callbacks, including a rejected + snapshot flight. They do not invent a database transaction around filesystem I/O. + +Startup registers the worker only when retention is enabled. Its first promise +is tracked as background work without delaying HTTP readiness. Shutdown removes +the cache subscription, stops and joins the delivery coordinator, then closes +retained directory iterators. No additional SQL table or migration is introduced. + +## Local verification and failure record + +At the slice-64 working tree on parent `31c12f1d68ff4a51e06fc2be2abf2c2aee9f9d3a`: + +- 91 tests pass in the full retention, immutable cache, and environment-task suites. + Retention/cache cases use separate guarded external PostgreSQL databases with + migrations applied. The environment-task suite uses its existing controlled + plugin boundary. No test is skipped in this acceptance run. +- 95 startup cases pass, including disabled registration and a held reconstruction + that remains tracked while readiness is available. +- 51 existing native-environment and retained-session cleanup lifecycle cases pass + against two additional isolated PostgreSQL databases after the notification fanout + changes. The later task-owner retirement change is covered in the 91-case run. +- Server TypeScript checking and the full server build pass. Repository-wide + tests/typecheck/build have not been rerun for this slice. + +The first acceptance run exposed three external-database fixture accesses that +still read the embedded database object. The fixture now resolves both clients +from the same selected external URL, and the full suite passes. The first startup +run asserted that all unrelated mock workers had no errors; their pre-existing +minimal DB doubles cannot run five unrelated workers. The assertion now checks +this worker's error queue and readiness/active-work behavior. No production +fallback or timeout was weakened for either harness correction. + +Evidence includes empty and permanently protected history quiet for three former +hourly periods; ordinary materialization/access/version creation/deletion and agent +pin mutation; outer commit/rollback; all three settlement fanouts; saved access, +staging, and controller deadlines; database clock fencing; restart; scoped failure +and incomplete-pin recovery; 55-company discovery and 18 live directory-page +continuations; a change during a held page; drain/resume; and shutdown joining. +Existing cache cases retain byte restoration, symlink, late-pin, lock-contention, +legacy manifest, native-finalization, and interrupted-trash protection coverage. + +The related in-flight [PR #17907](https://github.com/paperclipai/paperclip/pull/17907) +adds keepalive for an actively held skill-cache advisory-lock transaction. That is +an independent lock-liveness repair. This slice preserves that lock boundary and +does not duplicate its implementation. An active-operation keepalive is distinct +from the hourly idle discovery removed here. + +## Review repair: interrupted pages and operator reports + +Greptile identified unfinished page work that the old hourly sweep had recovered. +Controlled drain during byte inspection reproduced a retained old cache; drain +immediately after rename reproduced an abandoned trash directory behind the +continuation cursor. Both cases resumed, reached apparent quiet, and still left +work behind before this repair. A third regression confirmed that dry-run +candidate/byte reports were absent. + +The prune result now records whether its continuation predicate was ever denied. +The worker restarts that company's scan after interruption, even if admission +reopens before the page returns. This recovers both candidates and trash before +disarming. Ordinary bounded-page continuation still resumes its cursor. Successful +pages again log the existing counters, company scope, and dry-run mode, so an +operator can inspect a dry run before enabling deletion. + +All three controlled regressions fail before the fix. The full 34-case retention +suite and server typechecking pass after it on the owning slice-64 branch. +Evidence: `sleep64-interrupted-page-red`, `sleep64-interrupted-page-green`, and +`sleep64-interrupted-page-types` local logs. Earlier suite/build receipts remain +at their recorded revisions; current remote CI/review is required after the push. + +## Ponytail repair: remaining deletion producer and bounded fresh scans + +The full deletion-writer audit found `reconcileLocalPathSkillSources`, in addition +to the already covered explicit delete/fork/company paths. A normal list refresh +after a local source disappears reproduced the orphan leak: the row was deleted, +but its cache remained after the worker became quiet. The reconciliation delete +now registers its company hint in the owning transaction. + +Repeated instruction-collection transactions also reproduced multiple fresh +company scans inside thirty seconds. That first repair limited successfully +observed companies while keeping one-second page continuations and actual saved +deadlines precise. It initially excluded empty and uncertain passes because a +blanket floor hid the first materialization. The later security repair below +supersedes that exclusion: timestamped per-directory hints preserve those cache +deadlines while the floor applies to every completed fresh pass. No speculative +recurring timer is armed merely to retain the limit in memory. Disposal also +removes the cache-change subscription idempotently. + +The missing-source and busy-notification cases fail before repair. The initial +five-second test used a new wall-clock read rather than the stored access marker; +it was corrected to read the authoritative marker. An intermediate full run then +caught the empty/uncertain materialization race introduced by applying the new +scan limit indiscriminately. The original materialization/grace assertions remain +intact and passed at that revision. Its owning verification passed all 37 retention cases and +server typechecking. See local logs `sleep64-pony-red`, `sleep64-pony-green`, +`sleep64-pony-final`, `sleep64-pony-focused`, `sleep64-pony-acceptance`, and +`sleep64-pony-final-types` for the failure and repair sequence. + + +## CI fixture repair: fail the orphan write after replacement allocation + +The current remote CI run exposed the warm-task spool fixture rejecting every +database transaction. Retiring the prior physical-task owners now correctly uses +the environment-settlement transaction boundary, so that blanket fault stopped +the test before it allocated the unowned replacement or reached the orphan spool. +The unchanged fixture reproduces the exact CI assertion locally: the deferred +cleanup listener is never called. + +The fault now starts after the replacement ownership write fails. Prior-owner +settlement can commit, provider cleanup still fails, and orphan persistence still +encounters the simulated database outage. All existing assertions remain: one +wake, durable spool append, retained recovery after failed flush, shared concurrent +flush, and a recovered pending-cleanup lease. The complete 19-case suite passes +after this test-only repair. A guarded dedicated external test database makes the +real transaction path available locally without relying on embedded PostgreSQL. + + +## Security review repair: scoped wakes and bounded fresh inspection + +Five real-PostgreSQL regressions reproduced remaining amplification and deadline +loss: ten settlement writes caused eleven fresh scans for an empty company; an +uncertain company was inspected three times inside its existing retry window; +runless notifications repeated global discovery nine times; administrative +settings writes repeated discovery eleven times; and a cache written after a +company scan's read lost its five-second grace deadline. All five fail before +the repair and pass afterward. + +The producer audit confirmed all thirteen `writeEnvironmentCleanup` call sites +already supply company scope, including the native backup writer named by the +scanner. The helper now requires that scope at compile time. The actual unscoped +production source was custom-image setup: those writes carry no heartbeat run +(`issueId: null, heartbeatRunId: null`) and do not release captured skill pins. +They keep the existing environment cleanup notifications but no longer imply a +global skill-cache reconstruction. Scoped environment and instruction settlement +continues to notify retention for its owning company. Instance settings uses a +separate explicit reconstruction signal when its global execution setting changes. +No supported pin-release producer was removed. + +Every completed fresh company scan now records the thirty-second floor, including +empty and uncertain results. An already saved due deadline still bypasses that +floor, and bounded page continuations retain their one-second cadence. A fresh +instance reconstruction is also coalesced for thirty seconds; changes arriving +during discovery retain one follow-up reconstruction without restarting its cursor. +None of these in-memory limits arms a recurring scan after work is reconciled. + +Local cache events carry their directory, source and timestamp. An access event +uses the timestamp just persisted in the access marker; materialization uses the +start of its actual flight as a conservative inspection hint. The worker keeps +per-directory pending deadline hints until a company scan consumes them. Reuse +replaces the earlier hint for that same directory. Hints arriving during a scan +survive its completion, so they can register a new five-second deadline even while +fresh scanning is limited. The existing retention service exposes its configured +grace duration and remains the authority for reading filesystem state, pin checks +and removal eligibility. Notifications never authorize deletion themselves. + +All 42 retention cases pass, including the five new regressions and unchanged +materialization, five-second deadline, interrupted-page, drain and cleanup tests. +The combined cache, startup, environment-settlement and retained-session selection +passes 215 cases with no skipped cases; server typechecking passes, including the +required company scope on every cleanup writer. The startup fixtures include the +new explicit grace-duration contract. Fresh remote CI, Greptile, Ponytail and the +security scan remain required after publication. + +## Follow-up: overdue company eligibility and fresh test databases + +A real PostgreSQL and filesystem regression with 410 directory entries reproduced +a thirty-second pause between continuation pages while a different company's +grace deadline was overdue. Due deadlines now become ready company work: their +old timestamps leave the deadline maps and their fresh-scan floor is cleared. +The dirty company remains owed a scan even while another company owns the active +cursor. Future directory hints and failure backoffs remain intact. The same +regression passes with all directory-page gaps below three seconds. + +The external retained-plugin-task fixture now applies pending migrations before +opening its client. All nineteen cases passed against a newly created, blank +dedicated test database. The previous fixture correction assumed a migrated +local database; it did not establish this fresh-database path. + +All 216 cases across the five full retention/cache/startup/environment/session +suites pass, along with server typechecking and build. The new pagination test +and both interruption cases also pass together. One earlier child integration +run passed 203 cases and failed because the rename-interruption hook had not +fired before its test timeout. Both interruption cases passed unchanged on the +owning branch before this fix, and again in the full 216-case selection. That +first timing failure remains recorded; these passes are not evidence that it +never happened. An initial consumer command used incorrect filenames and an +incorrect cache database override and was interrupted; it is not acceptance. +Fresh remote CI and automated reviews remain required after publishing. + +## October 29 reconciliation, review repair and current acceptance + +This existing skill-retention slice was reconciled with the qualified agent-timer +parent through normal merges. Startup preserves both retention unsubscription and +task-drain expiry shutdown; the shared deadline helper keeps the new optional +settlement callback and the parent's continuation option. Other paths match the +automatic merge. Packages remain byte-identical to the qualified profile-80 tree. +The later monitor-parent audit changes no retention runtime or selected test. + +Greptile's remaining P2 about own-company page completion was reproduced. A cache +access hint can expire while that same company's directory cursor is active. The +worker records the company as dirty and consumes the expired timestamp, but scan +completion then applies a new 30-second fresh-scan floor. A directory already +passed by the cursor therefore waits despite its expired grace window. + +The repair records a `rescanOwed` bit on the active company when its saved expiry +or cache deadline is consumed. Completion keeps that company eligible instead of +reapplying the ordinary floor. Starting the fresh scan clears the bit. Ordinary +change batching and failure backoff keep their existing behavior; no expired +timestamp is retained to force a busy loop, and no new polling or durable table is +introduced. The existing filesystem records remain restart truth. + +The new worker regression uses real cache-change publication and the real shared +coordinator with a paged prune fixture. The unchanged prior worker completes four +pages, then delays the owed fresh scan. The repaired worker starts the fifth scan +with a fresh cursor within the same deadline pass, disarms, and performs no further +scan during another minute. The same assertions and timers fail before the repair +and pass afterward. This is scheduling-boundary evidence; the complete real-filesystem +retention suite separately checks actual pruning, leases, pins and path safety. + +Source `3d06de2a8f490e62f8022b8894ba7bbc3a985752` passed **799 cases across +32 complete files**, without skips, in 350.36 seconds. Coverage includes real +skill retention/cache/configuration and producers, runtime selection/concurrency, +environment-task and plugin-warm-task behavior, native environment/finalization +and retained-session cleanup, configuration/settings, startup, and the shared +deadline consumers including agent timers, monitors and routines. Database build, +production server typechecking/build, module boundaries, target migration order +and Runner profile identity pass. Eighteen fresh PostgreSQL 17 schemas plus +per-file fixtures applied all 412 journal entries through index 413. All selected +source hashes and temporary local fixture imports were restored exactly. + +An additional fresh-database check ran all 19 plugin-warm-task cases against a +blank dedicated PostgreSQL database with no separate migration command. The +fixture itself created the schema and the full file passed in 10.28 seconds, +proving the earlier external-database migration finding is addressed. These +19 cases are already included in the 799 unique-case count, not added again. + +This is bounded local acceptance and a verified repair of the stated P2. Final +publication must adopt the actual merged agent-timer prerequisite and receive +fresh CI/security and reviews on its exact head. No full local repository-suite, +live provider, production load, external wake, measured host/database suspension +or fleet-savings claim is made. Earlier historical results and limitations above +remain attached to their original revisions. + +## Final current-master qualification before publication + +The source for final publication is reconciled normally through +`d9277b1858a2230bdb726974c90a0c9bc5c5ebe6`, including actual agent-timer squash +`ba07f85cadede4673a06c96becc4c4eed2193b93` and current master +`7e27bc5007c8f9cb8cdce06c24f9813e27f7120f`. This preserves all 24 retention-owned +paths and all 95 unrelated upstream paths byte for byte. The complete package, +UI, E2E-test and map trees equal current master. The earlier profile-80 package +identity statement describes the earlier acceptance revision, not this later +upstream context. No unrelated upstream implementation is removed. + +On the first combined master context, 36 complete server files passed **869 +cases**; six complete adapter/Runner files passed **140 cases**, with one +intentional Linux-inotify-only launcher case skipped on this macOS host. Two +Rust targets passed **28 cases**, including both configured-environment tests +and the full 26-case native backend target. The Runner and adapter packages +were rebuilt; Runner typechecking includes Rust formatting/workspace checks. +These counts describe 1,009 passing TypeScript cases across 42 complete files +and 28 passing Rust cases, rather than a full repository or live-provider run. +All selected source hashes and temporary PostgreSQL fixture imports were restored. + +Two subsequent upstream changes introduced the task activity index and compact +task reads. On the final source, **85 cases across six complete files** +passed again: actual skill retention, cache and worker, task cold-open reads, +task-list query/result/planner behavior, and migration invalid-index/concurrent +replay recovery. These focused cases overlap the earlier retention selection; +they are not added to a unique-case total. Fresh PostgreSQL 17 schemas applied +all **413 journal entries through index 414**. Database build, production server +typechecking/build, module boundaries, target migration order and generated +TypeScript/Rust release-profile consistency all passed. + +The separate Copilot source-identity check now fails unchanged on upstream +inputs: the merged OpenCode observability change adds its reserved key to the +identity-bound Rust environment source without regenerating that provider +attestation. All 72 inputs to that identity computation, including the manifest +and historical/current fixtures, are identical to current master. This is an +explicit inherited validation exception, not a claimed pass. This retention PR +does not regenerate or claim new live qualification for a provider release +identity. Fresh exact-head required CI/security and Greptile 5/5 remain merge +gates; a real actionable review finding still requires repair. + +The local own-company-deadline P2 regression, fresh blank plugin fixture check +and earlier historical failures remain documented above. No test assertion or +timeout was weakened. No deployed suspension, external wake, full local suite +or fleet-savings claim is made. + +## Current-head CI browser-readiness repair + +Head `1923b2624be25974a9f83c480534a0374127985b` received fresh Greptile +5/5 and Ponytail 5/5 with an unconditional Ship verdict and only three optional +nits. Required CI/security completed cleanly except the fifth browser shard +and its aggregate. Four connection-review journeys failed their final saved +status visibility check immediately after reload. The browser artifacts show +Loading conversation with the retained history DOM concealed. Approval, decline, +failed execution and actual continuation checks had already passed. Upstream's +task-layout change now waits for stable width frames before revealing history; +the journey also retains a second Connections tab. This is a UI-readiness race, +not evidence of lost review execution. + +All four original journeys passed unchanged on the local macOS host (2.7 minutes), +so a deterministic local reproduction is not claimed. CI's four failures and +their screenshots remain the baseline evidence. No unchanged-head CI retry was +used. The fixture now returns focus to the task tab, reloads, and explicitly +requires visible conversation history within 30 seconds before checking the +saved result. The application already bounds its first reveal at 15 seconds. +Every existing saved-result visibility assertion, provider call-count assertion, +continuation check and timeout remains in place; the new readiness condition is +separate from the existing final assertion. There is no production change. + +The complete connection-review file then passed all four supported default-server +journeys on a new isolated PostgreSQL 17 database. Its controlled-restart case +remains intentionally skipped by the default configuration, matching CI. The +product interaction, real fixture provider calls, cross-tab synchronization, +resume run identity, persisted result details and Always allow continuation all +remain exercised. Runtime/package/UI sources and all previously accepted sleep +test sources are unchanged. Fresh required CI/security and automated reviews +must be obtained on the new published head; the earlier head's passing reviews +are not reused as a current-head pass. diff --git a/server/src/__tests__/environment-task-runtime.test.ts b/server/src/__tests__/environment-task-runtime.test.ts index ad59145248d..c0f01ae6b65 100644 --- a/server/src/__tests__/environment-task-runtime.test.ts +++ b/server/src/__tests__/environment-task-runtime.test.ts @@ -1,6 +1,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { PgDialect } from "drizzle-orm/pg-core"; -import { executionWorkspaces, type Db } from "@paperclipai/db"; +import { executionWorkspaces, environmentLeases, type Db } from "@paperclipai/db"; import { environmentTaskOperationSchema, parseEnvironmentTaskResult } from "@paperclipai/plugin-sdk"; import { executeEnvironmentTask } from "../services/environment-task-runtime.js"; @@ -19,14 +19,16 @@ const row = () => ({ environment: { id: "environment", config: { pluginKey: "test.provider", driverKey: "tasks", driverConfig: {} } } as { id: string; config: Record } | null, run: { id: "run", agentId: "agent", status: "running" } as { id: string; agentId: string; status: string } | null, }); -function database(value: ReturnType | null = row(), projectRows = [{ id: projectId }], workspaceRows: unknown[] = []) { +function database(value: ReturnType | null = row(), projectRows = [{ id: projectId }], workspaceRows: unknown[] = [], ownerRows: unknown[] = []) { let selectedRows: unknown[] = projectRows; const results = [value ? [value] : [], [{ id: "issue" }], value ? [value] : []]; - const query: any = { from: (table: unknown) => { selectedRows = table === executionWorkspaces ? workspaceRows : projectRows; return query; }, leftJoin: () => query, where: vi.fn(() => query), limit: () => Promise.resolve(results.shift()), then: (resolve: (rows: typeof projectRows) => unknown) => Promise.resolve(selectedRows).then(resolve as (rows: unknown[]) => unknown) }; + const query: any = { from: (table: unknown) => { selectedRows = table === executionWorkspaces ? workspaceRows : table === environmentLeases ? ownerRows : projectRows; return query; }, leftJoin: () => query, where: vi.fn(() => query), limit: () => Promise.resolve(results.shift()), then: (resolve: (rows: typeof projectRows) => unknown) => Promise.resolve(selectedRows).then(resolve as (rows: unknown[]) => unknown) }; const sets: Record[] = []; const update: any = { set: (value: Record) => { sets.push(value); return update; }, where: () => update, returning: async () => [{ id: leaseId }], then: (resolve: (value: unknown) => unknown) => Promise.resolve([]).then(resolve) }; - return { db: { select: () => query, update: () => update } as unknown as Db, query, sets }; + const db = { select: () => query, update: () => update, + transaction: vi.fn(async (write: (tx: Db) => Promise) => write(db as unknown as Db)) }; + return { db: db as unknown as Db, query, sets, transaction: db.transaction }; } function worker(result: unknown = { kind: "accepted", taskId: "attempt-1" }) { return { getWorker: vi.fn(() => ({ supportedMethods: ["environmentTask"] })), call: vi.fn(async () => result) }; @@ -53,6 +55,19 @@ describe("plugin-provided Runner execution", () => { operation: { ...submit, projectIds: [] } })).rejects.toThrow("fresh workspace preparation unavailable"); expect(denied.call).not.toHaveBeenCalled(); }); + it("commits prior retained task-owner retirement through the environment cleanup boundary", async () => { + const value = row(); + Object.assign(value.lease, { leasePolicy: "reuse_by_environment", provider: "sandbox" }); + Object.assign(value.lease.metadata, { runnerTaskExecutionStopped: true }); + const owner = { id: "previous-lease", companyId: "company", heartbeatRunId: "previous-run", provider: "sandbox", + providerLeaseId: "attempt-1", status: "retained", leasePolicy: "retain_on_failure", cleanupStatus: "failed" }; + const fixture = database(value, [], [], [owner]); + await executeEnvironmentTask(fixture.db, worker() as never, { companyId: "company", leaseId, operation: { kind: "complete" } }); + expect(fixture.transaction).toHaveBeenCalledOnce(); + expect(fixture.sets).toEqual(expect.arrayContaining([expect.objectContaining({ status: "released", cleanupStatus: "success", + releasedAt: expect.any(Date), metadata: expect.anything() })])); + }); + it("releases fresh transfer copies only after confirmed task completion", async () => { const value = row(); const fresh = { attemptId: leaseId, sourceRunId: secondProjectId, previousWorkspaceId: null, recoverySnapshots: [] }; diff --git a/server/src/__tests__/plugin-warm-tasks.integration.test.ts b/server/src/__tests__/plugin-warm-tasks.integration.test.ts index 3846c47dae8..6eb396b259e 100644 --- a/server/src/__tests__/plugin-warm-tasks.integration.test.ts +++ b/server/src/__tests__/plugin-warm-tasks.integration.test.ts @@ -1,18 +1,20 @@ import { flushDeferredEnvironmentCleanup, hasDeferredEnvironmentCleanup, registerDeferredEnvironmentCleanup, subscribeDeferredEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; -import { agents, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, plugins, projects } from "@paperclipai/db"; +import { agents, companies, createDb, applyPendingMigrations, environmentLeases, environments, heartbeatRuns, issues, plugins, projects } from "@paperclipai/db"; import { environmentRuntimeService } from "../services/environment-runtime.js"; import * as orphanCleanup from "../services/sandbox-orphan-cleanup-spool.js"; import { resolvePluginTaskResume } from "../services/environment-task-runtime.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; -const support=await getEmbeddedPostgresTestSupport(); +const external=resolveExternalTestDatabaseUrl("PAPERCLIP_PLUGIN_WARM_TASKS_TEST_DATABASE_URL"); +const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("retained Codex plugin task leases",()=>{ let db:ReturnType,stop:()=>Promise; - beforeAll(async()=>{const started=await startEmbeddedPostgresTestDatabase("plugin-warm-tasks");db=createDb(started.connectionString);stop=started.stop;}); + beforeAll(async()=>{if(external){await applyPendingMigrations(external);db=createDb(external);stop=()=>db.$client.end({timeout:1});return;}const started=await startEmbeddedPostgresTestDatabase("plugin-warm-tasks");db=createDb(started.connectionString);stop=started.stop;}); afterAll(async()=>{await stop?.();}); async function fixture(initialProjects:string[]=[],initialExpiry?:string,deadline?:Date){ const companyId=randomUUID(),agentId=randomUUID(),issueId=randomUUID(),environmentId=randomUUID(),pluginId=randomUUID(); @@ -137,12 +139,15 @@ const support=await getEmbeddedPostgresTestSupport(); const wake=vi.fn(); const unsubscribe=subscribeDeferredEnvironmentCleanup(db,wake); const spool=mode==="spool" ? vi.spyOn(orphanCleanup,"createSandboxOrphanCleanupSpool").mockReturnValue({append,load:async()=>[],remove:async()=>{}}) : undefined; - const transaction=mode==="spool" ? vi.spyOn(db,"transaction").mockRejectedValue(Error("database unavailable")) : undefined; + let replacementFailed=false; + const transact=db.transaction.bind(db); + const transaction=mode==="spool" ? vi.spyOn(db,"transaction").mockImplementation((...args)=>replacementFailed + ? Promise.reject(Error("database unavailable")) : transact(...args)) : undefined; const original=db.update.bind(db); const update=vi.spyOn(db,"update").mockImplementation(((table:any)=>{ const query=original(table),set=query.set.bind(query); query.set=((values:any)=>values.providerLeaseId === "task-2" - ? {where:()=>({returning:async()=>{if(mode!=="ownership-lost") throw Error("replacement write failed");return [];}})} + ? {where:()=>({returning:async()=>{replacementFailed=true;if(mode!=="ownership-lost") throw Error("replacement write failed");return [];}})} : set(values)) as never; return query; }) as never); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index cffbde81177..dc0e05e249c 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,3 +1,4 @@ +import { skillVersionRetentionService } from "../services/skill-version-retention.js"; import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js"; import { joinRequestRetentionService } from "../services/join-request-retention.js"; import { approvalRetentionService } from "../services/approval-retention.js"; @@ -538,6 +539,13 @@ vi.mock("../board-claim.js", () => ({ initializeBoardClaimChallenge: vi.fn(async () => undefined), })); +vi.mock("../services/skill-version-retention.js", () => ({ + skillVersionRetentionService: vi.fn(() => ({ + listCompanies: vi.fn(async () => []), pruneCompany: vi.fn(async () => ({ checked: 0, truncated: false, nextRunAt: null })), + dispose: vi.fn(async () => {}), graceMs: 60 * 60_000, + })), +})); + vi.mock("../auth/better-auth.js", () => ({ createBetterAuthHandler: vi.fn(() => undefined), createBetterAuthInstance: createBetterAuthInstanceMock, @@ -574,6 +582,28 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers skill cache retention only when enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ skillVersionRetentionEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(skillVersionRetentionService).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.skillVersionRetention)).toHaveLength(enabled ? 1 : 0); + }); + + it("keeps held skill-cache reconstruction tracked without delaying readiness", async () => { + loadConfigMock.mockReturnValue(buildTestConfig({ skillVersionRetentionEnabled: true })); + const pending = Promise.withResolvers>(); + const list = vi.fn(() => pending.promise); + vi.mocked(skillVersionRetentionService).mockReturnValueOnce({ listCompanies: list, + pruneCompany: vi.fn(), dispose: vi.fn(async () => {}), graceMs: 60 * 60_000 } as never); + try { + await startServer(); await settleStartupHousekeeping(); + expect(list).toHaveBeenCalledOnce(); expect(getStartupRecoveryState().phase).toBe("ready"); + expect(idleWorkSnapshot().active).toBe(1); + } finally { pending.resolve([]); await settleStartupHousekeeping(); } + expect(idleWorkSnapshot().active).toBe(0); + expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.skillVersionRetention)).toEqual([]); + }); + it.each([false, true])("registers response recovery and its shutdown subscription only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); diff --git a/server/src/__tests__/skill-version-cache.test.ts b/server/src/__tests__/skill-version-cache.test.ts index 4eca01f0a7b..438678eb8e6 100644 --- a/server/src/__tests__/skill-version-cache.test.ts +++ b/server/src/__tests__/skill-version-cache.test.ts @@ -1,3 +1,5 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { applyPendingMigrations } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; import os from "node:os"; @@ -17,19 +19,21 @@ async function sweepSkillVersionCache(db: ReturnType, options: return { ...result, evicted: result.removed }; } -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_SKILL_CACHE_TEST_DATABASE_URL"); +const support = external ? { supported: true, reason: undefined } : 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 database: Awaited> | undefined; let db: ReturnType; let home: string; const previousHome = process.env.PAPERCLIP_HOME; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("skill-version-cache-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("skill-version-cache-"); + if (external) await applyPendingMigrations(external); + db = createDb(external ?? database!.connectionString); home = await fs.mkdtemp(path.join(os.tmpdir(), "skill-version-cache-")); process.env.PAPERCLIP_HOME = home; }); @@ -38,6 +42,7 @@ describePostgres("immutable skill version cache", () => { 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 db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); @@ -293,7 +298,7 @@ describePostgres("immutable skill version cache", () => { .where(eq(heartbeatRuns.id, run.id)); }); await ready.promise; - const other = createDb(database.connectionString); + const other = createDb(external ?? database!.connectionString); try { const cleaning = await sweepSkillVersionCache(other, { companyId: f.companyId }); expect(cleaning.evicted).toBe(0); @@ -312,7 +317,7 @@ describePostgres("immutable skill version cache", () => { expect(entries[0]!.sourceStatus).toBe("available"); throw new Error("durable pin write failed"); })).rejects.toThrow("durable pin write failed"); - const other = createDb(database.connectionString); + const other = createDb(external ?? database!.connectionString); try { expect(await withSkillVersionCacheLock(other, f.companyId, async () => "released", true)).toBe("released"); } finally { await other.$client.end({ timeout: 0 }); } @@ -359,7 +364,7 @@ describePostgres("immutable skill version cache", () => { }) as typeof fs.lstat); const sweep = sweepSkillVersionCache(db, { companyId: f.companyId }); await inspecting.promise; - const other = createDb(database.connectionString); + const other = createDb(external ?? database!.connectionString); let timeout: ReturnType | undefined; const preparation = withSkillVersionCacheLock(other, f.companyId, async () => { await companySkillService(other).listRuntimeSkillEntries(f.companyId, { diff --git a/server/src/__tests__/skill-version-retention-work.test.ts b/server/src/__tests__/skill-version-retention-work.test.ts new file mode 100644 index 00000000000..6efd70e4d6d --- /dev/null +++ b/server/src/__tests__/skill-version-retention-work.test.ts @@ -0,0 +1,57 @@ +import { randomUUID } from "node:crypto"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { afterEach, expect, it, vi } from "vitest"; +import { resolvePaperclipInstanceRoot } from "../home-paths.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { noteSkillVersionCacheChange } from "../services/skill-version-retention-notifications.js"; +import { registerSkillVersionRetentionWork } from "../services/skill-version-retention-work.js"; +import type { skillVersionRetentionService } from "../services/skill-version-retention.js"; + +let coordinator: ReturnType | undefined; +let worker: ReturnType | undefined; +let home: string | undefined; +afterEach(async () => { + worker?.unsubscribe(); + await coordinator?.stop(); + await worker?.dispose(); + vi.useRealTimers(); + vi.unstubAllEnvs(); + if (home) rmSync(home, { recursive: true, force: true }); +}); + +it("keeps an expired cache change eligible when its own paged scan finishes", async () => { + home = mkdtempSync(path.join(tmpdir(), "sleep64-retention-work-")); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("PAPERCLIP_INSTANCE_ID", "sleep64-retention-work"); + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-29T12:00:00Z")); + const companyId = randomUUID(); + let page = 0; + const prune = vi.fn(async (_companyId: string, _options: { cursor: unknown }) => { + const truncated = ++page < 4; + return { checked: truncated ? 50 : 0, removed: 0, failed: 0, skipped: {}, + truncated, interrupted: false, nextRunAt: null, cursor: truncated ? { page } : null }; + }); + const service = { graceMs: 2_500, dryRun: false, listCompanies: async () => [{ id: companyId }], + pruneCompany: prune, dispose: async () => {} } as unknown as ReturnType; + const onError = vi.fn(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError }); + worker = registerSkillVersionRetentionWork(coordinator, service, () => true); + await worker.ready; + expect(prune).toHaveBeenCalledOnce(); + noteSkillVersionCacheChange(path.join(resolvePaperclipInstanceRoot(), "skills", companyId, + "__versions__", randomUUID(), randomUUID()), { source: "access" }); + await vi.advanceTimersByTimeAsync(1_000); + await vi.advanceTimersByTimeAsync(1_000); + await vi.advanceTimersByTimeAsync(1_000); + expect(prune).toHaveBeenCalledTimes(5); + expect(prune.mock.calls[4]).toEqual([companyId, expect.objectContaining({ cursor: null })]); + expect(coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(60_000); + expect(prune).toHaveBeenCalledTimes(5); + expect(onError).not.toHaveBeenCalled(); +}); diff --git a/server/src/__tests__/skill-version-retention.test.ts b/server/src/__tests__/skill-version-retention.test.ts index 05d4d0fbfd7..98ce5ccb9af 100644 --- a/server/src/__tests__/skill-version-retention.test.ts +++ b/server/src/__tests__/skill-version-retention.test.ts @@ -1,4 +1,19 @@ -import { touchSkillVersionCache } from "../services/skill-version-cache.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { logger } from "../middleware/logger.js"; +import { prepareAgentHireApproval } from "../services/agent-hiring.js"; +import { notifyRetainedNativeSessionCleanup } from "../services/retained-native-session-cleanup-notifications.js"; +import { setTimeout as realDelay } from "node:timers/promises"; +import { createAgentLifecycle } from "../services/agent-lifecycle.js"; +import { sql } from "drizzle-orm"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerSkillVersionRetentionWork } from "../services/skill-version-retention-work.js"; +import { notifySkillVersionRetention } from "../services/skill-version-retention-notifications.js"; +import { notifyInstructionCollection } from "../services/instruction-collection-notifications.js"; +import { notifyNativeEnvironmentSettlement } from "../services/native-environment-settlement-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { applyPendingMigrations } from "@paperclipai/db"; +import { touchSkillVersionCache, lastUsedAt, skillVersionCachePaths } from "../services/skill-version-cache.js"; import { randomUUID } from "node:crypto"; import os from "node:os"; import path from "node:path"; @@ -24,7 +39,8 @@ import { skillVersionRetentionService, } from "../services/skill-version-retention.ts"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_SKILL_RETENTION_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -54,16 +70,25 @@ describeEmbeddedPostgres("skill version snapshot retention", () => { let paperclipHome: string | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-skill-version-retention-"); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-skill-version-retention-"); + if (external) await applyPendingMigrations(external); 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); + db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }, 20_000); + let coordinator: ReturnType | undefined; + let worker: ReturnType | undefined; + let admitted = true; afterEach(async () => { + worker?.unsubscribe(); + await coordinator?.stop(); + await worker?.dispose(); + worker = undefined; coordinator = undefined; admitted = true; + vi.useRealTimers(); vi.restoreAllMocks(); await db.delete(heartbeatRuns); await db.delete(agents); @@ -82,6 +107,7 @@ describeEmbeddedPostgres("skill version snapshot retention", () => { 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 db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -131,6 +157,459 @@ describeEmbeddedPostgres("skill version snapshot retention", () => { return dir; } + async function startRetention(graceMs = 20_000, service = skillVersionRetentionService(db, { graceMs })) { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select 1`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => true, + onError: error => { throw error; } }); + worker = registerSkillVersionRetentionWork(coordinator, service, () => admitted); + await worker.ready; + return service; + } + async function settled(deadline: number | null = null) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBe(deadline); + }, { timeout: 10_000 }); + } + async function quiet(ms = 3 * HOUR_MS) { + const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"), directories = vi.spyOn(fs, "opendir"); + await vi.advanceTimersByTimeAsync(ms); + expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); expect(directories).not.toHaveBeenCalled(); + reads.mockRestore(); transactions.mockRestore(); directories.mockRestore(); + } + + it("disarms after empty reconstruction without hourly SQL or directory reads", async () => { + await startRetention(); await settled(); await quiet(); + }); + + it("keeps only permanent pins quiet with completed run history", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + await writeSnapshot(companyId, skillId, versionIds[0]!); + await writeSnapshot(companyId, skillId, versionIds[1]!); + const [agent] = await db.insert(agents).values({ companyId, name: "Finished", adapterType: "codex_local" }).returning(); + await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, invocationSource: "on_demand", status: "succeeded", + finishedAt: new Date(), contextSnapshot: { skillVersions: { old: versionIds[0] } } }); + await startRetention(); await settled(); + expect(await exists(path.join(resolveCompanySkillVersionsRoot(companyId), skillId, versionIds[0]!))).toBe(false); + expect(await exists(path.join(resolveCompanySkillVersionsRoot(companyId), skillId, versionIds[1]!))).toBe(true); + await quiet(); + }); + + it("wakes ordinary materialization and re-use at the saved access grace deadline", async () => { + const companyId = await createCompany(); + const { skillId, key, versionIds } = await createSkill(companyId, 2); + await startRetention(); await settled(); await quiet(); + const [entry] = await companySkillService(db).listRuntimeSkillEntries(companyId, { + onlyKeys: new Set([key]), versionSelections: new Map([[key, versionIds[0]!]]) }); + expect(entry!.sourceStatus).toBe("available"); + const marker = path.join(path.dirname(path.dirname(resolveCompanySkillVersionsRoot(companyId))), companyId, + "__version_access__", skillId, `${versionIds[0]}.json`); + const accessed = JSON.parse(await fs.readFile(marker, "utf8")).lastUsedAt as number; + await settled(accessed + 20_000); + await vi.advanceTimersByTimeAsync(5000); + await touchSkillVersionCache(companyId, skillId, versionIds[0]!); + const reused = JSON.parse(await fs.readFile(marker, "utf8")).lastUsedAt as number; + await settled(reused + 20_000); + await vi.advanceTimersByTimeAsync(reused + 20_000 - Date.now()); + await settled(); expect(await exists(entry!.source)).toBe(false); await quiet(); + }); + + it("releases current-version protection only after the real version transaction commits", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 1); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + await startRetention(); await settled(); await quiet(); + const inventory = [{ path: "SKILL.md", kind: "skill" as const, content: "---\nname: next\n---\n# Next" }]; + await expect(db.transaction(async tx => { + await companySkillService(db).createVersion(companyId, skillId, {}, null, + { database: tx, fileInventory: inventory, skipInventoryRefresh: true }); + await vi.advanceTimersByTimeAsync(60_000); expect(await exists(directory)).toBe(true); + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await settled(); await quiet(); expect(await exists(directory)).toBe(true); + await db.transaction(async tx => { + await companySkillService(db).createVersion(companyId, skillId, {}, null, + { database: tx, fileInventory: inventory, skipInventoryRefresh: true }); + await vi.advanceTimersByTimeAsync(60_000); expect(await exists(directory)).toBe(true); + }); + await settled(); expect(await exists(directory)).toBe(false); await quiet(); + }); + + it.each([notifyInstructionCollection, notifyNativeEnvironmentSettlement, notifyRetainedNativeSessionCleanup])("releases a captured run pin on its existing settlement notification %$", async notify => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + const [agent] = await db.insert(agents).values({ companyId, name: "Running", adapterType: "codex_local" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, invocationSource: "on_demand", status: "running", + contextSnapshot: { skillVersions: { selected: versionIds[0] } } }).returning(); + await startRetention(); await settled(); await quiet(); expect(await exists(directory)).toBe(true); + await db.transaction(async tx => { + await notify(tx, { companyId, runId: run!.id }); + await tx.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run!.id)); + }); + await settled(); expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("retains the first cache change during drain and resumes without passive inspection", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + await startRetention(); await settled(); admitted = false; + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + await quiet(60_000); expect(await exists(directory)).toBe(true); + admitted = true; await settled(); expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("retains changes arriving during a page", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + const service = skillVersionRetentionService(db, { graceMs: 0 }); + const prune = service.pruneCompany; + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const began = vi.fn(); + service.pruneCompany = async (...args) => { const result = await prune(...args); began(); await held; return result; }; + const starting = startRetention(0, service); + await vi.waitFor(() => expect(began).toHaveBeenCalledOnce()); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + release(); await starting; await settled(); expect(await exists(directory)).toBe(false); + await quiet(); + }); + + it.each(["configuration", "approval"])("wakes when agent %s removes a permanent pin", async mode => { + const companyId = await createCompany(); + const { skillId, key, versionIds } = await createSkill(companyId, 2); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + const [agent] = await db.insert(agents).values({ companyId, name: "Pinned", adapterType: "codex_local", + adapterConfig: { paperclipSkillSync: { desiredSkills: [{ key, versionId: versionIds[0] }] } } }).returning(); + await startRetention(); await settled(); await quiet(); expect(await exists(directory)).toBe(true); + const patch = { adapterConfig: { paperclipSkillSync: { desiredSkills: [] } } }; + if (mode === "configuration") await createAgentLifecycle(db).updateConfiguration(agent!.id, patch); + else await db.transaction(async tx => { + const approved = await prepareAgentHireApproval(tx as unknown as typeof db, agent!, patch); + await tx.update(agents).set(approved).where(eq(agents.id, agent!.id)); + }); + await settled(); expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("reconstructs saved staging grace on restart and stays quiet after removal", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 1); + const staging = await writeSnapshot(companyId, skillId, `.${versionIds[0]}-abc123`, 0); + const deadline = (await fs.stat(staging)).mtimeMs + 20_000; + await startRetention(); await settled(deadline); + worker!.unsubscribe(); await coordinator!.stop(); await worker!.dispose(); + vi.useRealTimers(); + await startRetention(); await settled(deadline); + await vi.advanceTimersByTimeAsync(deadline - Date.now()); await settled(); + expect(await exists(staging)).toBe(false); await quiet(); + }); + + it("uses the saved terminal controller deadline and preserves the database clock fence", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + const [agent] = await db.insert(agents).values({ companyId, name: "Lease", adapterType: "codex_local" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "succeeded", finishedAt: new Date(), + contextSnapshot: { skillVersions: { selected: versionIds[0] } }, + controllerLeaseExpiresAt: sql`date_trunc('milliseconds', clock_timestamp()) + interval '3 seconds'` }).returning(); + const deadline = run!.controllerLeaseExpiresAt!.getTime(); + const service = skillVersionRetentionService(db, { graceMs: 0 }); + expect((await service.pruneCompany(companyId)).nextRunAt).toBe(deadline); + await service.dispose(); + await startRetention(0); await settled(deadline); + await vi.advanceTimersByTimeAsync(deadline - Date.now()); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeGreaterThan(deadline)); + expect(await exists(directory)).toBe(true); + await realDelay(3100); + await vi.advanceTimersByTimeAsync(2000); await settled(); + expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("retains only a concrete failed company for retry while healthy companies settle", async () => { + const companyId = await createCompany(), healthyId = await createCompany(); + const failedSkill = await createSkill(companyId, 2), healthySkill = await createSkill(healthyId, 2); + const failedDir = await writeSnapshot(companyId, failedSkill.skillId, failedSkill.versionIds[0]!); + const healthyDir = await writeSnapshot(healthyId, healthySkill.skillId, healthySkill.versionIds[0]!); + const service = skillVersionRetentionService(db, { graceMs: 0 }), prune = service.pruneCompany; + let failed = true; + service.pruneCompany = vi.fn(async (...args) => { + if (args[0] === companyId && failed) throw new Error("inspection unavailable"); + return prune(...args); + }); + await startRetention(0, service); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).not.toBeNull()); + const retryAt = coordinator!.nextWakeAt()!; + expect(await exists(healthyDir)).toBe(false); expect(await exists(failedDir)).toBe(true); + await quiet(5000); expect(service.pruneCompany).toHaveBeenCalledTimes(2); + failed = false; await vi.advanceTimersByTimeAsync(retryAt - Date.now()); await settled(); + expect(await exists(failedDir)).toBe(false); await quiet(); + }); + + it("continues directory pages without evicting another company's iterator", async () => { + const ids: string[] = []; + for (let index = 0; index < 18; index++) { + const companyId = await createCompany(); ids.push(companyId); + const root = path.join(resolveCompanySkillVersionsRoot(companyId), randomUUID()); + await fs.mkdir(root, { recursive: true }); + for (let entry = 0; entry < 55; entry++) await fs.mkdir(path.join(root, `operator-${entry}`)); + } + const opened = vi.spyOn(fs, "opendir"); + await startRetention(); await settled(); + for (const id of ids) expect(opened.mock.calls.filter(([directory]) => directory === resolveCompanySkillVersionsRoot(id))).toHaveLength(1); + opened.mockRestore(); await quiet(); + }); + + it("wakes ordinary skill deletion and removes its unpinned orphan caches", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 1); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + await startRetention(); await settled(); await quiet(); + await companySkillService(db).deleteSkill(companyId, skillId); + await settled(); expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("observes abandoned staging after a failed materialization flight", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 1); + await startRetention(); await settled(); await quiet(); + let staging = ""; + await expect(runVersionSnapshotFlight(path.join(resolveCompanySkillVersionsRoot(companyId), skillId, versionIds[0]!), async () => { + staging = await writeSnapshot(companyId, skillId, `.${versionIds[0]}-abc123`, 0); + throw new Error("interrupted publication"); + })).rejects.toThrow("interrupted publication"); + const deadline = (await fs.stat(staging)).mtimeMs + 20_000; + await settled(deadline); await vi.advanceTimersByTimeAsync(deadline - Date.now()); await settled(); + expect(await exists(staging)).toBe(false); await quiet(); + }); + + it("retries incomplete pins only for the affected company and disarms after terminal settlement", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + const [agent] = await db.insert(agents).values({ companyId, name: "Unknown pins", adapterType: "codex_local" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running", contextSnapshot: null }).returning(); + await startRetention(); expect(await exists(directory)).toBe(true); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now()); await quiet(5000); + await db.transaction(async tx => { + await notifyInstructionCollection(tx, { companyId, runId: run!.id }); + await tx.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run!.id)); + }); + await settled(); expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("reconstructs beyond one company page without repeating settled companies", async () => { + for (let index = 0; index < 55; index++) await createCompany(); + const service = skillVersionRetentionService(db); + const listed = vi.spyOn(service, "listCompanies"), pruned = vi.spyOn(service, "pruneCompany"); + await startRetention(20_000, service); await settled(); + expect(listed).toHaveBeenCalledTimes(2); expect(pruned).toHaveBeenCalledTimes(55); + expect(new Set(pruned.mock.calls.map(([id]) => id)).size).toBe(55); await quiet(); + }); + + it.each(["inspection", "rename"])("restarts a company page interrupted during %s so cleanup is not forgotten", async stage => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + for (let index = 0; index < 60; index++) await fs.mkdir(path.join(path.dirname(directory), `operator-${index}`)); + let paused = false; + const list = fs.readdir, rename = fs.rename; + const reading = vi.spyOn(fs, "readdir").mockImplementation(async (...args) => { + const entries = await list(...args); + if (stage === "inspection" && !paused && args[0] === directory) { paused = true; admitted = false; } + return entries; + }); + const renaming = vi.spyOn(fs, "rename").mockImplementation(async (...args) => { + await rename(...args); + if (stage === "rename" && !paused && args[0] === directory) { paused = true; admitted = false; } + }); + await startRetention(0); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(paused).toBe(true); }); + reading.mockRestore(); renaming.mockRestore(); admitted = true; + await settled(); expect(await exists(directory)).toBe(false); + expect((await fs.readdir(path.dirname(directory))).filter(name => name.startsWith(".trash-"))).toEqual([]); + await quiet(); + }); + + it("reports dry-run candidates and bytes while keeping the cache intact", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + const reported = vi.spyOn(logger, "info"); + await startRetention(0, skillVersionRetentionService(db, { dryRun: true, graceMs: 0 })); await settled(); + expect(reported.mock.calls).toEqual(expect.arrayContaining([[expect.objectContaining({ companyId, + candidates: 1, candidateBytes: expect.any(Number), dryRun: true }), "skill version snapshot retention"]])); + const report = reported.mock.calls.find(([value]) => value && typeof value === "object" && "candidateBytes" in value)![0] as { candidateBytes: number }; + expect(report.candidateBytes).toBeGreaterThan(0); expect(await exists(directory)).toBe(true); await quiet(); + }); + + it("wakes the missing-local-source reconciliation deletion path", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 1); + const source = path.join(paperclipHome!, "local-source"); + await fs.mkdir(source, { recursive: true }); await fs.writeFile(path.join(source, "SKILL.md"), "# Local source"); + await db.update(companySkills).set({ sourceType: "local_path", sourceLocator: source }).where(eq(companySkills.id, skillId)); + const service = companySkillService(db); await service.list(companyId); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!); + await startRetention(); await settled(); await quiet(); + await fs.rm(source, { recursive: true, force: true }); await service.list(companyId); + expect(await db.select().from(companySkills).where(eq(companySkills.id, skillId))).toEqual([]); + await settled(); expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("coalesces repeated run settlement into one fresh company scan per thirty seconds", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 1); + await writeSnapshot(companyId, skillId, versionIds[0]!); + const service = skillVersionRetentionService(db), pruned = vi.spyOn(service, "pruneCompany"); + await startRetention(20_000, service); await settled(); + for (let index = 0; index < 10; index++) { + await db.transaction(tx => notifyInstructionCollection(tx, { companyId })); + await vi.advanceTimersByTimeAsync(1000); + } + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(pruned).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(30_000); await settled(); expect(pruned).toHaveBeenCalledTimes(2); await quiet(); + }); + + it.each(["empty", "uncertain"] as const)("bounds fresh $0-company scans across repeated scoped settlement", async state => { + const companyId = await createCompany(), otherId = await createCompany(); + let runId: string | undefined; + if (state === "uncertain") { + const { skillId, versionIds } = await createSkill(companyId, 2); + await writeSnapshot(companyId, skillId, versionIds[0]!); + const [agent] = await db.insert(agents).values({ companyId, name: "Unknown pins", adapterType: "codex_local" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running", contextSnapshot: null }).returning(); + runId = run!.id; + } + const service = skillVersionRetentionService(db), pruned = vi.spyOn(service, "pruneCompany"); + const discovered = vi.spyOn(service, "listCompanies"); + await startRetention(20_000, service); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + for (let index = 0; index < 10; index++) { + await db.transaction(tx => notifyNativeEnvironmentSettlement(tx, { companyId })); + await vi.advanceTimersByTimeAsync(1000); + } + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(discovered).toHaveBeenCalledOnce(); + expect(pruned.mock.calls.filter(([id]) => id === companyId)).toHaveLength(1); + expect(pruned.mock.calls.filter(([id]) => id === otherId)).toHaveLength(1); + if (runId) await db.transaction(async tx => { + await notifyInstructionCollection(tx, { companyId, runId }); + await tx.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId!)); + }); + await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect(pruned.mock.calls.filter(([id]) => id === companyId)).toHaveLength(2); + expect(pruned.mock.calls.filter(([id]) => id === otherId)).toHaveLength(1); + await quiet(); + }); + + it("does not rediscover tenant caches after a runless environment notification", async () => { + const companyId = await createCompany(), otherId = await createCompany(); + const service = skillVersionRetentionService(db), pruned = vi.spyOn(service, "pruneCompany"); + const discovered = vi.spyOn(service, "listCompanies"); + await startRetention(20_000, service); await settled(); + for (let index = 0; index < 10; index++) { + await db.transaction(tx => notifyNativeEnvironmentSettlement(tx)); + await vi.advanceTimersByTimeAsync(1000); + } + await settled(); + expect(discovered).toHaveBeenCalledOnce(); + expect(pruned.mock.calls.map(([id]) => id).sort()).toEqual([companyId, otherId].sort()); + await quiet(); + }); + + it("coalesces repeated instance reconstruction without restarting global discovery", async () => { + await createCompany(); await createCompany(); + const service = skillVersionRetentionService(db), discovered = vi.spyOn(service, "listCompanies"); + await startRetention(20_000, service); await settled(); + for (let index = 0; index < 10; index++) { + await instanceSettingsService(db).updateExperimental({ enableWorktreeRunExecution: index % 2 === 0 }); + await vi.advanceTimersByTimeAsync(1000); + } + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(discovered).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect(discovered).toHaveBeenCalledTimes(2); await quiet(); + }); + + it("retains a new cache grace deadline that arrives after a company scan read", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + await writeSnapshot(companyId, skillId, versionIds[1]!); + const service = skillVersionRetentionService(db, { graceMs: 5000 }); + const prune = service.pruneCompany; + const began = Promise.withResolvers(), release = Promise.withResolvers(); + let held = false; + service.pruneCompany = async (...args) => { + const result = await prune(...args); + if (held) { began.resolve(); await release.promise; } + return result; + }; + await startRetention(5000, service); await settled(); held = true; + await db.transaction(tx => notifyInstructionCollection(tx, { companyId })); + const advancing = vi.advanceTimersByTimeAsync(30_000); + let directory = "", deadline = 0; + try { + await began.promise; + directory = await writeSnapshot(companyId, skillId, versionIds[0]!, 0); + deadline = (await lastUsedAt(skillVersionCachePaths(companyId, skillId, versionIds[0]!).access))! + 5000; + } finally { held = false; release.resolve(); } + await advancing; await settled(deadline); + await vi.advanceTimersByTimeAsync(deadline - Date.now()); await settled(); + expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("lets an actual five-second grace deadline preempt the fresh scan limit", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 2); + const directory = await writeSnapshot(companyId, skillId, versionIds[0]!, 0); + const deadline = (await lastUsedAt(skillVersionCachePaths(companyId, skillId, versionIds[0]!).access))! + 5000; + await startRetention(5000); await settled(deadline); + await db.transaction(tx => notifyInstructionCollection(tx, { companyId })); + await vi.advanceTimersByTimeAsync(deadline - Date.now()); await settled(); + expect(await exists(directory)).toBe(false); await quiet(); + }); + + it("keeps directory pages moving while another company's grace deadline is due", async () => { + const dueCompany = await createCompany(), pagedCompany = await createCompany(); + const dueSkill = await createSkill(dueCompany, 2), pagedSkill = await createSkill(pagedCompany, 1); + const expiring = await writeSnapshot(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!, 0); + const parent = path.join(resolveCompanySkillVersionsRoot(pagedCompany), pagedSkill.skillId); + await fs.mkdir(parent, { recursive: true }); + for (let index = 0; index < 410; index++) await fs.mkdir(path.join(parent, `operator-${index}`)); + const service = skillVersionRetentionService(db, { graceMs: 5000 }); + vi.spyOn(service, "listCompanies").mockResolvedValue([{ id: dueCompany }, { id: pagedCompany }]); + const prune = service.pruneCompany, pageTimes: number[] = []; + service.pruneCompany = async (id, options) => { + if (id === pagedCompany) pageTimes.push(Date.now()); + return prune(id, options); + }; + await startRetention(5000, service); await settled(); + expect(pageTimes.length).toBeGreaterThan(5); + expect(Math.max(...pageTimes.slice(1).map((at, index) => at - pageTimes[index]!))).toBeLessThan(3000); + expect(await exists(expiring)).toBe(false); await quiet(); + }); + + it("waits for an admitted page before disposing its scan on shutdown", async () => { + const companyId = await createCompany(); + const { skillId, versionIds } = await createSkill(companyId, 1); + await writeSnapshot(companyId, skillId, versionIds[0]!); + const service = skillVersionRetentionService(db), prune = service.pruneCompany; + const release = Promise.withResolvers(), began = Promise.withResolvers(); + service.pruneCompany = async (...args) => { began.resolve(); await release.promise; return prune(...args); }; + const disposed = vi.spyOn(service, "dispose"); + const starting = startRetention(20_000, service); await began.promise; + worker!.unsubscribe(); + const stopping = coordinator!.stop().then(() => worker!.dispose()); + await realDelay(30); expect(disposed).not.toHaveBeenCalled(); + release.resolve(); await starting; await stopping; + expect(disposed).toHaveBeenCalledOnce(); expect(coordinator!.nextWakeAt()).toBeNull(); + }); + 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); diff --git a/server/src/index.ts b/server/src/index.ts index a6978f577e5..76ec3cf072b 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerSkillVersionRetentionWork } from "./services/skill-version-retention-work.js"; import { registerDatabaseBackupWork } from "./services/database-backup-work.js"; import { registerToolConnectionHealthWork } from "./services/tool-connection-health-work.js"; import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "./services/approval-queue-seeding-work.js"; @@ -24,8 +25,6 @@ import { registerLoginCleanupWork } from "./services/login-cleanup-work.js"; 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 { @@ -1259,39 +1258,6 @@ async function startServerWithDatabaseTeardown( heartbeatSchedulerInterval?.unref?.(); }; const skillSourceSync = skillSourceSyncService(db as any); - // 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 nativeHarnessBackupExpiration = nativeHarnessBackupExpirationService(db as any, { dryRun: config.nativeHarnessBackupExpirationDryRun, closedForDays: config.nativeHarnessBackupExpirationDays, @@ -1378,6 +1344,14 @@ async function startServerWithDatabaseTeardown( server.on("close", () => { void taskDrainExpiryWork.stop(); }); const executionSchedule = executionControlSchedule(db); const workspaceRecoveryRuntime = environmentRuntimeService(db, { pluginWorkerManager }); + const skillVersionRetentionWorker = config.skillVersionRetentionEnabled + ? registerSkillVersionRetentionWork(deliveryWork, skillVersionRetentionService(db as any, { + dryRun: config.skillVersionRetentionDryRun, + keepNewest: config.skillVersionRetentionKeepNewest, + graceMs: config.skillVersionRetentionGraceMinutes * 60 * 1000, + }), () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby()) + : null; + if (skillVersionRetentionWorker) trackHeartbeatSchedulerWork(skillVersionRetentionWorker.ready); const workspaceRecoveryWork = registerWorkspaceRecoveryWork(db, deliveryWork, workspaceRecoveryRuntime, accountingHeartbeat.resumeRemoteStopComments, () => !heartbeatSchedulerStopped); registerExecutionControlWork(deliveryWork, [ @@ -2212,21 +2186,19 @@ async function startServerWithDatabaseTeardown( ) => { await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]); heartbeatSchedulerStopped = true; + skillVersionRetentionWorker?.unsubscribe(); await taskDrainExpiryWork.stop(); - clearInterval(skillVersionRetentionInterval); - clearTimeout(skillVersionRetentionStartup); - clearTimeout(skillVersionRetentionContinuation); clearInterval(nativeHarnessBackupExpirationInterval); clearTimeout(nativeHarnessBackupExpirationStartup); clearTimeout(nativeHarnessBackupExpirationContinuation); unsubscribeNativeRunnerEvidence(); - await skillVersionRetention.dispose(); await skillSourceSync.stop(); unsubscribeChatCompletions(); unsubscribeNativeResponsePresentation?.(); unsubscribeNativeWorkspaceFinalization?.(); unsubscribeRetainedNativeSessionCleanup?.(); await deliveryWork.stop(); + await skillVersionRetentionWorker?.dispose(); await workspaceRecoveryWork.stop(); unregisterTerminalWorkspaceProbe?.(); unsubscribeNativeWorkspaceSyncRetention?.(); diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index 16fae652c95..4139dc6dfe0 100644 --- a/server/src/services/agent-configuration.ts +++ b/server/src/services/agent-configuration.ts @@ -1,3 +1,4 @@ +import { notifySkillVersionPinChange } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { and, eq } from "drizzle-orm"; @@ -85,6 +86,9 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) { await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); } + if (normalizedPatch.adapterConfig !== undefined) { + await notifySkillVersionPinChange(db, existing.companyId, existing.adapterConfig ?? {}, normalizedPatch.adapterConfig); + } return normalizedPatch; } diff --git a/server/src/services/agent-hiring.ts b/server/src/services/agent-hiring.ts index 8a08fa6052e..a150937a212 100644 --- a/server/src/services/agent-hiring.ts +++ b/server/src/services/agent-hiring.ts @@ -1,3 +1,4 @@ +import { notifySkillVersionRetention, notifySkillVersionPinChange } from "./skill-version-retention-notifications.js"; import { agentAppearanceSchema, randomAgentAppearance } from "@paperclipai/shared"; import { normalizePaperclipRunnerAdapterConfig } from "@paperclipai/adapter-utils/server-utils"; import { agents, toolConnectionInstalls, principalPermissionGrants, type Db } from "@paperclipai/db"; @@ -53,6 +54,7 @@ export async function prepareAgentHire(db: Db, companyId: string, input: CreateA childAdapterConfig: adapterConfig, claudeLogin: options?.claudeLogin, decision: assertClaudeOAuthBindingInvariant({ adapterType, nextConfig: adapterConfig }), }); + if (options?.setupAgentId) await notifySkillVersionRetention(db, companyId); return { ...data, name: uniqueName, appearance: data.appearance == null ? randomAgentAppearance() : agentAppearanceSchema.parse(data.appearance), role, adapterType, adapterConfig, permissions: normalizedPermissions, runtimeConfig }; @@ -128,6 +130,7 @@ export async function prepareAgentHireApproval(txDb: Db, existing: AgentHireReco nextConfig: patch.adapterConfig, priorConfig: existing.adapterConfig }), }); } + if (patch.adapterConfig !== undefined) await notifySkillVersionPinChange(txDb, existing.companyId, existing.adapterConfig ?? {}, patch.adapterConfig); return patch; } diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index 4a5925c484b..b43223f69c6 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -1,3 +1,4 @@ +import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { ensureSkillCacheDirectory, touchSkillVersionCache, withSkillVersionCacheLock } from "./skill-version-cache.js"; import { measureRunPhase } from "./execution-preparation-timing.js"; @@ -3229,7 +3230,10 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo await removeRuntimeSkillCache(resolveManagedSkillsRoot(companyId), skill.id, async () => { await fs.rm(resolveRuntimeSkillMaterializedPath(companyId, skill), { recursive: true, force: true }); - await db.delete(companySkills).where(eq(companySkills.id, skill.id)); + await db.transaction(async tx => { + await notifySkillVersionRetention(tx, companyId); + await tx.delete(companySkills).where(eq(companySkills.id, skill.id)); + }); }); } } @@ -3694,6 +3698,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo ); const previousVersion = options.fileInventory === undefined ? await getCurrentVersion(skill) : null; const persist = async (tx: DbOrTransaction) => { + await notifySkillVersionRetention(tx, companyId); await tx.execute(sql` select ${companySkills.id} from ${companySkills} @@ -4003,6 +4008,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo async function cleanupFailedFork(companyId: string, sourceSkillId: string, forkSkillId: string, forkDir: string) { await db.transaction(async (tx) => { + await notifySkillVersionRetention(tx, companyId); await tx .update(companySkills) .set({ currentVersionId: null, updatedAt: new Date() }) @@ -7393,6 +7399,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo } try { return await db.transaction(async tx => { + if (initial.length) await notifySkillVersionRetention(tx, companyId); await beforeDelete?.(tx); const deleted: CompanySkill[] = []; for (const before of initial) { diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index 392bb7fe095..306f340d9fc 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -9,6 +9,7 @@ export function registerDeadlineWork( next: () => Promise; canRun: () => boolean; retryMs: number; + onSettled?: (hints?: ReadonlySet) => void; continuationMs?: number; // Optional batching latency for notifications; already-known deadlines still preempt it. notificationCoalesceMs?: number; @@ -23,7 +24,7 @@ export function registerDeadlineWork( const worker = coordinator.register(queue, { retryMs: task.retryMs, pausePassiveMaintenance: true, - onSettled: () => { requested++; }, + onSettled: hints => { requested++; task.onSettled?.(hints); }, run: async signal => { if (!task.canRun()) { requested++; next = null; return; } if (Date.now() < earliest) { next = earliest; return; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 39f2d1d762f..ef031e443ef 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + skillVersionRetention: "skill-version-retention", databaseBackups: "database-backups", importSpoolCleanup: "import-spool-cleanup", externalConversationSettlement: "external-conversation-settlement", diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index a1a69f922ae..728f05b9303 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -8,7 +8,7 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Nati await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); } -export function writeEnvironmentCleanup(db: Db, write: (tx: Db) => PromiseLike, scope?: NativeEnvironmentSettlementScope): Promise { +export function writeEnvironmentCleanup(db: Db, write: (tx: Db) => PromiseLike, scope: NativeEnvironmentSettlementScope): Promise { return db.transaction(async tx => { await notifyEnvironmentCleanup(tx, scope); return write(tx as unknown as Db); diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index 8806bfca452..abc9b6b9fd8 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -346,10 +346,10 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage for(const owner of owners) { const receipt=remoteTerminationReceipt(owner,{providerLeaseId:taskId,state:"stopped"}); if(!receipt) continue; - await db.update(environmentLeases).set({ + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ ...(owner.status === "retained" ? {status:"released" as const,releasedAt:new Date(),cleanupStatus:"success" as const}:{}), metadata:sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({remoteExecutionTermination:receipt,runnerTaskExecutionStopped:true})}::jsonb`, - }).where(eq(environmentLeases.id,owner.id)); + }).where(eq(environmentLeases.id,owner.id)), { companyId: owner.companyId, runId: owner.heartbeatRunId ?? undefined }); } } await db.update(environmentLeases).set({ diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index f7c40d13617..0208653a86f 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js"; import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; @@ -633,6 +634,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyExternalAgentSettlement(tx); await notifyInstructionCommit(tx); await notifyDirectoryCleanup(tx); + await notifySkillVersionRetentionReconstruction(tx); await notifyInstructionCollection(tx); await notifyCancelledWorkspacePreservation(tx); await notifyExternalConversationSettlement(tx); diff --git a/server/src/services/instruction-collection-notifications.ts b/server/src/services/instruction-collection-notifications.ts index 7ce151cad52..d0c19355c00 100644 --- a/server/src/services/instruction-collection-notifications.ts +++ b/server/src/services/instruction-collection-notifications.ts @@ -1,7 +1,9 @@ +import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type InstructionCollectionScope = { companyId: string; runId?: string }; export async function notifyInstructionCollection(transaction: object, scope?: InstructionCollectionScope) { + if (scope) await notifySkillVersionRetention(transaction, scope.companyId); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.instructionCollection, scope ? JSON.stringify(scope) : "*"); } diff --git a/server/src/services/native-environment-settlement-notifications.ts b/server/src/services/native-environment-settlement-notifications.ts index f8dc0a966c4..5769b81015a 100644 --- a/server/src/services/native-environment-settlement-notifications.ts +++ b/server/src/services/native-environment-settlement-notifications.ts @@ -1,7 +1,9 @@ +import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string }; -export function notifyNativeEnvironmentSettlement(tx: object, scope?: NativeEnvironmentSettlementScope) { - return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeEnvironmentSettlement, +export async function notifyNativeEnvironmentSettlement(tx: object, scope?: NativeEnvironmentSettlementScope) { + if (scope) await notifySkillVersionRetention(tx, scope.companyId); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeEnvironmentSettlement, scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"); } diff --git a/server/src/services/retained-native-session-cleanup-notifications.ts b/server/src/services/retained-native-session-cleanup-notifications.ts index a359e8ad2b2..1ff39143e0e 100644 --- a/server/src/services/retained-native-session-cleanup-notifications.ts +++ b/server/src/services/retained-native-session-cleanup-notifications.ts @@ -1,3 +1,4 @@ +import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type RetainedNativeSessionCleanupScope = { @@ -7,8 +8,9 @@ export type RetainedNativeSessionCleanupScope = { agentId?: string; }; -export function notifyRetainedNativeSessionCleanup(tx: object, scope: RetainedNativeSessionCleanupScope) { - return notifyDeliveryWork(tx, DELIVERY_QUEUES.retainedNativeSessionCleanup, JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId, agentId: scope.agentId })); +export async function notifyRetainedNativeSessionCleanup(tx: object, scope: RetainedNativeSessionCleanupScope) { + await notifySkillVersionRetention(tx, scope.companyId); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.retainedNativeSessionCleanup, JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId, agentId: scope.agentId })); } diff --git a/server/src/services/skill-version-cache.ts b/server/src/services/skill-version-cache.ts index 57787129e80..9b243607806 100644 --- a/server/src/services/skill-version-cache.ts +++ b/server/src/services/skill-version-cache.ts @@ -9,6 +9,7 @@ import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { assertSkillSnapshotPath } from "./skill-snapshot.js"; import { repositorySkillFrontmatter } from "./skill-repository-package.js"; import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared"; +import { noteSkillVersionCacheChange } from "./skill-version-retention-notifications.js"; import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js"; import { beginRunPhase } from "./execution-preparation-timing.js"; @@ -70,12 +71,13 @@ export async function ensureSkillCacheDirectory(directory: string, create = fals } export async function touchSkillVersionCache(companyId: string, skillId: string, versionId: string, now = Date.now()) { - const { access } = skillVersionCachePaths(companyId, skillId, versionId); + const { access, directory } = 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); + noteSkillVersionCacheChange(directory, { source: "access", occurredAt: now }); } finally { await fs.rm(temporary, { force: true }); } } @@ -97,9 +99,12 @@ export async function lastUsedAt(filename: string): Promise { // 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> { +export async function protectedVersions(db: Db, companyId: string): Promise<{ versions: Set | null; leaseDeadlines: Map }> { const rows = await db.select({ id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, nativeIssueId: heartbeatRuns.nativeIssueId, + leaseDeadline: sql`case when ${heartbeatRuns.status} in ('succeeded', 'failed', 'timed_out', 'cancelled', 'interrupted') + and ${heartbeatRuns.finishedAt} is not null and ${heartbeatRuns.controllerLeaseExpiresAt} > now() + then extract(epoch from ${heartbeatRuns.controllerLeaseExpiresAt}) * 1000 end`, pins: sql`${heartbeatRuns.contextSnapshot}->'skillVersions'`, // Avoid loading large execution inputs for runs with the modern pin map. legacyInput: sql`case when ${heartbeatRuns.runtimeMode} = 'native' @@ -121,7 +126,13 @@ export async function protectedVersions(db: Db, companyId: string): Promise(); - for (const { pins, legacyInput, id, agentId, nativeIssueId } of rows) { + const leaseDeadlines = new Map(); + const unknown = () => ({ versions: null, leaseDeadlines }); + for (const { pins, legacyInput, id, agentId, nativeIssueId, leaseDeadline } of rows) { + const add = (version: string) => { + versions.add(version); + if (leaseDeadline !== null) leaseDeadlines.set(version, Math.min(leaseDeadlines.get(version) ?? Infinity, Number(leaseDeadline))); + }; if (legacyInput != null) { // Older native runs captured the complete selection in their immutable // execution input before contextSnapshot.skillVersions was introduced. @@ -131,25 +142,25 @@ export async function protectedVersions(db: Db, companyId: string): Promise void>(); +export function onSkillVersionCacheChange(listener: (change: CacheChange) => void) { + listeners.add(listener); + return () => { listeners.delete(listener); }; +} + +export function noteSkillVersionCacheChange(directory: string, change: { occurredAt?: number; source?: "access" | "publication" } = {}) { + const root = path.resolve(resolvePaperclipInstanceRoot(), "skills"); + const [companyId, versions] = path.relative(root, directory).split(path.sep); + if (versions !== "__versions__" || !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(companyId)) return; + const suppliedAt = change.occurredAt ?? Date.now(); + const occurredAt = Number.isSafeInteger(suppliedAt) && suppliedAt > 0 ? suppliedAt : Date.now(); + for (const listener of listeners) listener({ root, companyId, directory: path.resolve(directory), occurredAt, source: change.source ?? "publication" }); +} + +export async function notifySkillVersionPinChange(tx: object, companyId: string, before: Record, after: Record) { + if (JSON.stringify(readPaperclipSkillSyncPreference(before).desiredSkillEntries) + !== JSON.stringify(readPaperclipSkillSyncPreference(after).desiredSkillEntries)) { + await notifySkillVersionRetention(tx, companyId); + } +} diff --git a/server/src/services/skill-version-retention-work.ts b/server/src/services/skill-version-retention-work.ts new file mode 100644 index 00000000000..cbea75ffcf4 --- /dev/null +++ b/server/src/services/skill-version-retention-work.ts @@ -0,0 +1,150 @@ +import path from "node:path"; +import { resolvePaperclipInstanceRoot } from "../home-paths.js"; +import { logger } from "../middleware/logger.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { onSkillVersionCacheChange } from "./skill-version-retention-notifications.js"; +import type { skillVersionRetentionService, SkillVersionPruneResult } from "./skill-version-retention.js"; + +type RetentionService = ReturnType; +type ActiveCompany = { id: string; cursor: SkillVersionPruneResult["cursor"]; nextRunAt: number | null; rescanOwed: boolean }; + +export function registerSkillVersionRetentionWork( + coordinator: ReturnType, + service: RetentionService, + canRun: () => boolean, +) { + const root = path.resolve(resolvePaperclipInstanceRoot(), "skills"); + const dirty = new Set(); + const deadlines = new Map(); + const failures = new Map(); + const freshScanAfter = new Map(); + const cacheChanges = new Map>(); + const cacheDeadline = (id: string) => { + let next = Infinity; + for (const at of cacheChanges.get(id)?.values() ?? []) next = Math.min(next, at); + return next; + }; + let discoveryAfter = 0, discoveryLoading = false, discoveryAgain = false; + let discovery: { after?: string; retryAt: number } | null = { retryAt: 0 }; + let active: ActiveCompany | null = null; + let continuationAt = 0; + const readyAt = (id: string) => Math.max(failures.get(id) ?? 0, + Math.min(deadlines.get(id) ?? Infinity, cacheDeadline(id)) <= Date.now() ? 0 : freshScanAfter.get(id) ?? 0); + const changed = (hints?: ReadonlySet) => { + for (const hint of hints ?? ["*"]) { + if (hint === "*") { + if (!discovery) discovery = { retryAt: discoveryAfter }; + else if (discoveryLoading || discovery.after !== undefined) discoveryAgain = true; + } else dirty.add(hint); + } + }; + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.skillVersionRetention, { + canRun, + retryMs: 30_000, + notificationCoalesceMs: 1000, + onSettled: changed, + run: async signal => { + const admitted = () => !signal.aborted && canRun(); + for (const [id, at] of freshScanAfter) if (at <= Date.now()) freshScanAfter.delete(id); + for (const [id, at] of deadlines) if (at <= Date.now()) { + dirty.add(id); + if (active?.id === id) active.rescanOwed = true; + freshScanAfter.delete(id); + deadlines.delete(id); + } + for (const [id, entries] of cacheChanges) { + for (const [directory, at] of entries) if (at <= Date.now()) { + dirty.add(id); + if (active?.id === id) active.rescanOwed = true; + freshScanAfter.delete(id); + entries.delete(directory); + } + if (entries.size === 0) cacheChanges.delete(id); + } + let inspected = 0; + let companies = 0; + const scope = discovery; + if (scope && scope.retryAt <= Date.now() && admitted()) { + discoveryLoading = true; + try { + const rows = await service.listCompanies(scope.after); + for (const row of rows) dirty.add(row.id); + if (discovery === scope) { + if (rows.length === 51) discovery = { after: rows.at(-1)!.id, retryAt: 0 }; + else { + discoveryAfter = Date.now() + 30_000; + discovery = discoveryAgain ? { retryAt: discoveryAfter } : null; + discoveryAgain = false; + } + } + } catch (err) { + if (discovery === scope) scope.retryAt = Date.now() + 30_000; + logger.warn({ err }, "skill cache retention company discovery failed"); + } finally { discoveryLoading = false; } + } + while (companies < 50 && inspected < 50 && admitted()) { + if (!active) { + const id = [...dirty].find(id => readyAt(id) <= Date.now()); + if (!id) break; + dirty.delete(id); + active = { id, cursor: null, nextRunAt: null, rescanOwed: false }; + deadlines.delete(id); + cacheChanges.delete(id); + } + const company = active; + try { + const result = await service.pruneCompany(company.id, { + cursor: company.cursor, maxEntries: 50 - inspected, shouldContinue: admitted, + }); + if (result.checked > 0 || result.failed > 0 || Object.keys(result.skipped).length > 0) { + logger.info({ ...result, companyId: company.id, dryRun: service.dryRun }, "skill version snapshot retention"); + } + inspected += result.checked; + companies++; + if (result.nextRunAt !== null) company.nextRunAt = Math.min(company.nextRunAt ?? Infinity, result.nextRunAt); + company.cursor = result.cursor; + failures.delete(company.id); + if (result.interrupted || !admitted()) { + company.cursor = null; + company.nextRunAt = null; + continuationAt = Date.now() + 1000; + return; + } + if (result.truncated) break; + if (company.rescanOwed) freshScanAfter.delete(company.id); + else freshScanAfter.set(company.id, Date.now() + 30_000); + if (company.nextRunAt !== null) deadlines.set(company.id, company.nextRunAt); + active = null; + } catch (err) { + dirty.add(company.id); + failures.set(company.id, Date.now() + 30_000); + active = null; + companies++; + logger.warn({ err, companyId: company.id }, "skill cache retention company inspection failed"); + } + } + continuationAt = Date.now() + 1000; + }, + next: async () => { + let next: number | null = null; + const take = (at: number) => { next = Math.min(next ?? Infinity, at); }; + if (discovery) take(Math.max(continuationAt, discovery.retryAt)); + if (active) take(continuationAt); + for (const id of dirty) take(Math.max(continuationAt, readyAt(id))); + for (const [id, at] of deadlines) take(Math.max(at, failures.get(id) ?? 0)); + for (const id of cacheChanges.keys()) take(Math.max(cacheDeadline(id), failures.get(id) ?? 0)); + return next; + }, + }); + const unsubscribe = onSkillVersionCacheChange(change => { + if (change.root !== root) return; + let entries = cacheChanges.get(change.companyId); + if (!entries) { entries = new Map(); cacheChanges.set(change.companyId, entries); } + entries.set(change.directory, change.occurredAt + service.graceMs); + if (change.source === "publication") dirty.add(change.companyId); + worker.wake(); + }); + return { ...worker, unsubscribe, dispose: () => { unsubscribe(); cacheChanges.clear(); return service.dispose(); } }; +} diff --git a/server/src/services/skill-version-retention.ts b/server/src/services/skill-version-retention.ts index ab507fd8ee2..f959efa4ef4 100644 --- a/server/src/services/skill-version-retention.ts +++ b/server/src/services/skill-version-retention.ts @@ -5,6 +5,7 @@ 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 { noteSkillVersionCacheChange } from "./skill-version-retention-notifications.js"; import { logger } from "../middleware/logger.js"; import { allocatedBytes, ensureSkillCacheDirectory, hasRestorableSnapshot, lastUsedAt, protectedVersions, SkillCacheBusy, skillVersionCachePaths, touchSkillVersionCache, @@ -30,8 +31,6 @@ import { allocatedBytes, ensureSkillCacheDirectory, hasRestorableSnapshot, lastU 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; const VERSIONS_DIR = "__versions__"; const TRASH_PREFIX = ".trash-"; @@ -67,10 +66,12 @@ export async function runVersionSnapshotFlight(skillDir: string, work: () => Pro } const existing = versionSnapshotFlights.get(skillDir); if (existing) return existing; + const startedAt = Date.now(); const flight = work(); versionSnapshotFlights.set(skillDir, flight); try { return await flight; } finally { if (versionSnapshotFlights.get(skillDir) === flight) versionSnapshotFlights.delete(skillDir); + noteSkillVersionCacheChange(skillDir, { occurredAt: startedAt }); } } @@ -150,11 +151,13 @@ export type SkillVersionPruneResult = { skippedRecent: number; failed: number; truncated: boolean; skipped: Record; cursor: CompanyCursor | null; + nextRunAt: number | null; + interrupted: boolean; }; 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 }; + skippedRecent: 0, failed: 0, truncated: false, skipped: {}, cursor: null, nextRunAt: null, interrupted: false }; } type Candidate = { parent: string; skillId: string; name: string; unsafe?: boolean; discoveryOnly?: boolean }; @@ -214,7 +217,8 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune async function protectedVersionIds(companyId: string, skillIds: string[], keepNewest: number) { const keep = new Set(); - if (skillIds.length === 0) return keep; + const leaseDeadlines = new Map(); + if (skillIds.length === 0) return { versions: keep, leaseDeadlines }; 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))); @@ -235,9 +239,10 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune } } const runVersions = await protectedVersions(db, companyId); - if (!runVersions) return null; - for (const version of runVersions) keep.add(version); - return keep; + if (!runVersions.versions) return { versions: null, leaseDeadlines }; + for (const [version, at] of runVersions.leaseDeadlines) if (!keep.has(version)) leaseDeadlines.set(version, at); + for (const version of runVersions.versions) keep.add(version); + return { versions: keep, leaseDeadlines }; } async function pruneCompanyPage(companyId: string, options: SkillVersionPruneOptions & { cursor?: CompanyCursor | null } = {}): Promise { @@ -246,11 +251,20 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune 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 canContinue = 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 shouldContinue = () => { + const allowed = canContinue(); + if (!allowed) result.interrupted = true; + return allowed; + }; + const schedule = (at: number) => { result.nextRunAt = Math.min(result.nextRunAt ?? Infinity, at); }; + const skip = (reason: string) => { + result.skipped[reason] = (result.skipped[reason] ?? 0) + 1; + if (["cache_busy", "scan_busy", "unknown_run_pins", "unavailable_pins", "unsafe_or_unavailable", "cache_changed"].includes(reason)) schedule(now + 30_000); + }; const trash: string[] = []; const root = resolveCompanySkillVersionsRoot(companyId); const trashName = (name: string) => name.startsWith(TRASH_PREFIX) && UUID_PATTERN.test(name.slice(TRASH_PREFIX.length)); @@ -286,7 +300,7 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune 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; + let initialKeep: Awaited>; try { initialKeep = await withSkillVersionCacheLock(db, companyId, () => protectedVersionIds(companyId, skillIds, keepNewest), true); } catch (error) { result.checked = 1; @@ -296,11 +310,11 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune result.truncated = false; return result; } - if (!initialKeep) { await closeScan(companyId); result.truncated = false; result.checked = 1; skip("unknown_run_pins"); return result; } + if (!initialKeep.versions) { 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>; + eligible: (keep: Awaited>) => Promise<{ dev: number; ino: number } | null>; }> = []; for (const entry of entries) { if (!shouldContinue()) { scan.pending.push(...entries.slice(result.checked)); result.truncated = true; break; } @@ -317,21 +331,28 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune 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) => { + const eligible = async (keep: Awaited>) => { 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; } + if (!keep.versions) { skip("unknown_run_pins"); return null; } + if (isVersion && keep.versions.has(entry.name)) { + const leaseDeadline = keep.leaseDeadlines.get(entry.name); + if (leaseDeadline !== undefined) schedule(Math.max(now + 1000, leaseDeadline)); + 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); + if (!dryRun) { + await touchSkillVersionCache(companyId, entry.skillId, entry.name, now); + schedule(now + graceMs); + } 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; } + if (now - used < graceMs) { schedule(used + graceMs); result.skippedRecent++; skip("recent_use"); return null; } + } else if (now - stat.mtimeMs < graceMs) { schedule(stat.mtimeMs + graceMs); result.skippedRecent++; return null; } return stat; }; try { @@ -373,7 +394,7 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune result.candidateBytes += candidate.bytes; if (dryRun) continue; const renamed = await removeVersionDirectory(candidate.entry.parent, candidate.entry.name); - if (!renamed) continue; + if (!renamed) { schedule(now + 30_000); continue; } trash.push(renamed); result.removed++; if (candidate.isStaging) result.removedStaging++; @@ -406,8 +427,9 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune }, true); } catch (error) { if (!shouldContinue()) break; - if (error instanceof SkillCacheBusy || errorCode(error) === "ENOENT") continue; - result.failed++; + if (errorCode(error) === "ENOENT") continue; + if (error instanceof SkillCacheBusy) { schedule(now + 30_000); continue; } + result.failed++; schedule(now + 30_000); logger.warn({ err: error, companyId }, "skill version snapshot trash removal failed"); } } @@ -415,7 +437,7 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune } async function pruneCompany(companyId: string, options: SkillVersionPruneOptions & { cursor?: CompanyCursor | null } = {}) { - if (scanning.has(companyId)) return { ...emptyResult(), checked: 1, skipped: { scan_busy: 1 } }; + if (scanning.has(companyId)) return { ...emptyResult(), checked: 1, nextRunAt: Date.now() + 30_000, skipped: { scan_busy: 1 } }; scanning.add(companyId); try { return await pruneCompanyPage(companyId, options); } finally { scanning.delete(companyId); } @@ -438,6 +460,8 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune 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++; + totals.interrupted ||= result.interrupted; + if (result.nextRunAt !== null) totals.nextRunAt = Math.min(totals.nextRunAt ?? Infinity, result.nextRunAt); 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: "" }; @@ -451,5 +475,10 @@ export function skillVersionRetentionService(db: Db, defaults: SkillVersionPrune totals.cursor = cursor; return totals; } - return { pruneCompany, tick, dispose }; + async function listCompanies(after?: string) { + return db.select({ id: companies.id }).from(companies) + .where(after ? gt(companies.id, after) : undefined).orderBy(asc(companies.id)).limit(51); + } + return { pruneCompany, tick, listCompanies, dispose, dryRun: defaults.dryRun ?? false, + graceMs: Math.max(0, defaults.graceMs ?? SKILL_VERSION_RETENTION_DEFAULT_GRACE_MS) }; } diff --git a/tests/e2e/connection-reviews.spec.ts b/tests/e2e/connection-reviews.spec.ts index 6e05daef680..bf36a6db886 100644 --- a/tests/e2e/connection-reviews.spec.ts +++ b/tests/e2e/connection-reviews.spec.ts @@ -320,7 +320,11 @@ for (const journey of [ await expect( queue.getByRole("button", { name: "Decline", exact: true }), ).not.toBeVisible({ timeout: 20_000 }); + await page.bringToFront(); await page.reload(); + await expect(page.getByTestId("task-chat-history-content")).toBeVisible({ + timeout: 30_000, + }); await expect( page .getByText(