diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 4b0c0f25e8e..5cd374773d8 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -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. diff --git a/doc/plans/2026-10-25-directory-cleanup-sleep-audit.md b/doc/plans/2026-10-25-directory-cleanup-sleep-audit.md new file mode 100644 index 00000000000..9b5b670e108 --- /dev/null +++ b/doc/plans/2026-10-25-directory-cleanup-sleep-audit.md @@ -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. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 06becd2eba7..6a44691da1c 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -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", diff --git a/packages/db/src/migrations/0407_greedy_star_brand.sql b/packages/db/src/migrations/0407_greedy_star_brand.sql new file mode 100644 index 00000000000..3561ba967dd --- /dev/null +++ b/packages/db/src/migrations/0407_greedy_star_brand.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'); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0402_snapshot.json b/packages/db/src/migrations/meta/0407_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0402_snapshot.json rename to packages/db/src/migrations/meta/0407_snapshot.json index c4d702a1cc5..54cb791facf 100644 --- a/packages/db/src/migrations/meta/0402_snapshot.json +++ b/packages/db/src/migrations/meta/0407_snapshot.json @@ -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": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index b032c1891e4..cc5297bfcae 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -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 } ] } \ No newline at end of file diff --git a/packages/db/src/schema/agent_instruction_revisions.ts b/packages/db/src/schema/agent_instruction_revisions.ts index 39db9af7237..f13e13e9910 100644 --- a/packages/db/src/schema/agent_instruction_revisions.ts +++ b/packages/db/src/schema/agent_instruction_revisions.ts @@ -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), }), diff --git a/server/src/__tests__/agent-directory-working-copies.test.ts b/server/src/__tests__/agent-directory-working-copies.test.ts index 65353060433..a88c87121d3 100644 --- a/server/src/__tests__/agent-directory-working-copies.test.ts +++ b/server/src/__tests__/agent-directory-working-copies.test.ts @@ -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>>, 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(resolve => { release = resolve; }); + const entry = new Promise(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 | 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(resolve => { release = resolve; }); + const entry = new Promise(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 | 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); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index a8fe2e8dbc5..936218bbd3a 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -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((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 { diff --git a/server/src/index.ts b/server/src/index.ts index b059c4aad85..591ff902e1c 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -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(); diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index b2f4fe2bf0d..bcced794117 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -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)); diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts index ef5b7a21ceb..04c428be98e 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -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, 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) { 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, + canRun: () => boolean, canRecover: () => Promise) { + 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 | 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 }; } diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index b8cc5197d5a..50c293cd0e8 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -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)); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 432fc20006b..e023e149690 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + directoryCleanup: "directory-cleanup", instructionCommit: "instruction-commit", scheduledRetryCancellation: "scheduled-retry-cancellation", executionFinalization: "execution-finalization", diff --git a/server/src/services/directory-cleanup-notifications.ts b/server/src/services/directory-cleanup-notifications.ts new file mode 100644 index 00000000000..108df1cb715 --- /dev/null +++ b/server/src/services/directory-cleanup-notifications.ts @@ -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) : "*"); +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e11584ebdd7..91a832ce103 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -4084,6 +4084,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerInstructionDirectoryCleanupWork: ( + coordinator: Parameters[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[0], canRun: () => boolean, ) => instructionCopies.registerCapturedCommitWork(coordinator, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index ba1a51ce58a..3ab69cbff7e 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -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"); }); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 6a8894e4f45..bb6fd525c81 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -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