mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
fix(server): schedule skill cache retention from changes and saved deadlines (#17923)
<!-- entire-trail-link-start --> https://entire.com/gh/paperclipai/-/trails/83/changes/paperclip/83 <!-- entire-trail-link-end --> ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The server caches saved skill versions on disk so runs can restore their exact instructions. > - The old retention owner scans company rows every hour, even after all caches are protected or removed. > - Cache publication, version changes, and pin release already have local or transaction-owned boundaries. > - Grace windows and terminal controller leases already have saved timestamps. > - This pull request uses those changes and deadlines to drive the existing bounded prune service. > - Idle retention stops SQL and directory reads while pending cleanup remains recoverable. ## Linked Issues or Issue Description Stack slice 64. Parent: #17876 (`codex/sleep-63-agent-timer-deadlines`). Related: #15625 establishes transaction-owned work signals. #17907 independently keeps an active cache-lock transaction alive; this PR preserves that lock and does not duplicate that repair. **What happened?** At parent `31c12f1d68ff4a51e06fc2be2abf2c2aee9f9d3a`, skill-version retention has an hourly company scan, a startup timeout, and separate continuation timers. Empty or permanently protected caches still cause recurring SQL. **Expected behavior** Reconstruct once, inspect changed companies, and schedule saved access/staging grace or terminal controller deadlines. Stop recurring work when no timed or uncertain obligation remains. Keep current/newest/agent/run protection, restorable-byte validation, cache locking, final pin revalidation, atomic rename, and trash recovery. **Steps to reproduce** Start an enabled server with no cached versions, or only current/pinned versions. After the initial scan, advance an hour. The old interval selects companies again. The new worker has no deadline and performs no retention SQL or directory scan in that state. ## What Changed - Signal the missing-local-source reconciliation deletion in its transaction. Limit fresh scans of a successfully observed company to one per thirty seconds. Keep one-second page continuations and actual saved deadlines independent. Empty or uncertain passes do not delay first publication. - Restart a company scan if drain interrupts inspection or post-rename trash cleanup. Restore bounded candidate/byte logging with dry-run mode. Three controlled cases failed before repair and the complete 34-case retention suite passes afterward. - Register retention through the delivery coordinator and shared scheduler. Remove the hourly, startup, and continuation timers from `index.ts`. Track startup work without delaying readiness. Stop delivery work before closing retained iterators. - Keep one active company cursor and bounded 50-entry passes. Preserve observed deadlines across pages and bound company discovery to 51 rows per page. Retain concrete company failures and incomplete pins for retry. - Report saved grace and terminal controller deadlines from the existing cache owner. PostgreSQL remains the controller-lease authority. Current/newest/agent pins do not arm unrelated lease checks. - Wake on cache publication, rejected materialization, and access markers. Signal version creation/deletion and agent configuration/hire changes in their owning transactions. Extend existing instruction, finalization/environment, and retained-session fanouts with company hints. - Route reused provider-task prior-owner retirement through `writeEnvironmentCleanup`, so that supported pin-release write also settles a notification. - Add an optional scope callback to `registerDeadlineWork`. Existing consumers omit it and keep their behavior. The feature owns persistence predicates; startup only registers it. Add no table or migration. Full producer audit, shared-contract explanation, lifecycle evidence, and limitations: [`doc/plans/2026-10-27-skill-version-retention-sleep-audit.md`](doc/plans/2026-10-27-skill-version-retention-sleep-audit.md). ## Verification - Reconciled all existing sleep changes with actual master `7e27bc5007c8f9cb8cdce06c24f9813e27f7120f` through normal merges. All 24 retention-owned and 95 unrelated upstream paths are preserved; package, UI, E2E and map trees equal master. - Combined-source qualification: **1,009 TypeScript cases across 42 complete files**, plus **28 Rust cases** in two targets. One intentional Linux-inotify-only adapter case is skipped on this macOS host. Coverage includes actual retention/cache/producer/runtime behavior, startup, shared deadline consumers, changed OpenCode/observability and native execution boundaries. - After the subsequent task-index/task-read upstream merges, **85 cases across six complete files** passed on final source `d9277b1858a2230bdb726974c90a0c9bc5c5ebe6`: real retention/cache/work, task reads/query behavior and concurrent migration repair. Counts overlap earlier coverage. Fresh PostgreSQL 17 schemas apply all 413 journal entries through index 414. - Database/adapter/Runner builds, production server and Runner typechecking, module boundaries, target migration order and generated TypeScript/Rust release-profile consistency pass. Selected source hashes and temporary fixture imports restore exactly. - Reproduced and repaired Greptile's own-company due-cleanup P2 with identical old-worker-failing/new-worker-passing assertions. The 30-second ordinary scan floor remains; already-due cleanup receives its owed scan and disarms afterward. Earlier 799-case and fresh-blank-plugin acceptance remain in the audit. - **Inherited validation exception:** the separate Copilot source-identity check fails after upstream OpenCode reserved-environment changes; all 72 identity inputs equal current master. No provider release attestation is regenerated or new live qualification claimed here. Required exact-head CI/security and Greptile 5/5 still must pass. - Historical results, repair details and limitations remain in [the retention audit](doc/plans/2026-10-27-skill-version-retention-sleep-audit.md). No full local repository-suite, live provider, production-load, external-alarm, measured suspend/wake or fleet-savings claim. - CI on the prior published head found four connection-review assertions racing the concealed conversation after reload. Its fresh Greptile/Ponytail reviews were both 5/5, but the failing browser gate was not waived or retried unchanged. The fixture now foregrounds the task tab and waits for visible history before retaining all existing saved-result assertions. The full file passes four journeys locally; the controlled-restart case is intentionally skipped by the default configuration. Original local journeys also passed unchanged, so no deterministic local red reproduction is claimed. See the audit for CI failure artifacts and scope. Runtime and prior sleep-test sources remain unchanged. ## Risks - Supported live writers are the existing service/API owners in one process. Independent SQL/FS writers need an ingress contract. Startup reconstruction covers restarts. - Notifications cause a bounded inspection; they do not authorize deletion. Durable pin predicates and the cache lock remain authoritative. - A database clock behind application time can require another one-second check of the concrete expired controller lease. Unknown pins, lock contention, and failed inspection retain retries until resolved. - The in-process scheduler cannot wake stopped compute. This PR preserves Cloud sleep-safety gates and external-alarm coverage. It does not prove deployed suspension or fleet savings. Cache reclamation can wait until a supported wake when compute is stopped. - Native-harness backup retention and the other remaining duties are separate stack slices. ## Model Used OpenAI Codex (GPT-6-based coding agent), using repository inspection, normal Git merges, composition proofs, a controlled red/green regression and complete PostgreSQL/filesystem integration suites. No subagents were used in this reconciliation. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
25 files changed
+1393
-102
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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<string, unknown> } | null,
|
||||
run: { id: "run", agentId: "agent", status: "running" } as { id: string; agentId: string; status: string } | null,
|
||||
});
|
||||
function database(value: ReturnType<typeof row> | null = row(), projectRows = [{ id: projectId }], workspaceRows: unknown[] = []) {
|
||||
function database(value: ReturnType<typeof row> | 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<string, unknown>[] = [];
|
||||
const update: any = { set: (value: Record<string, unknown>) => { 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<unknown>) => 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: [] };
|
||||
|
||||
@@ -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<typeof createDb>,stop:()=>Promise<void>;
|
||||
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);
|
||||
|
||||
@@ -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<Array<{ id: string }>>();
|
||||
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();
|
||||
|
||||
@@ -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<typeof createDb>, 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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let home: string;
|
||||
const previousHome = process.env.PAPERCLIP_HOME;
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("skill-version-cache-");
|
||||
db = createDb(database.connectionString);
|
||||
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<typeof setTimeout> | undefined;
|
||||
const preparation = withSkillVersionCacheLock(other, f.companyId, async () => {
|
||||
await companySkillService(other).listRuntimeSkillEntries(f.companyId, {
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let worker: ReturnType<typeof registerSkillVersionRetentionWork> | 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<typeof skillVersionRetentionService>;
|
||||
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();
|
||||
});
|
||||
@@ -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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let worker: ReturnType<typeof registerSkillVersionRetentionWork> | 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<void>(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<void>(), release = Promise.withResolvers<void>();
|
||||
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<void>(), began = Promise.withResolvers<void>();
|
||||
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);
|
||||
|
||||
+11
-39
@@ -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<typeof setTimeout> | undefined;
|
||||
const sweepSkillVersionSnapshots = () => {
|
||||
if (!config.skillVersionRetentionEnabled || skillVersionRetentionInFlight || isIdleTaskDrainActive() || heartbeatSchedulerStopped || isWarmStandby()) return;
|
||||
skillVersionRetentionInFlight = true;
|
||||
trackHeartbeatSchedulerWork(skillVersionRetention.tick()
|
||||
.then((result) => {
|
||||
if (result.checked > 0 || result.failed > 0 || Object.keys(result.skipped).length > 0) {
|
||||
logger.info({ ...result, dryRun: config.skillVersionRetentionDryRun }, "skill version snapshot retention");
|
||||
}
|
||||
if (result.truncated && !heartbeatSchedulerStopped) {
|
||||
skillVersionRetentionContinuation = setTimeout(sweepSkillVersionSnapshots, 1000);
|
||||
skillVersionRetentionContinuation.unref?.();
|
||||
}
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "skill version snapshot retention failed");
|
||||
})
|
||||
.finally(() => { skillVersionRetentionInFlight = false; }));
|
||||
};
|
||||
const skillVersionRetentionInterval = setInterval(sweepSkillVersionSnapshots, SKILL_VERSION_RETENTION_INTERVAL_MS);
|
||||
skillVersionRetentionInterval.unref?.();
|
||||
const skillVersionRetentionStartup = setTimeout(sweepSkillVersionSnapshots, SKILL_VERSION_RETENTION_STARTUP_DELAY_MS);
|
||||
skillVersionRetentionStartup.unref?.();
|
||||
const 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?.();
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -9,6 +9,7 @@ export function registerDeadlineWork(
|
||||
next: () => Promise<number | null>;
|
||||
canRun: () => boolean;
|
||||
retryMs: number;
|
||||
onSettled?: (hints?: ReadonlySet<string>) => 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; }
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -8,7 +8,7 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Nati
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup);
|
||||
}
|
||||
|
||||
export function writeEnvironmentCleanup<T>(db: Db, write: (tx: Db) => PromiseLike<T>, scope?: NativeEnvironmentSettlementScope): Promise<T> {
|
||||
export function writeEnvironmentCleanup<T>(db: Db, write: (tx: Db) => PromiseLike<T>, scope: NativeEnvironmentSettlementScope): Promise<T> {
|
||||
return db.transaction(async tx => {
|
||||
await notifyEnvironmentCleanup(tx, scope);
|
||||
return write(tx as unknown as Db);
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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) : "*");
|
||||
}
|
||||
@@ -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 }) : "*");
|
||||
}
|
||||
@@ -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 }));
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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<number | null> {
|
||||
|
||||
// A terminal run can still own provider execution or recovery. Only fully settled
|
||||
// runs release their pins; resume preparation reacquires the same cache lock.
|
||||
export async function protectedVersions(db: Db, companyId: string): Promise<Set<string> | null> {
|
||||
export async function protectedVersions(db: Db, companyId: string): Promise<{ versions: Set<string> | null; leaseDeadlines: Map<string, number> }> {
|
||||
const rows = await db.select({
|
||||
id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
leaseDeadline: sql<number | null>`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<unknown>`${heartbeatRuns.contextSnapshot}->'skillVersions'`,
|
||||
// Avoid loading large execution inputs for runs with the modern pin map.
|
||||
legacyInput: sql<unknown>`case when ${heartbeatRuns.runtimeMode} = 'native'
|
||||
@@ -121,7 +126,13 @@ export async function protectedVersions(db: Db, companyId: string): Promise<Set<
|
||||
and l.lease_policy = 'reuse_by_environment' and l.cleanup_status = 'success', false))))
|
||||
)`));
|
||||
const versions = new Set<string>();
|
||||
for (const { pins, legacyInput, id, agentId, nativeIssueId } of rows) {
|
||||
const leaseDeadlines = new Map<string, number>();
|
||||
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<Set<
|
||||
const input = parseNativeExecutionInput(legacyInput);
|
||||
if (!("runtimeContext" in input) || input.binding.companyId !== companyId
|
||||
|| input.binding.runId !== id || input.binding.agentId !== agentId
|
||||
|| input.binding.issueId !== nativeIssueId) return null;
|
||||
|| input.binding.issueId !== nativeIssueId) return unknown();
|
||||
for (const skill of input.runtimeContext.skills) {
|
||||
// Unversioned skills have their own captured asset bundle; they do
|
||||
// not reference a directory under __versions__.
|
||||
if (skill.versionId === null) continue;
|
||||
if (!uuid.test(skill.versionId)) return null;
|
||||
versions.add(skill.versionId);
|
||||
if (!uuid.test(skill.versionId)) return unknown();
|
||||
add(skill.versionId);
|
||||
}
|
||||
continue;
|
||||
} catch { return null; }
|
||||
} catch { return unknown(); }
|
||||
}
|
||||
// Legacy/incomplete preparation is not proof that no skill is in use.
|
||||
if (!pins || typeof pins !== "object" || Array.isArray(pins)) return null;
|
||||
if (!pins || typeof pins !== "object" || Array.isArray(pins)) return unknown();
|
||||
for (const version of Object.values(pins)) {
|
||||
if (typeof version !== "string" || !uuid.test(version)) return null;
|
||||
versions.add(version);
|
||||
if (typeof version !== "string" || !uuid.test(version)) return unknown();
|
||||
add(version);
|
||||
}
|
||||
}
|
||||
return versions;
|
||||
return { versions, leaseDeadlines };
|
||||
}
|
||||
|
||||
function validateInventory(inventory: CompanySkillVersionFileInventoryEntry[]) {
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import { readPaperclipSkillSyncPreference } from "@paperclipai/adapter-utils/server-utils";
|
||||
import path from "node:path";
|
||||
import { resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export function notifySkillVersionRetention(tx: object, companyId: string) {
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.skillVersionRetention, companyId);
|
||||
}
|
||||
|
||||
export function notifySkillVersionRetentionReconstruction(tx: object) {
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.skillVersionRetention, "*");
|
||||
}
|
||||
|
||||
type CacheChange = { root: string; companyId: string; directory: string; occurredAt: number; source: "access" | "publication" };
|
||||
const listeners = new Set<(change: CacheChange) => 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<string, unknown>, after: Record<string, unknown>) {
|
||||
if (JSON.stringify(readPaperclipSkillSyncPreference(before).desiredSkillEntries)
|
||||
!== JSON.stringify(readPaperclipSkillSyncPreference(after).desiredSkillEntries)) {
|
||||
await notifySkillVersionRetention(tx, companyId);
|
||||
}
|
||||
}
|
||||
@@ -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<typeof skillVersionRetentionService>;
|
||||
type ActiveCompany = { id: string; cursor: SkillVersionPruneResult["cursor"]; nextRunAt: number | null; rescanOwed: boolean };
|
||||
|
||||
export function registerSkillVersionRetentionWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
service: RetentionService,
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
const root = path.resolve(resolvePaperclipInstanceRoot(), "skills");
|
||||
const dirty = new Set<string>();
|
||||
const deadlines = new Map<string, number>();
|
||||
const failures = new Map<string, number>();
|
||||
const freshScanAfter = new Map<string, number>();
|
||||
const cacheChanges = new Map<string, Map<string, number>>();
|
||||
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<string>) => {
|
||||
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(); } };
|
||||
}
|
||||
@@ -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<string, number>;
|
||||
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<string>();
|
||||
if (skillIds.length === 0) return keep;
|
||||
const leaseDeadlines = new Map<string, number>();
|
||||
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<SkillVersionPruneResult> {
|
||||
@@ -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<string> | null;
|
||||
let initialKeep: Awaited<ReturnType<typeof protectedVersionIds>>;
|
||||
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<string> | null) => Promise<{ dev: number; ino: number } | null>;
|
||||
eligible: (keep: Awaited<ReturnType<typeof protectedVersionIds>>) => 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<string> | null) => {
|
||||
const eligible = async (keep: Awaited<ReturnType<typeof protectedVersionIds>>) => {
|
||||
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) };
|
||||
}
|
||||
@@ -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(
|
||||
|
||||
Reference in new issue
Block a user