mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-29 12:59:13 +01:00
fix(server): schedule owed agent directory cleanup (#17671)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agent runs can save their directory before temporary-file cleanup finishes. > - The existing receipt records stop proof and unfinished cleanup. > - The orphan reaper still searches for this debt on every maintenance pass. > - This PR activates cleanup from committed changes and saved retry deadlines. > - The directory owner keeps its resource checks and file lock. > - Settled directory history no longer causes recurring cleanup queries. ## Linked Issues or Issue Description Refs #17670, the merged captured-instruction commit slice. This PR targets master. It does not include stopped instruction collection or change the saved-file contract from #14695. **What needs to be improved?** The post-save part of `recoverCaptured` queries directory copies on each orphan-reaper pass, even after all temporary directories have been released. **Why is this improvement important?** Empty database work prevents idle tenant instances from becoming quiet. Cleanup must still survive a crash after a save and preserve files when stop or resource ownership is uncertain. **Proposed solution** Use existing copy receipts as the source of truth. Notify the shared coordinator inside relevant copy-write transactions. Reconstruct pending work at startup and register each saved retry deadline. Keep physical cleanup in the directory owner. Add a partial index over unfinished cleanup rather than a new queue table. Related-PR search found workspace archive and SSH task-workspace cleanup changes (#17175 and #16464). They own different resources and do not replace this stopped agent-directory receipt recovery. ROADMAP.md has no duplicate migration. ## What Changed - Add a scoped directory cleanup worker with keyset discovery and batches of 20. - Signal relevant copy transitions, company/agent deletion, and worktree execution policy changes. - Preserve the due-time claim, current receipt read, directory lock, stop proof, and exact remote-resource checks. - Read the receipt after release; a returned promise does not establish cleanup completion. - Keep unresolved cleanup on its saved retry or shared failure backoff. Final receipts disarm the worker. - Register the worker during startup and skip only this lane in periodic recovery. Keep manual recovery compatible. - Add concurrent, replay-safe partial index migration `0407_greedy_star_brand.sql`. Retain the five newest schema snapshots. - Add the producer audit and lifecycle evidence in `doc/plans/2026-10-25-directory-cleanup-sleep-audit.md`. ## Verification Published head: `210fcb1215a3b5e5ed84e420ddb73ab5952d9365`. Source acceptance: `70a6800eea874bdb87685bf6eea66de79c48b577`. Parent #17670 landed as `b294de546bf9cd3a8578a7a66d7e8ec0eb939799`; normal squash-ancestry reconciliation preserves the entire accepted descendant tree. Final follow-ups change audits only. - All 284 cases in six complete filesystem, startup and heartbeat owner files pass, without skips, in 48.88 seconds. Four unused fresh PostgreSQL 17 databases migrated through 0407 isolate destructive fixtures. - Three fixtures use supported external database paths. The embedded-only lifecycle fixture temporarily uses its own external database on this IPC-exhausted host, then is restored byte for byte. All assertions and original timing limits remain unchanged. This qualifies the local result; CI must still pass with the original embedded fixture. - Actual database build, server typecheck/build, module boundaries and exact-current migration order pass. The original concurrent idempotent index SQL bytes are unchanged. All prior executable migration history, journal entries and retained snapshots stay intact. The five snapshots are 0403 through 0407. - Populated 0406-to0407 upgrade preserves all 405 old receipts and all 38 companies, adds exactly one receipt and adds none on replay. The cleanup index is valid and ready on fresh and upgraded databases. The entire current database tree equals the source used for this proof. - Both complete original and upstream patches are preserved. The actual parent squash tree equals the already-adopted reviewed parent; resolving eleven duplicate-ancestry text conflicts leaves the whole accepted tree unchanged, not just selected files. Complete server, package, UI and test trees still match the 284-case source acceptance. - The audit retains earlier ownership, stopped-resource, receipt, fixture and local-runner failures. The no-projects build filter was not counted as a build pass; corrected actual build and complete reruns passed before publication. Old draft review scores do not count for this new head. - Current-head Greptile gives 5/5 with no actionable findings, and all 57 checks pass. These checks establish worker behavior, not deployed host/database suspension or measured savings. ## Risks - Signals cover the existing single-process writer model. Independent SQL writers and replicas need separate integration. - The worker retains identifiers for actual pending work. Its index excludes completed history but does not make large pending backlogs free. - File/provider cleanup can outlive an admission change once started. Shutdown joins admitted operations. Unknown remote ownership remains pending and no replacement environment is acquired for cleanup. - The partial index adds storage and write cost. Its concurrent migration uses the existing interrupted-index repair path. - Cloud alarm coverage and host-sleep safety remain unchanged. This establishes worker quiet, not deployed host/database suspension or measured savings. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, shell tools, and PostgreSQL/filesystem tests. The runtime does not expose a more precise model ID or context-window size for this turn. ## 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
f03522ea8f
commit
267842fc3c
18 files changed
+969
-21
No files matched your search
@@ -2569,3 +2569,19 @@ startup/instruction-release wiring passes 64 (184 distinct tests across these
|
||||
selections). Workspace types/build pass. Fresh remote gates remain required.
|
||||
See the instruction recovery audit for producer coverage and all failed attempts.
|
||||
Earlier stack CI/review findings and final master reconciliation remain outstanding.
|
||||
|
||||
## Slice 40 remote acceptance and slice 41 directory cleanup
|
||||
|
||||
Captured instruction commits (#17670) at
|
||||
`03e1b3cd7270c7438a47562b170b3a9b2dbf112e` now have all 53 checks successful,
|
||||
fresh Greptile 5/5, and zero unresolved threads. The PR remains unmerged.
|
||||
|
||||
Slice 41 owns post-save directory release only. It reconstructs existing cleanup
|
||||
receipts, signals relevant copy writes and invalidations, and delegates deletion
|
||||
to the existing directory owner. It schedules saved retry times and leaves
|
||||
settled history quiet. See [directory cleanup audit](2026-10-25-directory-cleanup-sleep-audit.md)
|
||||
for producer coverage, lifecycle, index plan, failure history and local tests.
|
||||
Local verification currently has 200 passing cases across directory, startup,
|
||||
captured instruction and heartbeat release suites, plus workspace typecheck/build.
|
||||
Remote acceptance is pending. Stopped collection and the remaining
|
||||
original inventory and older stack review/CI gates remain outstanding.
|
||||
@@ -0,0 +1,263 @@
|
||||
# Post-save agent-directory cleanup
|
||||
|
||||
Date: 2026-10-25. Parent: captured instruction commit PR #17670 at
|
||||
`03e1b3cd7270c7438a47562b170b3a9b2dbf112e`.
|
||||
|
||||
## One reason and its owner
|
||||
|
||||
A stopped agent-directory copy may have saved its files but still owe deletion
|
||||
of its temporary local or remote materialization. Previously every orphan-reaper
|
||||
pass called the cleanup portion of `recoverCaptured`, including with no cleanup
|
||||
owed. This slice moves that reason to the existing delivery coordinator and
|
||||
scheduler. Captured instruction commits remain the separate parent slice;
|
||||
stopped instruction collection remains in the reaper for a subsequent slice.
|
||||
|
||||
`agent-instruction-working-copies.ts` owns discovery, scheduling and copy writes.
|
||||
`agent-directory-working-copies.ts` continues to own filesystem serialization,
|
||||
stop proof, materialization identity, remote lease checks, and physical deletion.
|
||||
The worker calls that owner; it does not move provider behavior into startup or
|
||||
the scheduler. No new table, independent timer, or provider branch is added.
|
||||
|
||||
## Durable obligation and deadline
|
||||
|
||||
The existing copy row owes this cleanup when all of these hold:
|
||||
|
||||
- State is saved, unchanged, resolved or unavailable.
|
||||
- `processStoppedAt` records stop proof.
|
||||
- Receipt schema is `paperclip.agent-files.v1`.
|
||||
- Ownership has not transferred through `retainedByRunId`.
|
||||
- Receipt has a baseline or `cleanupPending: true`, and is not final with
|
||||
`cleanupPending: false`.
|
||||
|
||||
A final false flag stays final even if an old receipt retains a baseline. This
|
||||
matches the directory owner's existing no-op rule. Preparation and live warm
|
||||
copies are not stop proof. Candidate bytes, saved canonical files, storage
|
||||
warnings, conflict policy and retention semantics are unchanged.
|
||||
|
||||
Startup reconstructs pending identifiers and saved `nextAttemptAt` in keyset
|
||||
pages of 100. It does not transfer baseline JSON or candidate bytes into the
|
||||
scheduler. Fair batches execute at most 20 entries per pass. The scheduler
|
||||
retains identifiers for all outstanding entries; memory scales with actual debt.
|
||||
|
||||
A due check rereads current state and reserves the existing 30-second retry with
|
||||
the copy CAS and a due-time guard before I/O. A concurrent later deadline wins.
|
||||
Physical release joins the existing canonical directory lock and rereads its
|
||||
current owner. Completion is established by the saved receipt, not a void return
|
||||
from release. Deferred remote cleanup remains scheduled. If no future retry was
|
||||
saved, an error retains the shared bounded backoff rather than an immediate loop
|
||||
or forgotten entry. Failed entries do not prevent other admitted entries from
|
||||
being attempted. Manual/startup `recoverCaptured` preserves its default behavior
|
||||
and per-row error handling; only its migrated periodic caller skips this lane.
|
||||
|
||||
## Producers and transaction boundary
|
||||
|
||||
The common copy patch boundary signals before writes that enter, change or leave
|
||||
cleanup eligibility. It reuses the short transaction introduced for captured
|
||||
commits. File operations, merge locks and remote calls stay outside that new
|
||||
transaction. Nested use retains the existing outer-commit and uncertain-outcome
|
||||
protocol. A signal only requests inspection of authoritative records.
|
||||
|
||||
The producer audit found only these two copy-table owners. Directory preparation
|
||||
inserts preparing rows; warm handoff inserts prepared successors and supersedes
|
||||
prior warm rows. Direct adoption requires `unchanged_turn` and no stop proof.
|
||||
Those direct writes cannot create this post-save obligation. Eligible transitions
|
||||
use the shared patch. Agent/company deletion invalidates the corresponding
|
||||
company before run cascades. Worktree execution policy changes reactivate the
|
||||
worker through the existing settings transaction. Changes to remote termination
|
||||
proof can make existing pending cleanup executable; the pending row's retry
|
||||
retains coverage without requiring an independent lease-table scan.
|
||||
|
||||
The supported model remains one process with instrumented writes. Independent
|
||||
SQL writers and additional replicas need their own integration. This PR does
|
||||
not claim those writers are observed by local notifications.
|
||||
|
||||
## Lifecycle and resource safety
|
||||
|
||||
Drain and standby retain requested scopes/deadlines without recurring SQL. The
|
||||
heartbeat facade applies the existing worktree execution policy and task-drain
|
||||
gate; enabling worktree execution signals a fresh reconstruction. Startup
|
||||
registration follows existing recovery and respects scheduler disablement.
|
||||
Shutdown joins admitted cleanup, including a held real filesystem deletion.
|
||||
|
||||
Remote cleanup still requires the exact current lease, materialization path and
|
||||
termination evidence. It never acquires a replacement environment to perform
|
||||
cleanup. Unknown ownership remains pending. A returned cleanup promise is not
|
||||
proof that the provider released its resource. Cached transports cannot override
|
||||
current lease state. Existing owner tests retain these fences.
|
||||
|
||||
No host-sleep safety check or Cloud wake capability is relaxed. This migration
|
||||
establishes worker quiet for settled directory history, not deployed suspension
|
||||
or measured cost savings.
|
||||
|
||||
## Index and migration
|
||||
|
||||
Migration `0407_greedy_star_brand.sql` adds a partial `agent_instruction_copies_cleanup_idx` on the
|
||||
existing table. Its predicate matches discovery and excludes final history.
|
||||
The migration uses `CREATE INDEX CONCURRENTLY IF NOT EXISTS` and the existing
|
||||
nontransactional runner, including interrupted-index repair. Schema generation
|
||||
retains snapshots 0403 through 0407 and preserves all executable migrations.
|
||||
The original draft used 0395; later regeneration history remains recorded below.
|
||||
|
||||
A rolled-back PostgreSQL fixture with 10,000 added completed records and one
|
||||
pending record used this index: one row, three shared-buffer hits, 0.005 ms local
|
||||
execution time. This establishes that this plan skips completed history; it is
|
||||
not a production latency guarantee or constant-cost claim for arbitrary pending
|
||||
work.
|
||||
|
||||
## Verification and failures preserved
|
||||
|
||||
- Full directory suite: 101 passed, including 14 new lifecycle/worker cases,
|
||||
against fresh PostgreSQL with real temporary files (26.68 seconds).
|
||||
- Startup suite: 52 passed, including enabled/disabled registration and removal
|
||||
of this lane from the periodic reaper.
|
||||
- Existing captured instruction and heartbeat release consumers: 47 passed
|
||||
against a fresh PostgreSQL database (17.03 seconds).
|
||||
- Workspace typecheck and build passed. Remote CI/review remain to be recorded.
|
||||
- Migration applied to fresh databases, resumed after a failed application, and
|
||||
runner replay passed. Query-plan fixture rolled back.
|
||||
|
||||
The first focused worker run had 7 passing and 3 failing cases because the test
|
||||
imported `idleWorkSnapshot` from the wrong module. The import now uses its owning
|
||||
`task-admission` module. The first concurrent migration attempt failed because
|
||||
0395 was not yet registered with the nontransactional runner; registration was
|
||||
added, then application resumed successfully.
|
||||
|
||||
The first full owner/startup run had 148 passing and one failing case. The existing
|
||||
reservation-failure test intercepted `db.update`, which no longer receives that
|
||||
write after it moved into a transaction. Its injection now fails the reservation
|
||||
transaction. The same substantive assertions remain: the first receipt and files
|
||||
stay intact, and the next copy cleans successfully. The final 101-case owner run
|
||||
passes this check. New coverage also proves reservation-failure backoff, a later
|
||||
deadline winning the claim, real collection racing recovery with one deletion,
|
||||
unknown remote ownership, and actual worktree setting reactivation.
|
||||
|
||||
Filesystem suites use fresh databases per full invocation: their teardown removes
|
||||
temporary roots, so reused retained rows would refer to deleted fixture paths.
|
||||
The baseline and failure logs are retained under `/tmp/sleep41-*` locally; the PR
|
||||
records the outcomes for reviewers without requiring access to those files.
|
||||
|
||||
|
||||
## Updated parent and current acceptance (2026-10-28)
|
||||
|
||||
The existing duty was merged normally with the verified instruction-commit
|
||||
parent. Only its unpublished 0399 migration artifacts were deferred while
|
||||
adopting parent history, then regenerated through `pnpm db:generate` as
|
||||
`0405_petite_frightful_four.sql`. The concurrent, idempotent index SQL is
|
||||
byte-identical to the original. Every prior SQL and journal entry and every
|
||||
retained parent snapshot is unchanged; the five newest snapshots are 0401-0405.
|
||||
The only snapshot merge conflict was resolved with the exact parent bytes.
|
||||
Owned domain/schema edits and unrelated upstream blobs are preserved.
|
||||
|
||||
Current source `7184877830e2b4f56d61fe7eb6b8c3d92e1353e6` passes **284 cases across six complete files**,
|
||||
with no skips. The verbose case records give these counts:
|
||||
|
||||
| Complete file | Passing cases |
|
||||
| --- | ---: |
|
||||
| src/__tests__/agent-directory-working-copies.test.ts | 101 |
|
||||
| src/services/heartbeat/run-lifecycle.test.ts | 31 |
|
||||
| src/__tests__/agent-instruction-working-copies.test.ts | 38 |
|
||||
| src/services/heartbeat/recovery.test.ts | 45 |
|
||||
| src/__tests__/server-startup-feedback-export.test.ts | 55 |
|
||||
| src/services/heartbeat/instruction-working-copy.test.ts | 14 |
|
||||
|
||||
Instruction copies, directory copies, and heartbeat recovery each use a distinct
|
||||
new disposable PostgreSQL database migrated before the fixtures. Lifecycle
|
||||
fixtures retain real embedded PostgreSQL. Database build, production server
|
||||
typecheck/build, module boundaries, target migration order, and all three fresh
|
||||
0405 migrations pass. The older failed fixtures and timing findings above remain
|
||||
part of the record; no passing retry is used to erase them.
|
||||
|
||||
A populated 0404 parent database upgraded to 0405: all 403 existing migration
|
||||
receipts remained identical and exactly one was appended. Replay added none.
|
||||
The cleanup index is valid and ready on both fresh and upgraded databases.
|
||||
This verifies safe migration and worker behavior, not deployed sleep or Cloud
|
||||
alarm delivery. Final publication still requires the actual landed parent
|
||||
ancestry and fresh current-head review and CI.
|
||||
|
||||
|
||||
## Published parent review cleanup
|
||||
|
||||
The published captured-save parent is now `a1a0ae5b02e8a650408bf4c55ae99071c3bff59c`.
|
||||
Adopting it through a normal merge carries its actual-parent audit and removes
|
||||
seven parent-owned code-comment lines under CONTRIBUTING.md. Every changed code
|
||||
file retains its exact non-comment TypeScript token stream; the entire database
|
||||
source tree is unchanged. This slice adds no code comments relative to that
|
||||
parent. The 284-case acceptance and 0405 upgrade evidence retain their original
|
||||
scope, with no runtime or assertion change hidden as a parent update. Actual
|
||||
landed 0404 ancestry and fresh published-head gates remain required.
|
||||
|
||||
## Current parent migration history and corrected acceptance
|
||||
|
||||
Current parent `a39d6fe84b582ace7b4a6d20312943d65e750d30` carries the actual
|
||||
merged webhook-recovery ancestry, the new Coordinator/Runner changes, and
|
||||
append-only migration 0406. Only this slice's unlanded 0405 artifacts were
|
||||
removed before a normal merge. Original runtime/schema edits and every unrelated
|
||||
upstream blob are preserved. The generated snapshot conflict uses exact parent
|
||||
bytes. The existing index was generated as `0407_greedy_star_brand.sql`, with
|
||||
identical original concurrent/idempotent SQL. All landed SQL files, prior journal
|
||||
entries, and retained snapshots remain unchanged. The five snapshots are
|
||||
0403 through 0407.
|
||||
|
||||
An initial local runner mistakenly replaced its database build with a pnpm
|
||||
filter that matched no projects. Its zero exit was not a build pass. The first
|
||||
fresh migration then failed because compiled migration assets were absent. Both
|
||||
logs and the original command receipt remain. No owner tests ran in that failed
|
||||
attempt. Commands now select the actual database build and current migration
|
||||
order by name rather than position.
|
||||
|
||||
At `e7753d731810ad342468213129ddab4345fb1e4c`, the corrected run passes all
|
||||
**284 cases in six complete files**, with no skips, in 50.01 seconds. Four unused
|
||||
PostgreSQL 17 databases migrated through 0407 isolate the destructive fixtures.
|
||||
The embedded-only lifecycle test temporarily used its own external fixture
|
||||
because this host cannot start more embedded clusters; its fixture import was
|
||||
restored byte for byte. Assertions and original timing limits are unchanged.
|
||||
This is qualified external-database acceptance. Published CI still must pass
|
||||
with the original embedded fixture. The actual database build, server types and
|
||||
build, boundaries, and exact current migration-order check pass.
|
||||
|
||||
A populated 0406-to0407 upgrade preserves every one of the 405 prior migration
|
||||
receipts and all 38 companies, adds exactly one receipt, and adds none on replay.
|
||||
The cleanup index is valid and ready on both fresh and upgraded databases. These
|
||||
results supersede the earlier local 0405 preparation after master's migration
|
||||
history advanced. No new published head or current-head review pass is implied;
|
||||
parent squash ancestry and new published-head gates remain required.
|
||||
|
||||
## Published bounded-parent consumer acceptance
|
||||
|
||||
A normal merge carries the updated instruction-recovery parent and its bounded
|
||||
retry-cancellation dependency. Both complete patches and every unrelated upstream
|
||||
blob are preserved. Source `70a6800eea874bdb87685bf6eea66de79c48b577` passes all
|
||||
**284 cases in six complete files**, without skips, in 48.88 seconds. Four new
|
||||
PostgreSQL 17 databases migrated through 0407 isolate the fixtures. The temporary
|
||||
external lifecycle fixture is restored exactly; all original assertions and
|
||||
limits remain unchanged. Actual database build, types/build, boundaries and exact
|
||||
current migration order pass. The complete 0407 database tree is unchanged from
|
||||
the populated-upgrade proof above.
|
||||
|
||||
Adopting published parent `915ef6f20a68fffe89711cc97a6196521ea9885d` records the
|
||||
actual 0406 parent squash ancestry. The resulting integration
|
||||
`d8edcd131924b1ad744e232df9b548cf02b1e249` changes only the parent's audit;
|
||||
complete server, package, UI and test trees equal the accepted source. This slice
|
||||
remains unpublished. A tree-only probe predicts text conflicts after the pending
|
||||
instruction parent is squashed, so publication waits for its actual merge and
|
||||
normal ancestry reconciliation. The probe is not a live merge proof or a reason
|
||||
to bypass any gate.
|
||||
|
||||
## Actual instruction parent landed; publication ready
|
||||
|
||||
Parent #17670 merged as `b294de546bf9cd3a8578a7a66d7e8ec0eb939799` with all
|
||||
58 checks and both current-head reviews at 5/5. Its entire tree equals the
|
||||
already-adopted reviewed parent `915ef6f20a68fffe89711cc97a6196521ea9885d`.
|
||||
A normal merge records the actual squash ancestry. The eleven predicted text
|
||||
conflicts came from duplicate history; the exact parent tree was already in the
|
||||
accepted branch ancestry. Resolving the inspected conflicts to the accepted
|
||||
versions leaves the **entire merge result tree identical** to the prior
|
||||
`c6f53662f4035a69d41f61f1759674de22f3163b`, including automatically merged files
|
||||
and the 0407 database tree. The integration is
|
||||
`531058e527fe4601958fbbf02ad2e7bbbf78af48`.
|
||||
|
||||
All source covered by the 284-case acceptance remains byte-identical. This final
|
||||
follow-up changes documentation only. The actual target now contains migrations
|
||||
through 0406, and this slice adds only append-only 0407. Fresh review and CI must
|
||||
run on the published head; old draft scores are not current passes. No merge or
|
||||
instance-sleep completion is inferred from publication readiness.
|
||||
@@ -556,6 +556,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
|
||||
"0402_ancient_malice.sql",
|
||||
"0403_natural_cerise.sql",
|
||||
"0406_zippy_war_machine.sql",
|
||||
"0407_greedy_star_brand.sql",
|
||||
|
||||
|
||||
"0384_funny_adam_warlock.sql",
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS "agent_instruction_copies_cleanup_idx" ON "agent_instruction_working_copies" USING btree ("run_id","company_id") WHERE "agent_instruction_working_copies"."state" IN ('saved', 'unchanged', 'resolved', 'unavailable')
|
||||
AND "agent_instruction_working_copies"."process_stopped_at" IS NOT NULL
|
||||
AND "agent_instruction_working_copies"."receipt"->>'schema' = 'paperclip.agent-files.v1'
|
||||
AND coalesce("agent_instruction_working_copies"."receipt"->>'retainedByRunId', '') = ''
|
||||
AND "agent_instruction_working_copies"."receipt"->>'cleanupPending' IS DISTINCT FROM 'false'
|
||||
AND ("agent_instruction_working_copies"."receipt" ? 'baseline' OR "agent_instruction_working_copies"."receipt"->>'cleanupPending' = 'true');
|
||||
+281
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"id": "6ce76555-38d9-4074-873f-db02c546e2c6",
|
||||
"prevId": "c4c8a780-68b0-4b79-8c3b-5847f506ed7c",
|
||||
"id": "856a147c-73dc-4908-a8d9-4f19cab7dbf6",
|
||||
"prevId": "c34fcd91-0d57-4241-b8b4-399bc9f6dcb0",
|
||||
"version": "7",
|
||||
"dialect": "postgresql",
|
||||
"tables": {
|
||||
@@ -2665,6 +2665,28 @@
|
||||
}
|
||||
},
|
||||
"indexes": {
|
||||
"agent_instruction_copies_cleanup_idx": {
|
||||
"name": "agent_instruction_copies_cleanup_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "run_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "company_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"agent_instruction_working_copies\".\"state\" IN ('saved', 'unchanged', 'resolved', 'unavailable')\n AND \"agent_instruction_working_copies\".\"process_stopped_at\" IS NOT NULL\n AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1'\n AND coalesce(\"agent_instruction_working_copies\".\"receipt\"->>'retainedByRunId', '') = ''\n AND \"agent_instruction_working_copies\".\"receipt\"->>'cleanupPending' IS DISTINCT FROM 'false'\n AND (\"agent_instruction_working_copies\".\"receipt\" ? 'baseline' OR \"agent_instruction_working_copies\".\"receipt\"->>'cleanupPending' = 'true')",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"agent_instruction_copies_pending_idx": {
|
||||
"name": "agent_instruction_copies_pending_idx",
|
||||
"columns": [
|
||||
@@ -9689,6 +9711,34 @@
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"chat_actions_github_stale_head_idx": {
|
||||
"name": "chat_actions_github_stale_head_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "company_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "endpoint_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "(\"payload\"->'event'->>'deliveryId')",
|
||||
"asc": true,
|
||||
"isExpression": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"chat_actions\".\"kind\" = 'github_review_check'\n and \"chat_actions\".\"status\" = 'cancelled' and \"chat_actions\".\"result\"->>'code' = 'stale_head'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"chat_actions_github_check_work_idx": {
|
||||
"name": "chat_actions_github_check_work_idx",
|
||||
"columns": [
|
||||
@@ -13209,6 +13259,22 @@
|
||||
}
|
||||
},
|
||||
"indexes": {
|
||||
"chat_github_reviews_projection_work_idx": {
|
||||
"name": "chat_github_reviews_projection_work_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"chat_github_reviews\".\"state\" in ('queued', 'running')\n and \"chat_github_reviews\".\"assessment\" is null\n and \"chat_github_reviews\".\"delivery_id\" ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"chat_github_reviews_delivery_uq": {
|
||||
"name": "chat_github_reviews_delivery_uq",
|
||||
"columns": [
|
||||
@@ -33699,6 +33765,22 @@
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"heartbeat_runs_scheduled_retry_cancellation_idx": {
|
||||
"name": "heartbeat_runs_scheduled_retry_cancellation_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"heartbeat_runs\".\"status\" = 'scheduled_retry' and jsonb_typeof(\"heartbeat_runs\".\"result_json\"->'startupCancellation') = 'object'",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"heartbeat_runs_cost_accounting_pending_idx": {
|
||||
"name": "heartbeat_runs_cost_accounting_pending_idx",
|
||||
"columns": [
|
||||
@@ -39067,6 +39149,140 @@
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.issue_thread_interaction_idempotency_keys": {
|
||||
"name": "issue_thread_interaction_idempotency_keys",
|
||||
"schema": "",
|
||||
"columns": {
|
||||
"id": {
|
||||
"name": "id",
|
||||
"type": "uuid",
|
||||
"primaryKey": true,
|
||||
"notNull": true,
|
||||
"default": "gen_random_uuid()"
|
||||
},
|
||||
"company_id": {
|
||||
"name": "company_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"issue_id": {
|
||||
"name": "issue_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"interaction_id": {
|
||||
"name": "interaction_id",
|
||||
"type": "uuid",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"idempotency_key": {
|
||||
"name": "idempotency_key",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"created_at": {
|
||||
"name": "created_at",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "now()"
|
||||
}
|
||||
},
|
||||
"indexes": {
|
||||
"issue_thread_interaction_idempotency_keys_scope_key_uq": {
|
||||
"name": "issue_thread_interaction_idempotency_keys_scope_key_uq",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "company_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "issue_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
},
|
||||
{
|
||||
"expression": "idempotency_key",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": true,
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"issue_thread_interaction_idempotency_keys_interaction_idx": {
|
||||
"name": "issue_thread_interaction_idempotency_keys_interaction_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "interaction_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
}
|
||||
},
|
||||
"foreignKeys": {
|
||||
"issue_thread_interaction_idempotency_keys_company_id_companies_id_fk": {
|
||||
"name": "issue_thread_interaction_idempotency_keys_company_id_companies_id_fk",
|
||||
"tableFrom": "issue_thread_interaction_idempotency_keys",
|
||||
"tableTo": "companies",
|
||||
"columnsFrom": [
|
||||
"company_id"
|
||||
],
|
||||
"columnsTo": [
|
||||
"id"
|
||||
],
|
||||
"onDelete": "cascade",
|
||||
"onUpdate": "no action"
|
||||
},
|
||||
"issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk": {
|
||||
"name": "issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk",
|
||||
"tableFrom": "issue_thread_interaction_idempotency_keys",
|
||||
"tableTo": "issues",
|
||||
"columnsFrom": [
|
||||
"issue_id"
|
||||
],
|
||||
"columnsTo": [
|
||||
"id"
|
||||
],
|
||||
"onDelete": "cascade",
|
||||
"onUpdate": "no action"
|
||||
},
|
||||
"issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk": {
|
||||
"name": "issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk",
|
||||
"tableFrom": "issue_thread_interaction_idempotency_keys",
|
||||
"tableTo": "issue_thread_interactions",
|
||||
"columnsFrom": [
|
||||
"interaction_id"
|
||||
],
|
||||
"columnsTo": [
|
||||
"id"
|
||||
],
|
||||
"onDelete": "cascade",
|
||||
"onUpdate": "no action"
|
||||
}
|
||||
},
|
||||
"compositePrimaryKeys": {},
|
||||
"uniqueConstraints": {},
|
||||
"policies": {},
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.issue_thread_interactions": {
|
||||
"name": "issue_thread_interactions",
|
||||
"schema": "",
|
||||
@@ -59919,6 +60135,69 @@
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.user_workspace_preferences": {
|
||||
"name": "user_workspace_preferences",
|
||||
"schema": "",
|
||||
"columns": {
|
||||
"id": {
|
||||
"name": "id",
|
||||
"type": "uuid",
|
||||
"primaryKey": true,
|
||||
"notNull": true,
|
||||
"default": "gen_random_uuid()"
|
||||
},
|
||||
"user_id": {
|
||||
"name": "user_id",
|
||||
"type": "text",
|
||||
"primaryKey": false,
|
||||
"notNull": true
|
||||
},
|
||||
"coordinator_workspace_enabled": {
|
||||
"name": "coordinator_workspace_enabled",
|
||||
"type": "boolean",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": false
|
||||
},
|
||||
"created_at": {
|
||||
"name": "created_at",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "now()"
|
||||
},
|
||||
"updated_at": {
|
||||
"name": "updated_at",
|
||||
"type": "timestamp with time zone",
|
||||
"primaryKey": false,
|
||||
"notNull": true,
|
||||
"default": "now()"
|
||||
}
|
||||
},
|
||||
"indexes": {
|
||||
"user_workspace_preferences_user_uq": {
|
||||
"name": "user_workspace_preferences_user_uq",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "user_id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": true,
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
}
|
||||
},
|
||||
"foreignKeys": {},
|
||||
"compositePrimaryKeys": {},
|
||||
"uniqueConstraints": {},
|
||||
"policies": {},
|
||||
"checkConstraints": {},
|
||||
"isRLSEnabled": false
|
||||
},
|
||||
"public.work_assessments": {
|
||||
"name": "work_assessments",
|
||||
"schema": "",
|
||||
+7
@@ -2836,6 +2836,13 @@
|
||||
"when": 1793235984352,
|
||||
"tag": "0406_zippy_war_machine",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 407,
|
||||
"version": "7",
|
||||
"when": 1793236707492,
|
||||
"tag": "0407_greedy_star_brand",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import {
|
||||
pgTable,
|
||||
uuid,
|
||||
@@ -113,6 +114,12 @@ export const agentInstructionWorkingCopies = pgTable(
|
||||
},
|
||||
(t) => ({
|
||||
owner: foreignKey({ columns: [t.companyId, t.agentId], foreignColumns: [agents.companyId, agents.id] }).onDelete("cascade"),
|
||||
directoryCleanup: index("agent_instruction_copies_cleanup_idx").on(t.runId, t.companyId).where(sql`${t.state} IN ('saved', 'unchanged', 'resolved', 'unavailable')
|
||||
AND ${t.processStoppedAt} IS NOT NULL
|
||||
AND ${t.receipt}->>'schema' = 'paperclip.agent-files.v1'
|
||||
AND coalesce(${t.receipt}->>'retainedByRunId', '') = ''
|
||||
AND ${t.receipt}->>'cleanupPending' IS DISTINCT FROM 'false'
|
||||
AND (${t.receipt} ? 'baseline' OR ${t.receipt}->>'cleanupPending' = 'true')`),
|
||||
pending: index("agent_instruction_copies_pending_idx").on(t.state, t.nextAttemptAt),
|
||||
history: index("agent_instruction_copies_agent_idx").on(t.companyId, t.agentId, t.createdAt),
|
||||
}),
|
||||
|
||||
@@ -1,3 +1,8 @@
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { notifyDirectoryCleanup } from "../services/directory-cleanup-notifications.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import fs from "node:fs/promises";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
@@ -8,7 +13,7 @@ const execFile = promisify(execFileCallback);
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agentFileStore, fileHash, inspectAgentFile, snapshotAgentFiles, MAX_AGENT_FILE_BYTES, MAX_AGENT_DIRECTORY_BYTES, MAX_AGENT_DIRECTORY_ENTRIES } from "../services/agent-file-store.js";
|
||||
import { agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, environmentLeases, environments, agentInstructionWorkingCopies, agentInstructionRevisions, agentInstructionHeads, createDb } from "@paperclipai/db";
|
||||
@@ -78,6 +83,274 @@ describe("persistent agent directories", () => {
|
||||
await fs.writeFile(path.join(root, entryFile), initial);
|
||||
});
|
||||
|
||||
describe("post-save directory cleanup worker", () => {
|
||||
afterEach(async () => {
|
||||
vi.useRealTimers(); vi.restoreAllMocks();
|
||||
await db.update(agentInstructionWorkingCopies).set({ receipt: { schema: "paperclip.agent-files.v1", cleanupPending: false } })
|
||||
.where(eq(agentInstructionWorkingCopies.companyId, companyId));
|
||||
});
|
||||
const clock = () => vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
const coordinator = (canRun = () => true, canReconcile = () => true, onError = (_error: unknown) => {}) =>
|
||||
createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile, onError });
|
||||
async function stopped(copy: NonNullable<Awaited<ReturnType<typeof copies.get>>>, dueAt = new Date(), owner = db) {
|
||||
await owner.transaction(async tx => {
|
||||
await notifyDirectoryCleanup(tx, { companyId: copy.companyId, runId: copy.runId });
|
||||
await tx.update(agentInstructionWorkingCopies).set({ state: "saved", processStoppedAt: new Date(), nextAttemptAt: dueAt })
|
||||
.where(eq(agentInstructionWorkingCopies.runId, copy.runId));
|
||||
});
|
||||
}
|
||||
async function released(runId: string) {
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
expect((await copies.get(companyId, runId))?.receipt?.cleanupPending).toBe(false);
|
||||
}, { timeout: 5000 });
|
||||
}
|
||||
it("leaves active, transferred, and completed receipts quiet even with retained baselines", async () => {
|
||||
const active = await run(), completed = await run(), transferred = await run();
|
||||
await db.update(agentInstructionWorkingCopies).set({ state: "saved", processStoppedAt: new Date(),
|
||||
receipt: { ...completed.receipt, cleanupPending: false } }).where(eq(agentInstructionWorkingCopies.runId, completed.runId));
|
||||
await db.update(agentInstructionWorkingCopies).set({ state: "saved", processStoppedAt: new Date(),
|
||||
receipt: { ...transferred.receipt, retainedByRunId: active.runId } }).where(eq(agentInstructionWorkingCopies.runId, transferred.runId));
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
const reads = vi.spyOn(db, "select"), removes = vi.spyOn(fs, "rm");
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); expect(removes).not.toHaveBeenCalled();
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
for (const copy of [active, completed, transferred]) expect((await fs.stat(copy.localRoot)).isDirectory()).toBe(true);
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it("reactivates from real collection, preserves saved files, retries failed cleanup, and becomes quiet", async () => {
|
||||
const copy = await run(); await fs.writeFile(path.join(copy.localRoot, "memory.txt"), "saved memory");
|
||||
clock(); const worker = coordinator();
|
||||
const remove = fs.rm.bind(fs);
|
||||
let failing = true;
|
||||
const removes = vi.spyOn(fs, "rm").mockImplementation(async (file, opts) => {
|
||||
if (failing && file === path.dirname(copy.localRoot)) throw new Error("temporary cleanup failure");
|
||||
return remove(file, opts);
|
||||
});
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
expect((await copies.collectStopped({ companyId, runId: copy.runId }))?.state).toBe("saved");
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
expect((await copies.get(companyId, copy.runId))?.nextAttemptAt).toBeInstanceOf(Date);
|
||||
});
|
||||
const pending = (await copies.get(companyId, copy.runId))!;
|
||||
expect(pending.receipt?.cleanupPending).toBe(true);
|
||||
expect(await fs.readFile(path.join(root, "memory.txt"), "utf8")).toBe("saved memory");
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBe(pending.nextAttemptAt!.getTime());
|
||||
failing = false;
|
||||
await vi.advanceTimersByTimeAsync(30_000); await released(copy.runId);
|
||||
await expect(fs.stat(copy.localRoot)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
const reads = vi.spyOn(db, "select"); removes.mockClear();
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); expect(removes).not.toHaveBeenCalled();
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
} finally { removes.mockRestore(); await worker.stop(); }
|
||||
});
|
||||
it.each(["restart", "drain", "standby"] as const)("preserves cleanup across %s", async mode => {
|
||||
const copy = await run();
|
||||
if (mode === "restart") await stopped(copy);
|
||||
let admitted = mode === "restart";
|
||||
clock(); const worker = coordinator(() => admitted, () => mode !== "standby");
|
||||
try {
|
||||
await agentInstructionWorkingCopyService(db).registerDirectoryCleanupWork(worker, () => admitted, async () => true).ready;
|
||||
if (mode !== "restart") {
|
||||
await stopped(copy); await vi.advanceTimersByTimeAsync(1000);
|
||||
if (mode === "drain") await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
const reads = vi.spyOn(db, "select");
|
||||
await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
expect((await fs.stat(copy.localRoot)).isDirectory()).toBe(true);
|
||||
admitted = true; await vi.advanceTimersByTimeAsync(30_000);
|
||||
}
|
||||
await released(copy.runId);
|
||||
await expect(fs.stat(copy.localRoot)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it("retains a future cleanup deadline across restart without early file operations", async () => {
|
||||
const copy = await run(); clock(); const due = new Date(Date.now() + 90_000);
|
||||
await stopped(copy, due); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
expect(worker.nextWakeAt()).toBe(due.getTime());
|
||||
const removes = vi.spyOn(fs, "rm"); await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(removes).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(30_000); await released(copy.runId);
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it("honors outer rollback and only releases after the outer commit", async () => {
|
||||
const copy = await run(); clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
await expect(db.transaction(async tx => {
|
||||
await stopped(copy, new Date(), tx as unknown as typeof db);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect((await fs.stat(copy.localRoot)).isDirectory()).toBe(true);
|
||||
throw new Error("rollback cleanup");
|
||||
})).rejects.toThrow("rollback cleanup");
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect((await copies.get(companyId, copy.runId))?.state).toBe("prepared");
|
||||
await db.transaction(async tx => {
|
||||
await stopped(copy, new Date(), tx as unknown as typeof db);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect((await fs.stat(copy.localRoot)).isDirectory()).toBe(true);
|
||||
});
|
||||
await released(copy.runId);
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it("joins destructive cleanup on shutdown and restores quiet after restart", async () => {
|
||||
const copy = await run(); await stopped(copy); clock(); const worker = coordinator();
|
||||
let release!: () => void, entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const entry = new Promise<void>(resolve => { entered = resolve; });
|
||||
const remove = fs.rm.bind(fs);
|
||||
const removes = vi.spyOn(fs, "rm").mockImplementation(async (file, opts) => {
|
||||
if (file === path.dirname(copy.localRoot)) { entered(); await gate; }
|
||||
return remove(file, opts);
|
||||
});
|
||||
let stopping: Promise<void> | undefined;
|
||||
try {
|
||||
const registration = copies.registerDirectoryCleanupWork(worker, () => true, async () => true);
|
||||
await entry; let finished = false;
|
||||
stopping = worker.stop().then(() => { finished = true; });
|
||||
await vi.advanceTimersByTimeAsync(1000); expect(finished).toBe(false);
|
||||
release(); await stopping; await registration.ready;
|
||||
expect((await copies.get(companyId, copy.runId))?.receipt?.cleanupPending).toBe(false);
|
||||
const restarted = coordinator();
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(restarted, () => true, async () => true).ready;
|
||||
expect(restarted.nextWakeAt()).toBeNull();
|
||||
} finally { await restarted.stop(); }
|
||||
} finally { release(); removes.mockRestore(); await stopping; await worker.stop(); }
|
||||
});
|
||||
it("drains more than one discovery page in bounded execution batches", async () => {
|
||||
const rows = [];
|
||||
for (let index = 0; index < 103; index++) rows.push(await run());
|
||||
for (const copy of rows) await stopped(copy);
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
const first = await db.select().from(agentInstructionWorkingCopies).where(eq(agentInstructionWorkingCopies.companyId, companyId));
|
||||
expect(first.filter(row => row.receipt?.cleanupPending === false)).toHaveLength(20);
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
const all = await db.select().from(agentInstructionWorkingCopies).where(eq(agentInstructionWorkingCopies.companyId, companyId));
|
||||
expect(all.every(row => row.receipt?.cleanupPending === false)).toBe(true);
|
||||
}, { timeout: 10_000 });
|
||||
for (const row of rows) await expect(fs.stat(row.localRoot)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally { await worker.stop(); }
|
||||
}, 30_000);
|
||||
it("honors a newer deadline committed before the cleanup claim", async () => {
|
||||
const copy = await run(); await stopped(copy); clock(); const worker = coordinator();
|
||||
const later = new Date(Date.now() + 60_000), transaction = db.transaction.bind(db);
|
||||
let changed = false;
|
||||
const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => {
|
||||
if (!changed) {
|
||||
changed = true;
|
||||
await transaction(async tx => {
|
||||
await notifyDirectoryCleanup(tx, { companyId, runId: copy.runId });
|
||||
await tx.update(agentInstructionWorkingCopies).set({ nextAttemptAt: later })
|
||||
.where(eq(agentInstructionWorkingCopies.runId, copy.runId));
|
||||
});
|
||||
}
|
||||
return transaction(callback, options);
|
||||
}) as typeof db.transaction);
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
transactions.mockRestore();
|
||||
expect(changed).toBe(true);
|
||||
expect((await fs.stat(copy.localRoot)).isDirectory()).toBe(true);
|
||||
expect((await copies.get(companyId, copy.runId))?.nextAttemptAt).toEqual(later);
|
||||
await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBe(later.getTime());
|
||||
await vi.advanceTimersByTimeAsync(60_000); await released(copy.runId);
|
||||
} finally { transactions.mockRestore(); await worker.stop(); }
|
||||
});
|
||||
it("backs off a failed reservation without losing cleanup or blocking another owner", async () => {
|
||||
const first = await run(), second = await run();
|
||||
clock(); await stopped(first, new Date(Date.now() - 1000)); await stopped(second);
|
||||
const errors: unknown[] = [], worker = coordinator(() => true, () => true, error => errors.push(error));
|
||||
const transaction = vi.spyOn(db, "transaction").mockImplementationOnce(async () => { throw new Error("cleanup reservation unavailable"); });
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
transaction.mockRestore();
|
||||
expect(errors).toHaveLength(1);
|
||||
expect((await fs.stat(first.localRoot)).isDirectory()).toBe(true);
|
||||
expect((await copies.get(companyId, second.runId))?.receipt?.cleanupPending).toBe(false);
|
||||
await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(15_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await vi.advanceTimersByTimeAsync(30_000); await released(first.runId);
|
||||
} finally { transaction.mockRestore(); await worker.stop(); }
|
||||
});
|
||||
it("shares the directory lock with foreground collection and deletes only once", async () => {
|
||||
const copy = await run(); await fs.writeFile(path.join(copy.localRoot, "memory.txt"), "last save");
|
||||
clock(); const worker = coordinator();
|
||||
let release!: () => void, entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const entry = new Promise<void>(resolve => { entered = resolve; });
|
||||
const remove = fs.rm.bind(fs); let removals = 0;
|
||||
const removes = vi.spyOn(fs, "rm").mockImplementation(async (file, options) => {
|
||||
if (file === path.dirname(copy.localRoot)) { removals += 1; entered(); await gate; }
|
||||
return remove(file, options);
|
||||
});
|
||||
let foreground: ReturnType<typeof copies.collectStopped> | undefined;
|
||||
try {
|
||||
await copies.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
foreground = copies.collectStopped({ companyId, runId: copy.runId });
|
||||
await entry;
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
expect((await copies.get(companyId, copy.runId))?.nextAttemptAt).toBeInstanceOf(Date);
|
||||
});
|
||||
release(); await foreground; await released(copy.runId);
|
||||
await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(removals).toBe(1);
|
||||
expect(await fs.readFile(path.join(root, "memory.txt"), "utf8")).toBe("last save");
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
} finally { release(); await foreground; removes.mockRestore(); await worker.stop(); }
|
||||
});
|
||||
it("reactivates from the real worktree setting without polling disabled policy", async () => {
|
||||
const copy = await run(); await stopped(copy);
|
||||
const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "sleep41-directory-activation" };
|
||||
const settings = instanceSettingsService(db, { runtimeEnv });
|
||||
await settings.updateExperimental({ enableWorktreeRunExecution: false });
|
||||
const service = heartbeatService(db, { runtimeEnv });
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await service.registerInstructionDirectoryCleanupWork(worker, () => true).ready;
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await service.resolveSchedulingSuppression();
|
||||
await settings.updateExperimental({ enableWorktreeRunExecution: true });
|
||||
await released(copy.runId);
|
||||
} finally { await worker.stop(); await settings.updateExperimental({ enableWorktreeRunExecution: false }); }
|
||||
});
|
||||
it("retains unknown remote cleanup without acquiring a replacement environment", async () => {
|
||||
const copy = await run();
|
||||
await db.update(agentInstructionWorkingCopies).set({ location: `remote:${randomUUID()}` })
|
||||
.where(eq(agentInstructionWorkingCopies.runId, copy.runId));
|
||||
await stopped(copy); clock(); const worker = coordinator();
|
||||
const execute = vi.fn();
|
||||
const service = agentInstructionWorkingCopyService(db, { environmentRuntime: { execute } as unknown as EnvironmentRuntimeService });
|
||||
try {
|
||||
await service.registerDirectoryCleanupWork(worker, () => true, async () => true).ready;
|
||||
const pending = (await copies.get(companyId, copy.runId))!;
|
||||
expect(pending.receipt?.cleanupPending).toBe(true); expect(pending.nextAttemptAt).toBeInstanceOf(Date);
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
await expect.poll(async () => (await copies.get(companyId, copy.runId))!.nextAttemptAt!.getTime()).toBeGreaterThan(pending.nextAttemptAt!.getTime());
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
});
|
||||
|
||||
describe("warm directory ownership", () => {
|
||||
async function nextRun() {
|
||||
const runId = randomUUID();
|
||||
@@ -1173,7 +1446,7 @@ describe("persistent agent directories", () => {
|
||||
for (const runId of deferred) expect(await copies.get(companyId, runId)).toMatchObject({ state: "unavailable", processStoppedAt: null, attempts: 3 });
|
||||
});
|
||||
|
||||
it("continues cleanup when both retry-reservation writes fail for the first copy", async () => {
|
||||
it("continues cleanup when the retry reservation fails for the first copy", async () => {
|
||||
const first = await run(), second = await run();
|
||||
for (const [index, copy] of [first, second].entries()) {
|
||||
await db.update(agentInstructionWorkingCopies).set({ state: "unavailable", processStoppedAt: new Date(), updatedAt: new Date(index),
|
||||
@@ -1182,7 +1455,7 @@ describe("persistent agent directories", () => {
|
||||
}
|
||||
const before = await copies.get(companyId, first.runId);
|
||||
const failReservation = () => { throw new Error("fixture retry-reservation write failure"); };
|
||||
const update = vi.spyOn(db, "update").mockImplementationOnce(failReservation).mockImplementationOnce(failReservation);
|
||||
const update = vi.spyOn(db, "transaction").mockImplementationOnce(failReservation);
|
||||
try { await copies.recoverCaptured(); } finally { update.mockRestore(); }
|
||||
expect(await copies.get(companyId, first.runId)).toEqual(before);
|
||||
expect(await fs.readFile(path.join(first.localRoot, entryFile), "utf8")).toBe(initial);
|
||||
|
||||
@@ -115,6 +115,7 @@ const {
|
||||
reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })),
|
||||
reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })),
|
||||
registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
registerInstructionDirectoryCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
registerCapturedInstructionCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })),
|
||||
resumeQueuedRuns: vi.fn(async () => undefined),
|
||||
@@ -551,6 +552,13 @@ describe("startServer feedback export wiring", () => {
|
||||
process.env.BETTER_AUTH_SECRET = "test-secret";
|
||||
});
|
||||
|
||||
it.each([false, true])("registers directory cleanup recovery only when maintenance is enabled (%s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
|
||||
await startServer(); await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.registerInstructionDirectoryCleanupWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
|
||||
if (enabled) expect(heartbeatServiceMock.registerInstructionDirectoryCleanupWork.mock.calls[0][1]()).toBe(true);
|
||||
});
|
||||
|
||||
it.each([false, true])("registers captured instruction recovery only when maintenance is enabled (%s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
|
||||
await startServer(); await settleStartupHousekeeping();
|
||||
@@ -949,7 +957,7 @@ describe("startServer feedback export wiring", () => {
|
||||
intervalCallback?.();
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
|
||||
expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true });
|
||||
expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true });
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(retiredDetector).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
|
||||
+3
-1
@@ -1822,6 +1822,8 @@ async function startServerWithDatabaseTeardown(
|
||||
|
||||
if (config.heartbeatSchedulerEnabled) await heartbeat.registerCapturedInstructionCommitWork(deliveryWork,
|
||||
() => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
|
||||
if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork,
|
||||
() => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
|
||||
|
||||
const runRetentionSweep = async () => {
|
||||
const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active"));
|
||||
@@ -1950,7 +1952,7 @@ async function startServerWithDatabaseTeardown(
|
||||
// Periodically reap orphaned runs (5-min staleness threshold) and make sure
|
||||
// persisted queued work is still being driven forward.
|
||||
runPeriodicRecovery(() => heartbeat
|
||||
.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true })
|
||||
.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true })
|
||||
.then(() => heartbeat.promoteDueScheduledRetries())
|
||||
.then(async (promotion) => {
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js";
|
||||
import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
|
||||
import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
|
||||
import { and, eq, or, sql } from "drizzle-orm";
|
||||
@@ -46,6 +47,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str
|
||||
await tx.delete(issueComments).where(eq(issueComments.authorAgentId, id));
|
||||
await notifyScheduledRetryCancellation(tx, { companyId: companyId });
|
||||
await notifyInstructionCommit(tx, { companyId: companyId });
|
||||
await notifyDirectoryCleanup(tx, { companyId: companyId });
|
||||
await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, id));
|
||||
await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, id));
|
||||
await tx.delete(agentApiKeys).where(eq(agentApiKeys.agentId, id));
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyDirectoryCleanup, type DirectoryCleanupScope } from "./directory-cleanup-notifications.js";
|
||||
import { notifyInstructionCommit, type InstructionCommitScope } from "./instruction-commit-notifications.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
@@ -8,7 +9,7 @@ import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { and, asc, desc, eq, gt, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, gt, inArray, lte, or, isNull, sql } from "drizzle-orm";
|
||||
import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, databaseWorkScopeKey, type Db } from "@paperclipai/db";
|
||||
import {
|
||||
prepareAdapterExecutionTargetRuntime,
|
||||
@@ -95,10 +96,12 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
return row ?? null;
|
||||
}
|
||||
async function tryPatch(row: Copy, values: Partial<typeof copies.$inferInsert>, requireDue = false) {
|
||||
const signalsCleanup = directoryCleanupEntry(row) !== null || directoryCleanupEntry({ ...row, ...values }) !== null;
|
||||
const signalsCommit = row.state === "pending_commit" || values.state === "pending_commit";
|
||||
const write = async (tx: Db) => {
|
||||
if (signalsCommit)
|
||||
await notifyInstructionCommit(tx, { companyId: row.companyId, runId: row.runId });
|
||||
if (signalsCleanup) await notifyDirectoryCleanup(tx, { companyId: row.companyId, runId: row.runId });
|
||||
const [updated] = await tx.update(copies).set({ ...values, updatedAt: new Date() }).where(and(
|
||||
scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash),
|
||||
sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`,
|
||||
@@ -106,7 +109,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
)).returning();
|
||||
return updated;
|
||||
};
|
||||
return signalsCommit ? db.transaction(tx => write(tx as unknown as Db)) : write(db);
|
||||
return signalsCommit || signalsCleanup ? db.transaction(tx => write(tx as unknown as Db)) : write(db);
|
||||
}
|
||||
async function patch(row: Copy, values: Partial<typeof copies.$inferInsert>) {
|
||||
const updated = await tryPatch(row, values);
|
||||
@@ -335,7 +338,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
}
|
||||
|
||||
/** Restarts can retry persisted bytes without touching a live filesystem or launching a model. */
|
||||
async function recoverCaptured(options: { skipCommit?: boolean } = {}) {
|
||||
async function recoverCaptured(options: { skipCommit?: boolean; skipDirectoryCleanup?: boolean } = {}) {
|
||||
const pending = options.skipCommit ? [] : await db.select().from(copies).where(and(
|
||||
eq(copies.state, "pending_commit"),
|
||||
or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())),
|
||||
@@ -344,13 +347,16 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
for (const row of pending) await recoverCommit(commitEntry(row)!, () => true);
|
||||
// A crash between a durable save and release must not retain run snapshots
|
||||
// forever. Only discard copies with recorded stop proof and a terminal receipt.
|
||||
const cleanup = await db.select().from(copies).where(and(
|
||||
inArray(copies.state, [...completed, "unavailable"]), isNotNull(copies.processStoppedAt),
|
||||
sql`${copies.receipt}->>'schema' = 'paperclip.agent-files.v1'`,
|
||||
or(sql`${copies.receipt} ? 'baseline'`, sql`${copies.receipt}->>'cleanupPending' = 'true'`),
|
||||
or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())),
|
||||
)).orderBy(asc(copies.updatedAt)).limit(20);
|
||||
for (const row of cleanup) await recoverDirectoryCleanup(row, () => directories.release(row));
|
||||
if (!options.skipDirectoryCleanup) {
|
||||
const cleanup = await db.select({ id: copies.runId, companyId: copies.companyId,
|
||||
dueAt: copies.nextAttemptAt }).from(copies).where(and(directoryCleanupPredicate(),
|
||||
or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())),
|
||||
)).orderBy(asc(copies.updatedAt)).limit(20);
|
||||
for (const row of cleanup) {
|
||||
try { await recoverDirectoryRelease({ ...row, dueAt: row.dueAt?.getTime() ?? 0 }, () => true); }
|
||||
catch (error) { logger.warn({ err: error, runId: row.id }, "Agent file cleanup deferred; save receipt unchanged"); }
|
||||
}
|
||||
}
|
||||
return pending.length;
|
||||
}
|
||||
|
||||
@@ -406,6 +412,62 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
return pending.length;
|
||||
}
|
||||
|
||||
type DirectoryCleanupEntry = { id: string; companyId: string; dueAt: number };
|
||||
function directoryCleanupEntry(row: Copy | null): DirectoryCleanupEntry | null {
|
||||
return row && ["saved", "unchanged", "resolved", "unavailable"].includes(row.state)
|
||||
&& row.processStoppedAt && isAgentDirectoryCopy(row) && !row.receipt?.retainedByRunId
|
||||
&& row.receipt?.cleanupPending !== false
|
||||
&& (Object.hasOwn(row.receipt!, "baseline") || row.receipt?.cleanupPending === true)
|
||||
? { id: row.runId, companyId: row.companyId, dueAt: row.nextAttemptAt?.getTime() ?? 0 } : null;
|
||||
}
|
||||
function directoryCleanupPredicate() {
|
||||
return sql`${copies.state} IN ('saved', 'unchanged', 'resolved', 'unavailable')
|
||||
AND ${copies.processStoppedAt} IS NOT NULL
|
||||
AND ${copies.receipt}->>'schema' = 'paperclip.agent-files.v1'
|
||||
AND coalesce(${copies.receipt}->>'retainedByRunId', '') = ''
|
||||
AND ${copies.receipt}->>'cleanupPending' IS DISTINCT FROM 'false'
|
||||
AND (${copies.receipt} ? 'baseline' OR ${copies.receipt}->>'cleanupPending' = 'true')`;
|
||||
}
|
||||
async function loadDirectoryCleanup(scope: DirectoryCleanupScope | undefined, admitted: () => boolean) {
|
||||
const entries: DirectoryCleanupEntry[] = [];
|
||||
let cursor: string | undefined;
|
||||
while (admitted()) {
|
||||
const rows = await db.select({ id: copies.runId, companyId: copies.companyId, nextAttemptAt: copies.nextAttemptAt })
|
||||
.from(copies).where(and(directoryCleanupPredicate(),
|
||||
scope ? eq(copies.companyId, scope.companyId) : undefined,
|
||||
scope?.runId ? eq(copies.runId, scope.runId) : undefined,
|
||||
cursor ? gt(copies.runId, cursor) : undefined,
|
||||
)).orderBy(asc(copies.runId)).limit(100);
|
||||
for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
|
||||
if (rows.length < 100) return entries;
|
||||
cursor = rows[rows.length - 1].id;
|
||||
}
|
||||
throw new Error("Directory cleanup reconstruction stopped");
|
||||
}
|
||||
async function recoverDirectoryRelease(entry: DirectoryCleanupEntry, admitted: () => boolean) {
|
||||
const row = await get(entry.companyId, entry.id);
|
||||
const current = directoryCleanupEntry(row);
|
||||
if (!row || !current || current.dueAt > Date.now()) return current;
|
||||
if (!admitted()) throw new Error("Directory cleanup recovery stopped");
|
||||
const claimed = await tryPatch(row, { nextAttemptAt: new Date(Date.now() + 30_000) }, true);
|
||||
if (claimed) await directories.release(claimed);
|
||||
const remaining = directoryCleanupEntry(await get(entry.companyId, entry.id));
|
||||
if (remaining && remaining.dueAt <= Date.now()) throw new Error("Directory cleanup remains unresolved without a future retry");
|
||||
return remaining;
|
||||
}
|
||||
function registerDirectoryCleanupWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
canRun: () => boolean, canRecover: () => Promise<boolean>) {
|
||||
return registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.directoryCleanup,
|
||||
batchSize: 20,
|
||||
decodeScope: (hint: string): DirectoryCleanupScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
|
||||
matches: (entry: DirectoryCleanupEntry, scope?: DirectoryCleanupScope) => !scope ||
|
||||
entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId),
|
||||
load: async (scope, signal) => await canRecover() ? loadDirectoryCleanup(scope, () => canRun() && !signal?.aborted) : [],
|
||||
check: async (entry, signal) => await canRecover() ? recoverDirectoryRelease(entry, () => canRun() && !signal?.aborted) : null,
|
||||
}, canRun);
|
||||
}
|
||||
|
||||
type CommitEntry = { id: string; companyId: string; dueAt: number };
|
||||
function commitEntry(row: Pick<Copy, "runId" | "companyId" | "state" | "attempts" | "nextAttemptAt"> | null): CommitEntry | null {
|
||||
return row?.state === "pending_commit" && row.attempts < MAX_COLLECTION_ATTEMPTS
|
||||
@@ -504,5 +566,5 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
}
|
||||
return row;
|
||||
}
|
||||
return { registerCapturedCommitWork, prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
|
||||
return { registerDirectoryCleanupWork, registerCapturedCommitWork, prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js";
|
||||
import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
|
||||
import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
|
||||
import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-work-notifications.js";
|
||||
@@ -100,6 +101,7 @@ export async function deleteCompany(db: Db, id: string) {
|
||||
await tx.delete(runIdentityContexts).where(eq(runIdentityContexts.companyId, id));
|
||||
await notifyScheduledRetryCancellation(tx, { companyId: id });
|
||||
await notifyInstructionCommit(tx, { companyId: id });
|
||||
await notifyDirectoryCleanup(tx, { companyId: id });
|
||||
await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, id));
|
||||
await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, id));
|
||||
await tx.delete(agentApiKeys).where(eq(agentApiKeys.companyId, id));
|
||||
|
||||
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
|
||||
/** Topics for existing durable queues; register intent before their writes. */
|
||||
export const DELIVERY_QUEUES = {
|
||||
directoryCleanup: "directory-cleanup",
|
||||
instructionCommit: "instruction-commit",
|
||||
scheduledRetryCancellation: "scheduled-retry-cancellation",
|
||||
executionFinalization: "execution-finalization",
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type DirectoryCleanupScope = { companyId: string; runId?: string };
|
||||
|
||||
export async function notifyDirectoryCleanup(transaction: object, scope?: DirectoryCleanupScope) {
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.directoryCleanup, scope ? JSON.stringify(scope) : "*");
|
||||
}
|
||||
@@ -4084,6 +4084,14 @@ export function heartbeatService(
|
||||
triggerIssueMonitor,
|
||||
|
||||
reportRunActivity: clearDetachedRunWarning,
|
||||
registerInstructionDirectoryCleanupWork: (
|
||||
coordinator: Parameters<typeof instructionCopies.registerDirectoryCleanupWork>[0], canRun: () => boolean,
|
||||
) => instructionCopies.registerDirectoryCleanupWork(coordinator,
|
||||
() => canRun() && readTaskDrain(new Date()) === null,
|
||||
async () => {
|
||||
const state = await getSchedulingSuppression(true);
|
||||
return !state.suppressed || state.reason === "task_drain";
|
||||
}),
|
||||
registerCapturedInstructionCommitWork: (
|
||||
coordinator: Parameters<typeof instructionCopies.registerCapturedCommitWork>[0], canRun: () => boolean,
|
||||
) => instructionCopies.registerCapturedCommitWork(coordinator,
|
||||
|
||||
@@ -1947,7 +1947,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
|
||||
let settledProcessesDiscovered = false;
|
||||
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean }) {
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean }) {
|
||||
await preserveCancelledNativeWorkspaces(db);
|
||||
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
||||
const now = new Date();
|
||||
@@ -1957,7 +1957,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
await instructionCopies.recoverStopped().catch(error => {
|
||||
logger.warn({ err: error }, "failed to recover stopped instruction copies");
|
||||
});
|
||||
await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit }).catch(error => {
|
||||
await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit, skipDirectoryCleanup: opts?.skipInstructionDirectoryCleanup }).catch(error => {
|
||||
logger.warn({ err: error }, "failed to retry captured instruction revisions");
|
||||
});
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js";
|
||||
import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
|
||||
import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
|
||||
import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js";
|
||||
@@ -618,7 +619,10 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
|
||||
if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
|
||||
if (patch.enableWorktreeRunExecution !== undefined) await notifyInstructionCommit(tx);
|
||||
if (patch.enableWorktreeRunExecution !== undefined) {
|
||||
await notifyInstructionCommit(tx);
|
||||
await notifyDirectoryCleanup(tx);
|
||||
}
|
||||
if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx);
|
||||
await notifyToolConnectionHealth(tx, "*");
|
||||
return tx
|
||||
|
||||
Reference in new issue
Block a user