From c8b7afee06c183ab922d097de1183d61c0bc4387 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 06:37:48 -0500 Subject: [PATCH] fix(server): schedule native environment lease settlement (#17801) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit https://entire.com/gh/paperclipai/-/trails/24/changes/paperclip/24 ## Thinking Path > - Paperclip manages AI agents and the resources used by their work. > - Native run completion has several independent settlement duties. > - The recurring run reaper repeated environment-release discovery for completed native history. > - Existing lease rows already record releases that still need work. > - This pull request wakes the existing release owners from committed changes and schedules only unfinished release debt. > - Completed history becomes quiet while stop evidence, workspace preservation, and retention policy keep their existing owners. ## Linked Issues or Issue Description Refs: #17793, #17717. **What problem does this solve?** Periodic native finalization replay calls environment release even when no lease needs work. This adds recurring database traffic to an idle instance. **What is the proposed solution?** Register one scoped worker on the existing delivery coordinator. Discover work from native finalizations and existing environment leases. Wake it from native, lease, and settings transactions. Keep the release owner's saved pending-cleanup deadlines. Keep retries only while a concrete lease still needs settlement. **What alternatives have you considered?** A slower unconditional scan still queries idle history. A new job table duplicates existing lease truth. Replacing the compound callback would also change runtime-service and agent-status duties; those remain separate slices. This is slice 58. Parent #17793 has merged. This publication targets master and preserves that landed prerequisite. Fresh current-head checks and review remain required. ## What Changed - Keep ordinary company-only finalization hints out of environment settlement. Prove zero candidate SQL after quiet and preserve real lease delivery. - Reuse the existing durable destroy claim for run-owned cleanup, excluding concurrent attempts and fencing late receipts while preserving cleanup recovery. - Scope lease cleanup notifications to their company/run so settlement does not rediscover the entire backlog after each release. - Read only `nativeWorkspaceSync` and `workspaceExportRetry` JSON fields when discovering environment cleanup. Preserve saved run data and existing release decisions. - Reuse the environment runtime's release predicate for discovery. Add no table or independent timer. - Wake from existing finalization, projection, lifecycle, recovery, lease, and settings transaction boundaries. Simple lease acquisition now uses a transaction too. - Preserve recorded retention policy, pending-cleanup deadlines, original lease/provider ownership, and current release guards. - Re-read durable lease state after a release. A logged or swallowed per-lease failure does not discharge the obligation. - Delegate only periodic environment release. Preserve default live/manual behavior, runtime-service release, and agent-status callbacks. - Add an owner/producer/lifecycle audit and real database tests. ## Verification Accepted production source: `9596fa022b57feea026c4728c80bdc36037b804f`. Publication head: `fb252c8b0e52ce355e7d98db2e392e4831f99c61`. The actual parent squash is adopted by a normal merge with the complete accepted child tree unchanged. Only audit text follows. - **4,172 distinct cases in 64 complete server files pass**, plus **20 selected chat cases** (4,192 selected total). The main batch reports 3,440 passes and 49 skips. Both complete Stop/read files then pass 51 + 6 cases on separate fresh databases through their supported override. Their eight pure cases are counted once; skipped cases are never counted as passes. Separate process recovery passes 410; review response passes 273. The other 1,794 chat cases were not run locally. - **Quiet producer proof:** the controlled unscoped baseline fails with three candidate SELECTs after three routine company-only hints. The scoped case performs zero, releases a real eligible committed-run lease, and returns to quiet. The same test hash spans red and green. - **Static gates pass:** database build; server typecheck/build; module boundaries; migration order; current Copilot profile identity. Packages equal the accepted parent byte for byte. - **Fresh fixtures:** 39 prepared PostgreSQL 17 databases plus fresh helper-owned databases. The schema has all 412 journal entries through index/tag 0413_tired_kronos. All 65 original test hashes restore exactly; the indirect Stop fixture is also hashed. No assertion or timeout changes. - **Preserved records:** original 49 skips and their completed coverage remain documented. Parent handoff timeout and prior chat-idle failure records remain in their owning audits. This does not claim full repository, full chat, live provider, or deployed sleep acceptance. - [Current integration and native lease audit](https://github.com/paperclipai/paperclip/blob/fb252c8b0e52ce355e7d98db2e392e4831f99c61/doc/plans/2026-10-26-native-environment-settlement-sleep.md#current-master-integration-and-native-lease-quiet-proof-2026-10-29). Fresh CI, security, and exact-head Greptile 5/5 must pass before merge.
Earlier revision verification and failure history Receiver failure fairness: a controlled 1,100 ms batch reproduced a 30-second delay for healthy due receivers. The worker now supplies its existing per-item deadline through `nextRunAtOnError`. All 20 receiver lifecycle tests pass on the owner and combined tip `31c12f1d68ff4a51e06fc2be2abf2c2aee9f9d3a`; server production typechecking passes on the owner. The failed receiver retains its own backoff, and settled receivers return to zero recurring SQL. The five-second assertion budget is unchanged. See [the owning audit](https://github.com/paperclipai/paperclip/blob/6f3a0b8de149fb5620f15909473eb7c4455b9824/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md#healthy-receiver-progress-after-a-slow-failed-batch). Current-head CI and review remain required. Approved delegated Test-tab calls now evaluate caller and target policies in the existing claim transaction. Real HTTP regressions first showed a missing caller charge and shared-limit self-rejection. All 78 connection-management cases and 92 gateway cases now pass on the owning branch, along with server production typechecking. All 78 connection-management cases also pass at combined tip `d25f830db89c1623bff3547bfc68015c1c66e0cf`. The three new cases cover shared charges, separate charges, rollback on denial, duplicate approval, and rejection of a later authorized call. Intermediate counter/identity fixture failures and a reused-database Slack collision are recorded in [the owning audit](https://github.com/paperclipai/paperclip/blob/ae9bff1c06e34702f64a399e58b45a657c133bd1/doc/idle-approved-tool-recovery.md#delegated-test-tab-admission-after-policy-changes). The earlier combined checks below remain evidence for their stated revisions. Current-head CI and review remain required. Inherited voice-cleanup verification: the controlled slow-provider regression and all 13 lifecycle tests passed on its owner and combined tip `ab22d8ed0c09028315607414b83b375e5cc0a60a`. That repair changed only the fixture clock advancement and its audit note. Current head: `6f3a0b8de149fb5620f15909473eb7c4455b9824`. This remains a single-duty PR stacked on #17793. - Integrated pinned master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5` with normal merges. Every branch passed generated-schema checks and its target-branch migration-order gate. Existing SQL is unchanged; the 25 stack migrations follow master’s new migration. - Duty-specific producer coverage, lifecycle tests, and earlier failure records: [2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/7e619453567e078a7d745f28bfe8e5ede571f282/doc/plans/2026-10-24-recurring-server-sleep-stack.md), [2026-10-26-native-environment-settlement-sleep.md](https://github.com/paperclipai/paperclip/blob/7e619453567e078a7d745f28bfe8e5ede571f282/doc/plans/2026-10-26-native-environment-settlement-sleep.md). - Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355 startup/cancellation/recovery/dispatch/preservation tests**, **75 connection-management tests**, and production server typechecking pass. These combined checks do not mean every intermediate PR independently ran the full repository suite. - Real master-to-stack database upgrade passes: **379 unchanged prior migration receipts plus 25 new migrations**, with the gateway responsibility fixture preserved. - Owning-branch checks: 84 runner installation/settlement cases plus runner typechecking and generated TypeScript/Rust profile agreement; 253 GitHub review workflow cases. Stop-attribution and held-worker readiness regressions failed before their fixes. - Failure accounting: the initial combined run passed 349 cases and failed six idle assertions in a reused preservation database containing old cancelled runs. Repeating the combined command with a fresh preservation database passed. Earlier invalid project selection, merge import/type errors, and the interrupted full local suite are not counted as passing checks. - **Latest-head CI and review remain required.** Prior CI findings include GitHub idle observation, route teardown, attention cleanup, touch-picker E2E, workspace health, and voice-cleanup timing; local passes do not erase those failures. No full local repository-suite, deployed sleep, fleet cost, or live Copilot qualification claim is made.
## Risks - Existing lease owners retain destruction claims, provider identity, stop evidence, retention policy, and pending-cleanup deadlines. Incorrect changes here can release a live resource or lose a cleanup obligation. - Logged or swallowed release failures remain concrete debt. Real retries must retain their saved floor across unrelated hints. - In-process signals do not cover independent SQL or other replicas without ingress. - This delegates only periodic environment settlement. Runtime-service and agent-status duties remain separate. Existing Cloud blockers remain; no deployed sleep, database suspension, external alarm coverage, or cost savings is claimed. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, code execution, real PostgreSQL tests, and GitHub tools. The exact runtime model identifier and context-window size are not exposed in this session. ## 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 URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: Codex --- ...2026-10-24-recurring-server-sleep-stack.md | 125 +++++ ...-26-native-environment-settlement-sleep.md | 303 ++++++++++++ .../src/__tests__/environment-runtime.test.ts | 53 +- ...native-environment-settlement-work.test.ts | 451 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 11 +- server/src/index.ts | 8 +- .../services/delivery-work-notifications.ts | 1 + .../environment-cleanup-notifications.ts | 8 +- .../environment-lease-release-state.ts | 31 ++ .../services/environment-run-orchestrator.ts | 10 +- server/src/services/environment-runtime.ts | 55 ++- .../src/services/environment-task-runtime.ts | 4 +- server/src/services/environments.ts | 40 +- .../execution-control-reconciliation.ts | 2 + server/src/services/heartbeat.ts | 9 + server/src/services/heartbeat/recovery.ts | 35 +- server/src/services/heartbeat/run-control.ts | 4 +- .../src/services/heartbeat/run-lifecycle.ts | 2 + server/src/services/instance-settings.ts | 2 + ...ve-environment-settlement-notifications.ts | 7 + .../native-finalization-notifications.ts | 2 + .../native-environment-settlement-work.ts | 114 +++++ .../native-runtime/native-restart-recovery.ts | 2 + .../native-runtime/native-run-finalizer.ts | 2 + .../native-runtime/native-run-projection.ts | 2 + .../native-runtime/native-session-executor.ts | 4 +- .../native-workspace-export-resume.ts | 4 +- .../native-workspace-export-retry.ts | 4 +- .../native-runtime/native-workspace-sync.ts | 4 +- server/src/services/recovery/service.ts | 2 + .../src/services/sandbox-stop-and-retain.ts | 4 +- server/src/services/workspace-recovery.ts | 2 +- 32 files changed, 1232 insertions(+), 75 deletions(-) create mode 100644 doc/plans/2026-10-26-native-environment-settlement-sleep.md create mode 100644 server/src/__tests__/native-environment-settlement-work.test.ts create mode 100644 server/src/services/environment-lease-release-state.ts create mode 100644 server/src/services/native-environment-settlement-notifications.ts create mode 100644 server/src/services/native-runtime/native-environment-settlement-work.ts 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 54cbb1e16ec..ec628794216 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2,6 +2,10 @@ Baseline: `785d54636e1c8513ee5503d602daf6a5ee1970d6` (2026-10-24). +Current integration baseline: master `507dc70be26c191e372f09e58dc0965e73bf8f57` +(October 26). Earlier gate results below are historical; current acceptance is +pending. See the final integration/review-repair section for the current state. + The requested outcome is a reviewed PR stack for every remaining recurring server work item below. A PR owns one reason for background activity. Split compound items into separate PRs as their producer and lifecycle audits identify distinct @@ -3330,3 +3334,124 @@ Whole-stack master integration, older failed gates and suppression audits, and all remaining inventory reasons remain active. Nothing was merged. Existing active-transport replay polling remains a separate reason to inventory and migrate, rather than a false claim of whole-instance quiet. + +## Slice 57 acceptance and slice 58 environment settlement + +Slice 57 [#17793](https://github.com/paperclipai/paperclip/pull/17793) is published +at `be8f3284c2c8a05903f2eca000ad86d046780fec`. Latest-head checks show 53 success +and one neutral result, and the PR is mergeable. Greptile is 5/5 with no unresolved +threads. One failed-job rerun recovered initial runner shutdowns and missing or +truncated logs. The latter individual causes remain unknown. No source change +was needed for that rerun. Nothing was merged. + +Slice 58, `codex/sleep-58-native-environment-settlement`, moves the environment +release duty out of periodic native finalization replay. Existing lease records +and the existing release owner's predicate supply debt. Existing native and lease +transactions wake the shared coordinator. Saved cleanup deadlines, workspace +preservation, provider stop evidence, and warm-retention policy keep their owners. +See the [owner and lifecycle audit](2026-10-26-native-environment-settlement-sleep.md). +The initial combined run passed 127 affected tests. Final workspace checks and +remote acceptance remain required before this slice is accepted. + +The compound callback's runtime-service release and agent-status projection are +still outstanding. Workspace descriptor cleanup remains separate. All earlier +inventory reasons, master propagation, older failed gates, and suppression debts +remain active; this section does not narrow the original goal. + +## October 26 master integration and review repairs + +All 60 published branches (58 numbered duties and two auxiliary slices) now +contain master `507dc70be26c191e372f09e58dc0965e73bf8f57` through normal parent +merges. Earlier acceptance entries in this document describe their recorded +historical heads; they do not certify these new integrated heads. The current +stack still requires fresh CI, fresh review, and resolution of outstanding +findings. Nothing has been merged into master. + +Three review/compatibility repairs were made at their owning stack boundaries: + +- Slice 01 / [#17443](https://github.com/paperclipai/paperclip/pull/17443): + a successfully read future custom-image expiry now runs at its saved deadline. + A five-second deadline and an earlier committed replacement previously waited + for the 30-second retry floor. Failure and unfinished-overdue work retain + backoff. Slice 02 carries the same correction into the shared deadline helper. + The new unit and real-producer regressions failed first; root verification + passed 47 custom-image cases and 39 startup cases. Slice 02 passed 122 + custom-image, secret-proposal and startup cases. +- Slice 17 / [#17518](https://github.com/paperclipai/paperclip/pull/17518): + plugin inventory notification batching is now an explicit, optional shared + deadline contract. Only the plugin scheduler opts into its existing bounded + batching latency; consumers without the option reconcile notifications + immediately. Known saved deadlines preempt batching, including a ten-second + job deadline. This preserves bounded full-inventory reads for bursts of + job-status writes without making every successful expiry pass a cooldown. + Newly changed plugin configuration retains bounded batching latency. Idle + queues still disarm. The focused owner run passed 76 cases. +- Slice 51 / [#17717](https://github.com/paperclipai/paperclip/pull/17717): + HTTP readiness and later worker registration no longer wait for retained + physical-session cleanup. The existing scheduler-work and delivery owners + retain the original cleanup promise through idle drain and shutdown. A held + cleanup regression failed before the fix; 73 startup cases and 28 real-DB + retained-cleanup cases passed after it. + +Combined source revision `6a5ff3683614d8f954986157f158d74b1f653d22` passed +workspace typechecking, the full build, and 246 focused cases across seven +suites. The first combined run exposed the plugin batching dependency: 244 +passed and one failed. The explicit batching repair fixed that failure. This +ledger-only follow-up does not alter the tested code. Earlier integrated-tip +validation also covered 470 cases; that remains evidence for its recorded +revision, not a substitute for fresh checks. Full local `test:run` was not +repeated because of the previously recorded embedded PostgreSQL capacity +limitation. Full remote CI remains required. + +Known acceptance debt still includes approved-tool rate-limit admission and +permanent recovery classification, native recovery-notice lock ordering, +workspace-export scheduling and plugin completion-persistence CI failures, +cancellation lock failures, several retention/backlog/producer review findings, +and the native-worker scheduling-suppression audit. Passing focused tests does +not close those investigations. The remaining original inventory also remains +in scope; the next native duty audit is agent-status projection. An empty local +runtime-service map is already SQL/network quiet and does not by itself justify +another queue. No deployed host suspension, external alarm coverage, or fleet +savings are claimed by these repairs. + +## October 26 approved-review and interrupted-outcome repairs + +Slice 05 now retains the originally admitted rate slot through human review, +reconciles durable request/invocation history before dispatch, and settles +permanently invalid records without recurring recovery. Formal approvals still +awaiting human input remain quiet until a committed decision/cancellation; startup +rediscovers them. Configuration/dependency failures retain retry debt, and saved +or unknown provider outcomes are never replayed. The originating branch passed +208 gateway/policy/approval/signature cases and server typechecking. + +Propagation preserved each existing owner boundary. Slice 06 retains its shared +Test-tab fixture with optional rate admission. Slice 14 folds formal-review wakes +into the existing approval write transaction; slice 16 carries that signal through +the company-scoped approval notification owner; slice 49 also retains native task +scopes. Eight affected suites passed 506 cases at source +`d4ff4b9c04cbaa3739d1b62f7c29a3bff670c901`, including retention, seeding, and native +review consumers. This is evidence for those integration changes, not separate +certification of each intermediate branch. + +Slice 06 additionally preserves saved `cancelled` invocation status, completion +time, and error while repairing a stranded executing request. Its new regression +reproduced the unwanted overwrite; all 14 interrupted-review worker cases passed +at that branch. Final combined source `b91de5102faf548217ea489667fe51de98c9071a` +passes all 129 gateway tests, workspace typechecking, and the full build. This +ledger-only follow-up leaves the tested source unchanged. Full local test:run was +not repeated because of the previously documented embedded PostgreSQL capacity +limitation. Fresh remote CI is still required. + +No stack PR was merged. Earlier CI debts, other review findings (including native +notice lock order), the eight native facade suppression audits, the next native +agent-status projection duty, and the rest of the original inventory remain in +scope. These repairs do not establish deployed instance/database suspension. + +Slice 58 readiness follow-up: fresh review found its first physical provider +release could block HTTP readiness, just as retained-session cleanup had done. +The startup/shutdown regression reproduced that blockage. Registration now tracks +the worker's ready promise through the existing heartbeat work owner without +awaiting provider cleanup inline. All 84 startup and 20 real-database environment +settlement cases pass, plus workspace typechecking and the full build. This +changes only the last PR; earlier stack qualification and unresolved scope remain +as recorded above. Fresh CI is required and nothing is merged. diff --git a/doc/plans/2026-10-26-native-environment-settlement-sleep.md b/doc/plans/2026-10-26-native-environment-settlement-sleep.md new file mode 100644 index 00000000000..f6105655f57 --- /dev/null +++ b/doc/plans/2026-10-26-native-environment-settlement-sleep.md @@ -0,0 +1,303 @@ +# Native environment settlement without idle polling + +## Scope and owner + +Slice 58 of the recurring-server sleep stack. Parent: slice 57, PR #17793, +`be8f3284c2c8a05903f2eca000ad86d046780fec`. This slice moves only the environment +lease portion of native finalization recovery out of the recurring run reaper. +Runtime-service release, agent status projection, workspace descriptor cleanup, +active-transport replay, and the remaining reaper duties stay separate. + +The prior `settleRecoveredNativeWorkspace` callback did three jobs whenever the +native finalizer replayed committed results. Even historical runs with no leases +caused environment discovery and stop-receipt reads. The callback still serves +live/manual recovery. Periodic replay now delegates its environment portion to +one shared-coordinator worker. The other two callbacks retain their behavior. + +`run-control.releaseEnvironmentLeasesForRun` owns terminal resource policy and +warm-session stop checks. `environment-runtime.releaseRunLeases` owns exact lease +and provider operations. The worker invokes those owners; it does not interpret +provider receipts, select another provider, infer a stopped process from a terminal +status, or implement teardown. The release predicate is extracted without changing +its existing states. No schema, migration, independent timer, or job table is added. + +## Existing truth and deadlines + +A candidate has all of the following: + +- A native run bound to the same company and issue as its finalization. +- An accepted result bound to that exact run, company, issue, and finalization. +- A terminal run status. Running/unknown execution ownership is excluded. +- Either a committed finalization, or an exhausted workspace-finalization failure + identified by its existing `workspaceFinalizeAttempt` receipt. +- At least one lease still accepted by the release owner: active; a stop/export + handoff in `pending_cleanup`; or failed retained cleanup when the recorded + disposition calls for destruction. + +Released/expired/failed lease history and retained success do not keep this +worker scheduled. An unrelated pending-cleanup row belongs to the existing +cleanup worker. A retained failed lease does not earn destruction when the native +run's recorded disposition calls for retention. Disposition still comes from the +saved native workspace reference and export-retry evidence. Failed-turn warm +retention is normalized by the existing run-control owner. + +Stop/export handoffs keep the existing saved cleanup claim/retry deadline and +five-minute spacing. An attempted release that leaves real debt keeps a +30-second minimum retry. Exceptions use the shared coordinator's bounded backoff. +A release call returning successfully does not prove success: the existing API +logs per-lease failures. The worker re-reads lease truth before disarming. + +Discovery is paginated by run ID. Scoped hints are coalesced and fair dispatch +uses the existing scoped deadline worker. Current run status is checked again +before an automatic release; the original release owner retains its own guards. + +## Producer and transaction audit + +| Producer | Existing boundary / new hint | +| --- | --- | +| Accepted result, committed finalization, workspace failure | Existing `notifyNativeFinalizationWork` fanout | +| Run projection and metadata repair | Projection's owning transaction before its write | +| Terminal run lifecycle | Existing lifecycle transaction before status changes | +| Finalizer's direct projection path | Existing finalizer transaction | +| Restart/control recovery terminal writers | Existing terminal-write transactions | +| Lease acquisition and same-run reacquisition | Acquisition transaction; the simple insert now has the same transaction boundary | +| Lease release, cleanup handoff, metadata update, retry receipt | Existing environment-cleanup notification boundary | +| Worktree execution setting | Existing settings transaction, global reactivation hint | +| Startup/restart | Reconstruct from existing records without requiring a remembered hint | + +Hints serialize only the company/run/issue identifiers. They are awaited before +writes, and the shared database signal protocol owns outer transactions, +savepoints, rollback, and uncertain commit outcomes. No per-call recovery counter +is threaded through routes. No private Muse contract is used for environment work. +The supported writer model remains the server's instrumented writers. Independent +SQL writers or unrelated replicas do not acquire a new ingress guarantee here. + +Warm-session or workspace guards can defer an attempt. While the exact lease +still needs release, its scoped retry remains. A completed lease, changed owner, +or a no-longer-terminal run removes the entry on reconciliation. Concrete retry +work may still prevent host sleep; that is not idle-history polling. + +## Lifecycle and boundaries + +Startup registration uses the heartbeat facade's restore/worktree suppression +and task-drain checks. The app also gates warm standby, shutdown, and scheduler +enablement. Shared coordinator admission preserves dirty scopes across drain, +coalesces writes during execution, and joins worker shutdown. Settings changes +reactivate work that was suppressed rather than leaving a passive scan running. + +The environment-cleanup fanout is a shared change because lease writes can create +or discharge native settlement debt. With no eligible native run, it produces +one empty reconciliation and no recurring work. The generic cleanup worker, +provider claim fences, workspace preservation, retained-session cleanup, and +remote stop acknowledgment retain their owners. + +## Verification record + +- 17 new real-PostgreSQL lifecycle cases pass through the public heartbeat facade, + real native result/finalizer writers, and the existing environment runtime. + Controlled local-driver operations stand in for provider effects. +- Cases cover empty/retained history, first acquisition after finalization, + finalizer-produced release, restart, outer commit/rollback, swallowed release + failures, writes during an active release, drain/resume with zero SQL, + running/unverified owners, exact company scope, pending cleanup deadlines, + worktree/restore suppression, settings reactivation, keep-running retention, + exhausted workspace failure, and partial periodic delegation. +- The combined run passed 127 tests: 17 new cases, 28 existing native notice and + finalizer integration cases, and 82 startup cases. +- 85 existing run-control, heartbeat recovery, and environment-cleanup tests also + pass. The affected total is 212. Final workspace checks are recorded in the PR. +- The first test setup failed because migrations already create the unique local + environment. The fixture now uses that existing environment. The subsequent + lifecycle runs passed; no production change was made to bypass the constraint. + +The dedicated test databases use the existing local PostgreSQL server. No further +OS semaphore cleanup was performed. The full local suite still requires its +previously recorded environment capacity constraints to be resolved; CI supplies +full-suite acceptance. A green targeted run does not close the whole stack. + +## Remaining acceptance and limitations + +Remote checks, Greptile review, full-stack master propagation, and older stack +review/gate debts remain separately tracked. The latest observed `origin/master` +was `8c702d59dbc45ea5d491379b11b4883c17caebab`; this slice stays on its stacked +parent. No merge authority is assumed. + +No deployed provider teardown, database suspension, external Cloud alarm coverage, +or fleet cost reduction is claimed. Future deadlines still need their existing +external-wake/blocker contract before a host can sleep. The next distinct duties +in this callback are runtime-service lease settlement and agent terminal-status +projection; descriptor filesystem cleanup is another separate owner. + +## Review repair: per-lease deadlines and notification-safe retries + +Greptile's first review found two valid gaps at `7126f8e394`: + +1. One active lease made the whole run immediately due, and the existing batch + release then also attempted a second lease with a future cleanup deadline. +2. A notification reload rebuilt active debt at deadline zero and erased the + worker's 30-second retry floor. Unrelated lease changes could cause one-second + retries of the same failed release. + +Both defects reproduced before the repair. The mixed-lease test observed an +unexpected stop attempt. The notification test observed four release attempts +where one was expected. + +The background release capability now passes `respectCleanupDeadlines` through +run-control and the orchestrator to the existing environment runtime. The runtime +filters each lease by the shared deadline expression and rechecks that expression +at a stop-only claim. A stop intent prepared by this very call may make its first +attempt immediately, fenced by its fresh attempt identity and zero saved retry. +Manual/live release retains its prior default behavior. + +The worker retains a per-run in-memory retry floor across scope reloads. The saved +cleanup deadline remains authoritative and can change; the map retains only the +last attempt's minimum spacing. Entries are removed once their scoped debt no +longer exists. Restart reconstructs durable lease truth as before. No new +persistence or framework-wide scheduler semantics were added for this repair. + +## Review repair: readiness during physical cleanup + +The environment-settlement registration awaited its initial worker pass. A saved +stop handoff from exhausted workspace finalization can be eligible here while +excluded from the earlier startup scan, so a slow provider release could block +HTTP readiness and registration of later workers. + +Startup now registers the worker and tracks its ready promise with the existing +heartbeat work owner instead of awaiting physical cleanup inline. The delivery +coordinator still owns the worker, and drain/shutdown still join the same work. +The held-cleanup readiness/shutdown test is shared with retained-session cleanup; +its environment-settlement case failed before the fix. All 84 startup cases and +20 real-database environment-settlement lifecycle cases pass, as do workspace +typechecking and the full build. Full local test:run remains unqualified for the +previously recorded embedded-PostgreSQL capacity reason; remote checks remain +required. No provider resource debt was dropped and no deployed-sleep claim is made. + +## Narrow cleanup candidate reads + +Review finding `PRRT_kwDORclPyM6wgKt1` correctly identified broad JSON reads in +the candidate selector. Startup, notification reload, and the checks before and +after settlement selected complete `runnerProfileJson` and `resultJson` values, +although disposition needs only `nativeWorkspaceSync` and `workspaceExportRetry`. +The selector now projects those two nested values in SQL, preserving nulls and +the existing disposition contract. The actual settlement callback already uses +`getRunWorkspaceSettlement`, whose SQL has the same narrow projection; it was +inspected and did not require another change. No provider policy, retry deadline, +lease predicate, notification, or stored run record changes. + +A real PostgreSQL regression seeds four finalized runs with large unrelated saved +inputs/output plus destroy, keep-running, stop-and-retain, and invalid workspace +references. Candidate discovery and the checks before/after a controlled release +returned 7,208,692 response-value bytes before the fix and failed the 16 KiB bound. +The same test now passes that bound, still selects only explicit destruction for +a failed retained lease, preserves the conservative invalid-reference fallback, +and verifies that the saved input/output remains intact. All 21 cases in the +complete environment-settlement lifecycle suite pass, including actual release, +keep-running retention, export retries, per-lease deadlines, restart, outer +transactions, drain, and shutdown. Logs: `/tmp/sleep-environment-projection-red.log` +and `/tmp/sleep-environment-projection-green.log` with matching status files. + +The capture measures returned values at the postgres-js boundary, before Drizzle +decoding. It does not measure wire bytes or PostgreSQL storage/CPU savings: JSON +extraction still runs in PostgreSQL. This is evidence of reduced response payload +and application decoding work, not deployed suspension or fleet-cost savings. +Workspace verification and fresh PR checks remain separate gates. + +## Scoped lease-write notifications (2026-10-27) + +The shared cleanup fanout originally sent a global native-settlement hint for +all lease updates. A controlled six-run backlog reproduced seven full inventory +discoveries: startup plus one after every successful release. Each discovery also +recomputed remaining run deadlines. This was unnecessary database activity during +real work, even though the worker eventually became quiet. + +The existing cleanup helpers now accept the native worker's existing company/run +scope. Lease writers pass identity from their own lease/run records. Release, +bulk run release, cleanup claims and receipts, export recovery, stopped-sandbox +handoffs, recovery cleanup, plugin task failure, and harness/workspace metadata +writes preserve the same transaction and intent-before-write contract. ID-only +metadata updates read and lock the two scope columns within their transaction; +other reads/returns that already fetch the lease include those scalar columns. +No caller receives a new lifecycle object or manual recovery wrapper. + +The generic environment-cleanup and historical-cleanup topics retain their prior +contracts. Startup and broad custom-image cleanup changes still allow global +native discovery because they have no single run scope. Lease writes no longer +cause cross-company global reloads. The worker, deadlines, cleanup authority, +provider operations, and supported process-local writer model are unchanged. + +Evidence: +- Baseline backlog regression failed with seven unscoped discoveries instead of + one. With scoped writes, the same real finalizer/lease/runtime path discovers + globally once, releases all six leases, and returns to zero SQL/network work. +- A real nested metadata write produces no discovery before outer commit, wakes + only its run after commit, and produces no discovery after outer rollback. +- Seven selected suites completed with 203 passing cases and 60 embedded-only + cases skipped. This includes native settlement, export retry, workspace + recovery, plugin task contracts, and workspace-sync unit coverage. +- Initial typechecking caught an attempt to access a run ID on the plugin cleanup + API's deliberately narrow lease input. The existing update now returns that + scalar field instead. Final server typechecking passes; the plugin task suite + then passes all 32 cases (four embedded driver cases remain skipped). + +These local checks do not establish full CI or deployed suspension. The pending +whole-stack checks, remaining review findings, and unmigrated recurring duties +remain open. + +## Current master integration and native lease quiet proof (2026-10-29) + +Accepted production source `9596fa022b57feea026c4728c80bdc36037b804f` contains reviewed parent `305014b030d6c5eb891c70bc57036f4192d4fdad`. +That parent merged as `829f9e7eafe57588d214697efca336b35aecf45d`. A normal merge adopts the actual squash +after proving complete parent-tree identity. The complete accepted child tree +stays unchanged. Only this audit follows that source in the publication. + +Composition preserves 1,218 parent-only paths, 14 owned-only paths, 11,516 +identical paths, and 17 shared automatic merge results. The only manual +resolution adds environment settlement inside the existing ordinary issue/run +guard in native finalization notifications. Packages equal the parent byte for +byte, including current Copilot profile 80 and its historical profile fixtures. +No package or live provider qualification is added by this slice. + +The new database regression exposes the quiet-path requirement. Moving only +the new environment notification outside the issue/run guard makes three +routine company-only finalization hints perform three environment candidate +SELECTs. That controlled baseline fails the quiet assertion. With the guard +retained, the same test passes with zero candidate SELECTs. It then commits real +native work, creates an eligible lease through the owning service, observes +physical release, and returns to quiet. The test hash is identical across the +controlled red/green pair, and the temporary production edit restores exactly. +No assertion or timeout changes. Real lease, run, and settings writers retain +their scoped or global notifications. + +Acceptance covers 4,172 distinct passing cases in 64 complete server test +files, plus 20 targeted chat cases (4,192 selected cases total). The main +62-file command reports 3,440 passes and 49 skips. All skips belong to the two +Stop/control-read files: their indirect shared fixture has its own supported +PAPERCLIP_RUN_CONTROL_TEST_DATABASE_URL override. The initial live log showed +43 Stop skips; the final report includes another six read cases. Both complete +files then pass on separate fresh databases: 51 control cases and six read +cases. Their eight already-passing pure cases are counted once. This closes +the coverage gap without counting skipped tests as passing. + +Separate process recovery passes 410, and review response passes 273. +The targeted chat selection passes 20 committed-response and GitHub idle cases; +1,794 other chat cases are outside the local selection. The complete +Photon file is part of the main batch. Coverage also includes lease release, +retention, environment runtime and drivers, workspace sync, native execution, +Stop, startup readiness, queue, goals, interactions, monitor and status workers. + +All 65 selected test files restore to their exact committed hashes after fixture +imports. The follow-up also hashes the indirect run-control fixture helper. +Fixtures use 39 prepared fresh PostgreSQL 17 databases, plus fresh per-call +helper databases. There are 412 migration journal entries through index/tag +0413_tired_kronos; the final tag number is not an entry count. Database build, +server production typecheck/build, module boundaries, migration order, and +Copilot identity pass. + +Failure and scope accounting remains explicit. The unscoped baseline fails as +expected. The original 49 skipped cases are recorded and completed as above. +Earlier parent handoff timeout and chat-idle failure records remain in their +own audits. Latest-head CI, security, and exact-head Greptile 5/5 remain separate +merge requirements. There is no full local repository or full chat-suite claim, +no live provider qualification, and no deployed sleep, database suspension, +external alarm coverage, or cost claim. The writer model remains supported +in-process producers. Existing Cloud sleep blockers remain. diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index fda31d55fb8..a4b82d7b113 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -2,6 +2,7 @@ import { JsonRpcCallError } from "@paperclipai/plugin-sdk"; import { collectRunFailureDiagnostics } from "../services/run-failure-diagnostics.js"; import { createSandboxOrphanCleanupSpool } from "../services/sandbox-orphan-cleanup-spool.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; +import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { createHash, randomUUID } from "node:crypto"; import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; import os from "node:os"; @@ -62,7 +63,10 @@ import { ROOT_CONTEXT, trace } from "@opentelemetry/api"; import { buildNativeHarnessBackupManifest } from "../services/native-runtime/native-session-executor.ts"; import { createNativeHarnessBackupStamp } from "../services/native-runtime/native-harness-backup-stamp.ts"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ENVIRONMENT_RUNTIME_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; const sshFixtureSupport = await getSshEnvLabSupport(); const managedCloudConfig = { v: 1, mode: "cloud", catalogVersion: "test", features: {}, plugins: { autoInstall: ["daytona"] } }; @@ -199,9 +203,12 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { const previousOrphanCleanupSpoolDir = process.env.SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR; beforeAll(async () => { - const started = await startEmbeddedPostgresTestDatabase("environment-runtime"); - stopDb = started.stop; - connectionString = started.connectionString; + if (externalTestDatabaseUrl) connectionString = externalTestDatabaseUrl; + else { + const started = await startEmbeddedPostgresTestDatabase("environment-runtime"); + stopDb = started.stop; + connectionString = started.connectionString; + } }); beforeEach(async () => { @@ -6246,12 +6253,12 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { ).resolves.toMatchObject({ status }); }); - it.each(["issue", "workspace", "environment"] as const)( + it.each(["issue", "workspace", "environment", "run"] as const)( "durably claims failed reusable cleanup before %s teardown and recovers a provider failure", async (scope) => { const seeded = await seedReusablePluginSandboxLease(); await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, seeded.runId)); - await environmentService(db).releaseLease(seeded.reusableLease.id, "failed"); + if (scope !== "run") await environmentService(db).releaseLease(seeded.reusableLease.id, "failed"); const issueId = randomUUID(); if (scope === "issue") { await db.insert(issues).values({ id: issueId, companyId: seeded.companyId, title: "Cleanup crash window", status: "done" }); @@ -6276,7 +6283,9 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { getWorker: () => ({ supportedMethods: ["environmentDestroyLease"] }), } as unknown as PluginWorkerManager }); const runtime = makeRuntime(); - const close = (service: ReturnType) => scope === "environment" + const close = (service: ReturnType) => scope === "run" + ? service.releaseRunLeases(seeded.runId, "released", undefined, "destroy") + : scope === "environment" ? service.destroyReusableSandboxLeasesForEnvironment({ environmentId: seeded.environment.id }) : service.destroyReusableSandboxLeases({ companyId: seeded.companyId, ...(scope === "issue" ? { issueId } : { executionWorkspaceId: seeded.executionWorkspaceId }) }); @@ -6312,6 +6321,36 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { }, ); + it.each([false, true])("preserves a newer destroy receipt when an old run cleanup returns (failure=%s)", async failure => { + const seeded = await seedReusablePluginSandboxLease(); + const service = environmentService(db); + let confirmed: Awaited>; + const call = vi.fn(async (_id: string, method: string) => { + if (method !== "environmentDestroyLease") throw new Error(`Unexpected method ${method}`); + + + const current = (await service.getLeaseById(seeded.reusableLease.id))!; + const attemptId = randomUUID(); + await db.update(environmentLeases).set({ status: "pending_cleanup", + metadata: { ...current.metadata, pendingCleanupAttemptId: attemptId }, + }).where(eq(environmentLeases.id, current.id)); + confirmed = await service.releaseLease(current.id, "expired", { expectedPendingCleanupAttemptId: attemptId, + cleanupStatus: "success", remoteExecutionTermination: remoteTerminationReceipt(current, + { providerLeaseId: current.providerLeaseId, state: "destroyed" }), + }); + if (failure) throw new Error("Late failure from the old provider request"); + return { providerLeaseId: current.providerLeaseId, state: "destroyed" }; + }); + const runtime = environmentRuntimeService(db, { pluginWorkerManager: { + isRunning: () => true, call, getWorker: () => ({ supportedMethods: ["environmentDestroyLease"] }), + } as unknown as PluginWorkerManager }); + await expect(runtime.releaseRunLeases(seeded.runId, "released", undefined, "destroy")).resolves.toEqual([]); + expect(call).toHaveBeenCalledOnce(); + expect(confirmed!).toMatchObject({ status: "expired", cleanupStatus: "success", + metadata: { remoteExecutionTermination: { state: "destroyed" } } }); + expect(await service.getLeaseById(seeded.reusableLease.id)).toEqual(confirmed!); + }); + it.each(["queued", "scheduled_retry", "running"] as const)( "fences scoped cleanup when its holding run becomes %s after selection", async (status) => { const seeded = await seedReusablePluginSandboxLease(); diff --git a/server/src/__tests__/native-environment-settlement-work.test.ts b/server/src/__tests__/native-environment-settlement-work.test.ts new file mode 100644 index 00000000000..16a6e147f69 --- /dev/null +++ b/server/src/__tests__/native-environment-settlement-work.test.ts @@ -0,0 +1,451 @@ +import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; +import { recordNativeWorkspaceFinalizationFailure } from "../services/native-runtime/native-workspace-finalizer.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js"; +import * as workspaceRuntime from "../services/workspace-runtime.js"; +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, companies, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { environmentRuntimeService, type EnvironmentDriver } from "../services/environment-runtime.js"; +import { environmentService } from "../services/environments.js"; +import { nativeEnvironmentSettlementWork } from "../services/native-runtime/native-environment-settlement-work.js"; +import { notifyNativeEnvironmentSettlement } from "../services/native-environment-settlement-notifications.js"; +import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; +import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; +import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; +import { workspaceOperationService } from "../services/workspace-operations.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +vi.mock("../services/agent-task-run-telemetry.js", () => ({ emitAgentTaskRun: vi.fn(async () => undefined) })); +vi.mock("../services/run-failure-report.js", async importOriginal => ({ + ...await importOriginal(), reportRunFailure: vi.fn(async () => undefined), +})); + +describe("native environment release scheduling", () => { + let temporary: Awaited> | undefined; + let db: Db, root: string; + let coordinator: ReturnType | undefined; + let worker: ReturnType["register"]> | undefined; + let release: ReturnType>; + let failing = false, admitted = true; + let stopOnly: ReturnType; + const companyId = randomUUID(), agentId = randomUUID(); + let environmentId: string; + const priorLogs = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + const errors: unknown[] = []; + beforeAll(async () => { + root = await mkdtemp(join(tmpdir(), "native-environment-")); + process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = root; + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_ENVIRONMENT_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-environment-"); + db = createDb(external ?? temporary!.connectionString); + await db.insert(companies).values({ id: companyId, name: "Environment settlement", issuePrefix: `NE${randomUUID().slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Release owner", adapterType: "paperclip_runner" }); + environmentId = (await db.select({ id: environments.id }).from(environments).where(eq(environments.driver, "local")))[0].id; + }, 30_000); + beforeEach(() => { + admitted = true; failing = false; errors.length = 0; + stopOnly = vi.fn(async () => ({ state: "stopped" })); + release = vi.fn(async input => { + if (failing) throw new Error("fixture release unavailable"); + return environmentService(db).releaseLease(input.lease.id, input.status ?? "released", { cleanupStatus: "success" }); + }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; worker = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); await rm(root, { recursive: true, force: true }); + if (priorLogs === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorLogs; + }); + async function start(runtimeEnv?: NodeJS.ProcessEnv) { + const runtime = environmentRuntimeService(db, { drivers: ["local", "sandbox"].map(driver => ({ driver, releaseRunLease: release, retryPendingSandboxTeardown: stopOnly }) as EnvironmentDriver) }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = heartbeatService(db, { environmentRuntime: runtime, runtimeEnv }).registerNativeEnvironmentSettlementWork(coordinator, () => admitted); + await worker.ready; + } + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + } + async function quiet() { + await flush(); + const previous = db.$client.options.debug; + const queries: string[] = [], calls = release.mock.calls.length; + db.$client.options.debug = (_id, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); await delay(10); expect(queries).toEqual([]); expect(release).toHaveBeenCalledTimes(calls); } + finally { db.$client.options.debug = previous; } + } + async function seed() { + const runId = randomUUID(), issueId = randomUUID(), contractId = randomUUID(), sourceInstanceId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Commit accepted work", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "result-worker-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "result-worker-v1", objective: "Finish", criteria: [{ id: "objective", requirement: "Complete the task" }] }, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, runtimeMode: "native", status: "running", + nativeIssueId: issueId, nativeSessionId: runId, runnerInstanceId: sourceInstanceId, + completionContractId: contractId, completionContractSha256: contractId, contextSnapshot: { issueId } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + return { companyId, issueId, runId, agentId, sessionId: runId, completionContractId: contractId, + completionContractSha256: contractId, sourceInstanceId, controlPlaneSourceInstanceId: "result-worker-test" }; + } + async function complete(binding: Awaited>, producer: "control_plane" | "coordinator", writer: Db = db, terminalStatus: "succeeded" | "failed" | "cancelled" = "succeeded") { + const port = producer === "control_plane" ? new PaperclipControlPlanePort(writer, binding) + : new NativeRunCoordinatorStore(writer, { ...binding, normalizedSessionId: binding.sessionId, + runnerSourceInstanceId: binding.sourceInstanceId, + completionContractRevision: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.contractRevision, + completionContractCriterionIds: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.criteria.map(item => item.criterionId) }); + await port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, runTerminalState: terminalStatus }, + callerResultId: `${binding.runId}:result` }); + } + async function barrier(binding: Awaited>) { + await workspaceOperationService(db).createRecorder({ companyId, issueId: binding.issueId, heartbeatRunId: binding.runId }) + .recordOperation({ phase: "workspace_finalize", cwd: root, run: async () => ({ status: "succeeded", exitCode: 0 }) }); + } + async function state(runId: string) { + return db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)).then(rows => rows[0]); + } + async function read(runId: string) { + return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0]); + } + async function accepted(status: "succeeded" | "failed" | "cancelled" = "succeeded") { + const binding = await seed(); await complete(binding, "control_plane", db, status); await barrier(binding); return binding; + } + async function committed() { + const binding = await accepted(); + const result = await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect(result.phase, JSON.stringify(await state(binding.runId))).toBe("committed"); + return binding; + } + async function lease(binding: Awaited>, writer: Db = db) { + return environmentService(writer).acquireLease({ companyId, environmentId, heartbeatRunId: binding.runId, + issueId: binding.issueId, leasePolicy: "ephemeral", provider: "local", providerLeaseId: randomUUID() }); + } + it("discovers and rechecks cleanup without loading unrelated saved run payloads", async () => { + const payload = "unrelated saved run data ".repeat(24_000); + const fixtures = []; + for (const disposition of ["destroy", "keep_running", "stop_and_retain", "invalid"] as const) { + const binding = await committed(); + const run = await read(binding.runId); + await db.update(heartbeatRuns).set({ + runnerProfileJson: { ...run.runnerProfileJson, savedInputs: payload, nativeWorkspaceSync: { + schema: disposition === "invalid" ? "unknown" : "paperclip.native-workspace-sync/v3", state: "finalized", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: "c".repeat(64), + workspaceId: randomUUID(), leaseId: randomUUID(), providerLeaseId: "saved", remoteCwd: "/workspace", + resourceDisposition: disposition === "invalid" ? "destroy" : disposition, + } }, + resultJson: { ...run.resultJson, savedOutput: payload, workspaceExportRetry: { pending: true } }, + }).where(eq(heartbeatRuns.id, binding.runId)); + const acquired = await lease(binding); + await environmentService(db).releaseLease(acquired.id, "retained", { cleanupStatus: "failed" }); + fixtures.push(binding); + } + const events: Record[] = []; + const now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { nativeRunFinalizations, heartbeatRuns, environmentLeases }, capture); + const settle = vi.fn(async () => {}); + const work = nativeEnvironmentSettlementWork(db, settle); + try { + const entries = await work.load({ companyId }); + + + expect(entries.map(entry => entry.id)).toEqual([fixtures[0].runId]); + expect(await work.check(entries[0])).toMatchObject({ id: fixtures[0].runId }); + expect(settle).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ runId: fixtures[0].runId, succeeded: true })); + capture.flush(); + const reads = events.filter(event => event.table === "native_run_finalizations" && event.operation === "select"); + expect(reads.length).toBeGreaterThan(0); + expect(reads.reduce((calls, event) => calls + Number(event.calls), 0)).toBeGreaterThanOrEqual(3); + expect(reads.reduce((bytes, event) => bytes + Number(event.responseValueBytes), 0)).toBeLessThan(16_384); + } finally { capture.flush(); } + const stored = await read(fixtures[0].runId); + expect(stored.runnerProfileJson?.savedInputs).toBe(payload); + expect(stored.resultJson?.savedOutput).toBe(payload); + }); + + it("ignores routine company-only finalization hints and still releases a real native lease", async () => { + await start(); await quiet(); + const previous = db.$client.options.debug, queries: string[] = []; + db.$client.options.debug = (_id, query) => queries.push(query); + try { + for (let attempt = 0; attempt < 3; attempt++) { + await db.transaction(tx => notifyNativeFinalizationWork(tx, { companyId })); + await flush(); + } + expect(queries.filter(query => query.startsWith("select") && query.includes('"environment_leases"'))).toEqual([]); + } finally { db.$client.options.debug = previous; } + const f = await committed(); await lease(f); await flush(); + expect(release).toHaveBeenCalledTimes(1); await quiet(); + }); + + it("stays quiet with no release debt and wakes on a lease committed after finalization", async () => { + const binding = await committed(); await start(); await quiet(); + const acquired = await lease(binding); await flush(); + expect(release).toHaveBeenCalledTimes(1); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("released"); + await quiet(); + }); + it("drains a backlog without globally rediscovering it after each lease release", async () => { + const bindings = []; + for (let index = 0; index < 6; index += 1) { + const binding = await committed(); await lease(binding); bindings.push(binding); + } + const previous = db.$client.options.debug; + const discovery: { query: string; parameters: readonly unknown[] }[] = []; + db.$client.options.debug = (_id, query, parameters) => { + if (query.includes('from "native_run_finalizations" inner join "heartbeat_runs"')) { + discovery.push({ query, parameters }); + } + }; + try { + await start(); await flush(); + expect(release).toHaveBeenCalledTimes(bindings.length); + + + expect(discovery.filter(read => !bindings.some(binding => read.parameters.includes(binding.runId)))) + .toHaveLength(1); + await quiet(); + } finally { db.$client.options.debug = previous; } + }); + + it("keeps a metadata wake scoped and defers it until the outer transaction commits", async () => { + const binding = await committed(), acquired = await lease(binding); + await environmentService(db).releaseLease(acquired.id, "pending_cleanup", { cleanupStatus: "failed" }); + const metadata = { nativeWorkspaceExportResume: {}, pendingCleanupRetryAfterMs: Date.now() + 600_000 }; + await environmentService(db).updateLeaseMetadata(acquired.id, metadata); + await start(); + const previous = db.$client.options.debug; + const reads: (readonly unknown[])[] = []; + db.$client.options.debug = (_id, query, parameters) => { + if (query.includes('from "native_run_finalizations" inner join "heartbeat_runs"')) reads.push(parameters); + }; + let finish!: () => void, entered!: () => void; + const held = new Promise(resolve => { finish = resolve; }); + const reached = new Promise(resolve => { entered = resolve; }); + let write: Promise | undefined; + try { + write = db.transaction(async tx => { + await environmentService(tx as unknown as Db).updateLeaseMetadata(acquired.id, metadata); + entered(); await held; + }); + await reached; await vi.advanceTimersByTimeAsync(2000); + expect(reads).toEqual([]); + finish(); await write; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(reads.length).toBeGreaterThan(0); }); + expect(reads.every(parameters => parameters.includes(binding.runId))).toBe(true); + expect(release).not.toHaveBeenCalled(); + reads.length = 0; + await expect(db.transaction(async tx => { + await environmentService(tx as unknown as Db).updateLeaseMetadata(acquired.id, metadata); + throw new Error("rolled back metadata"); + })).rejects.toThrow("rolled back metadata"); + await vi.advanceTimersByTimeAsync(2000); expect(reads).toEqual([]); + } finally { finish(); await write; db.$client.options.debug = previous; } + }); + + it("uses the real finalizer commit to release an existing active lease", async () => { + const binding = await accepted(); const acquired = await lease(binding); + await start(); await quiet(); expect(release).not.toHaveBeenCalled(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + await flush(); expect(release).toHaveBeenCalledTimes(1); + expect((await environmentService(db).getLeaseById(acquired.id))?.cleanupStatus).toBe("success"); await quiet(); + }); + it("reconstructs an unfinished release on restart and ignores released history", async () => { + const binding = await committed(), acquired = await lease(binding); + await start(); await flush(); expect(release).toHaveBeenCalledTimes(1); + await coordinator!.stop(); coordinator = undefined; + await start(); await quiet(); expect(release).toHaveBeenCalledTimes(1); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("released"); + }); + it("does not release a nested acquisition until the outer commit, or after rollback", async () => { + const binding = await committed(); await start(); await quiet(); + let finish!: () => void, ready!: () => void; + const entered = new Promise(resolve => { ready = resolve; }); + const hold = new Promise(resolve => { finish = resolve; }); + const writing = db.transaction(async tx => { const acquired = await lease(binding, tx as unknown as Db); ready(); await hold; return acquired; }); + await entered; await vi.advanceTimersByTimeAsync(2000); expect(release).not.toHaveBeenCalled(); + finish(); await writing; await flush(); expect(release).toHaveBeenCalledTimes(1); + await expect(db.transaction(async tx => { await lease(binding, tx as unknown as Db); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await quiet(); expect(release).toHaveBeenCalledTimes(1); + }); + it("retains a concrete failure deadline even though the release API logs and swallows errors", async () => { + const binding = await committed(), acquired = await lease(binding); failing = true; + await start(); expect(release).toHaveBeenCalledTimes(1); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(10_000); expect(release).toHaveBeenCalledTimes(1); + failing = false; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(release).toHaveBeenCalledTimes(2); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("released"); await quiet(); + }); + it("retains a first wake while drained without SQL, then releases on resume", async () => { + const binding = await committed(); await start(); await quiet(); admitted = false; + const acquired = await lease(binding); await vi.advanceTimersByTimeAsync(1000); + const reads: string[] = [], previous = db.$client.options.debug; + db.$client.options.debug = (_id, query) => { reads.push(query); }; + await vi.advanceTimersByTimeAsync(120_000); await delay(10); + db.$client.options.debug = previous; + expect(reads).toEqual([]); expect(release).not.toHaveBeenCalled(); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("released"); await quiet(); + }); + it("does not confuse a running or unverified native owner with settled work", async () => { + const binding = await committed(); const acquired = await lease(binding); + await db.update(heartbeatRuns).set({ status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" }).where(eq(heartbeatRuns.id, binding.runId)); + await start(); await quiet(); expect(release).not.toHaveBeenCalled(); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("active"); + }); + it("ignores retained success and ordinary pending cleanup owned by the cleanup worker", async () => { + const binding = await committed(); const retained = await lease(binding), cleanup = await lease(binding); + await environmentService(db).releaseLease(retained.id, "retained", { cleanupStatus: "success" }); + await environmentService(db).releaseLease(cleanup.id, "pending_cleanup", { cleanupStatus: "failed" }); + await start(); await quiet(); expect(release).not.toHaveBeenCalled(); + }); + it("honors the saved pending-cleanup deadline and the exact company scope", async () => { + const binding = await committed(), acquired = await lease(binding); + await environmentService(db).updateLeaseMetadata(acquired.id, { nativeWorkspaceExportResume: {}, pendingCleanupRetryAfterMs: Date.now() + 600_000 }); + await environmentService(db).releaseLease(acquired.id, "pending_cleanup", { cleanupStatus: "failed" }); + const work = nativeEnvironmentSettlementWork(db, async () => undefined); + expect(await work.load({ companyId: randomUUID(), runId: binding.runId })).toEqual([]); + const [pending] = await work.load({ companyId, runId: binding.runId }); + expect(pending.dueAt).toBeGreaterThanOrEqual(Date.now() + 599_000); + await start(); expect(release).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(120_000); expect(release).not.toHaveBeenCalled(); + await environmentService(db).releaseLease(acquired.id, "released", { cleanupStatus: "success" }); await quiet(); + }); + it("wakes from the real lifecycle writer when an execution hold is resolved", async () => { + const binding = await committed(); const acquired = await lease(binding); + await db.update(heartbeatRuns).set({ status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" }).where(eq(heartbeatRuns.id, binding.runId)); + await start(); await quiet(); + const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: vi.fn() } as unknown as HeartbeatLifecycleDependencies); + await lifecycle.setRunStatus(binding.runId, "succeeded", { nativePhase: "committed", errorCode: null }); + await flush(); expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("released"); await quiet(); + }); + it.each(["restore", "worktree"])("preserves %s suppression and reactivates through settings", async mode => { + const binding = await committed(), acquired = await lease(binding); + const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" } + : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "environment-suppression-test" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + if (mode === "worktree") await settings.updateExperimental({ enableWorktreeRunExecution: false }); + await start(runtimeEnv); await quiet(); expect(release).not.toHaveBeenCalled(); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("active"); + if (mode === "worktree") { + try { + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await flush(); expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("released"); await quiet(); + } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + } + }); + it("preserves keep-running retention instead of turning a completed native run into destruction", async () => { + const binding = await committed(); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v3", state: "finalized", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), + finalHostSha256: "c".repeat(64), workspaceId: randomUUID(), leaseId: randomUUID(), providerLeaseId: "kept", remoteCwd: "/workspace", + resourceDisposition: "keep_running", + } } }).where(eq(heartbeatRuns.id, binding.runId)); + const acquired = await environmentService(db).acquireLease({ companyId, environmentId, heartbeatRunId: binding.runId, + issueId: binding.issueId, leasePolicy: "reuse_by_environment", provider: "local", providerLeaseId: "kept" }); + await start(); await flush(); expect(release).not.toHaveBeenCalled(); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("retained"); await quiet(); + }); + it.each([true, false])("delegates only periodic environment settlement, keeping manual behavior and other callbacks: %s", async delegated => { + const binding = await committed(), acquired = await lease(binding); + const runtimeRelease = vi.spyOn(workspaceRuntime, "releaseRuntimeServicesForRun"); + const runtime = environmentRuntimeService(db, { drivers: ["local", "sandbox"].map(driver => ({ driver, releaseRunLease: release, retryPendingSandboxTeardown: stopOnly }) as EnvironmentDriver) }); + await heartbeatService(db, { environmentRuntime: runtime }).reapOrphanedRuns({ staleThresholdMs: 300_000, + skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, + skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, + skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, + skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, + skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, + skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, + skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: delegated, + }); + expect(release.mock.calls.filter(([input]) => input.lease.heartbeatRunId === binding.runId)).toHaveLength(delegated ? 0 : 1); + expect(runtimeRelease).toHaveBeenCalledWith(binding.runId); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe(delegated ? "active" : "released"); + }); + + it("wakes from a durable exhausted workspace-export failure and uses failed-run retention", async () => { + const binding = await seed(); await complete(binding, "control_plane"); const acquired = await lease(binding); + await start(); await quiet(); + const failure = await recordNativeWorkspaceFinalizationFailure(db, binding.runId, { stderrExcerpt: "workspace_sync_out_unrecoverable" }); + expect(failure.phase).toBe("terminal_failure"); + await flush(); expect(release).toHaveBeenCalledTimes(1); + expect(release).toHaveBeenCalledWith(expect.objectContaining({ resourceDisposition: "stop_and_retain", status: "failed" })); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("failed"); await quiet(); + }); + it("retains a lease committed during a release and does not overlap physical attempts", async () => { + const binding = await committed(); await lease(binding); + let finish!: () => void, started!: () => void; + const began = new Promise(resolve => { started = resolve; }); + const hold = new Promise(resolve => { finish = resolve; }); + release.mockImplementationOnce(async input => { + started(); await hold; + return environmentService(db).releaseLease(input.lease.id, "released", { cleanupStatus: "success" }); + }); + const starting = start(); await began; + const later = await lease(binding); await vi.advanceTimersByTimeAsync(2000); + expect(release).toHaveBeenCalledTimes(1); finish(); await starting; + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(release).toHaveBeenCalledTimes(2); + expect((await environmentService(db).getLeaseById(later.id))?.status).toBe("released"); await quiet(); + }); + + it("does not let an active lease pull a second lease's cleanup deadline forward", async () => { + const binding = await committed(); const active = await lease(binding), deferred = await lease(binding); + await environmentService(db).updateLeaseMetadata(deferred.id, { + nativeWorkspaceExportResume: {}, pendingCleanupRetryAfterMs: Date.now() + 600_000, + }); + await environmentService(db).releaseLease(deferred.id, "pending_cleanup", { cleanupStatus: "failed" }); + await start(); + expect((await environmentService(db).getLeaseById(active.id))?.status).toBe("released"); + expect(stopOnly).not.toHaveBeenCalled(); + expect((await environmentService(db).getLeaseById(deferred.id))?.metadata?.pendingCleanupInFlight).not.toBe(true); + await environmentService(db).releaseLease(deferred.id, "released", { cleanupStatus: "success" }); await quiet(); + }); + it("preserves a failed release's retry floor across unrelated lease notifications", async () => { + const binding = await committed(); await lease(binding); failing = true; + await start(); expect(release).toHaveBeenCalledTimes(1); + const other = await seed(), otherLease = await lease(other); + for (let i = 0; i < 3; i++) { + await environmentService(db).updateLeaseMetadata(otherLease.id, { change: i }); + await vi.advanceTimersByTimeAsync(2000); await delay(10); + } + expect(release).toHaveBeenCalledTimes(1); + failing = false; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(release).toHaveBeenCalledTimes(2); await quiet(); + }); + + it("permits the first attempt for a freshly prepared stop without bypassing an old retry", async () => { + const binding = await committed(), sandboxId = randomUUID(); + await db.insert(environments).values({ id: sandboxId, name: `Fresh stop ${sandboxId}`, driver: "sandbox" }); + const acquired = await environmentService(db).acquireLease({ companyId, environmentId: sandboxId, + heartbeatRunId: binding.runId, issueId: binding.issueId, leasePolicy: "ephemeral", provider: "e2b", providerLeaseId: "fresh-stop" }); + await start(); expect(stopOnly).toHaveBeenCalledTimes(1); + + + await environmentService(db).releaseLease(acquired.id, "released", { cleanupStatus: "success" }); await quiet(); + }); + +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 65e59ea8884..4cc764a5e9a 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -123,6 +123,7 @@ const { registerExternalConversationSettlementWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeRecoveryNoticeWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeRunProjectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerNativeEnvironmentSettlementWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeStatusReassessmentWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeReviewResponseWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeResponsePresentationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), @@ -619,7 +620,9 @@ describe("startServer feedback export wiring", () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); expect(heartbeatServiceMock.registerNativeRunProjectionWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(heartbeatServiceMock.registerNativeEnvironmentSettlementWork).toHaveBeenCalledTimes(enabled ? 1 : 0); if (enabled) expect(heartbeatServiceMock.registerNativeRunProjectionWork.mock.calls[0][1]()).toBe(true); + if (enabled) expect(heartbeatServiceMock.registerNativeEnvironmentSettlementWork.mock.calls[0][1]()).toBe(true); }); it.each([false, true])("registers native result commit only when maintenance is enabled (%s)", async enabled => { @@ -1090,7 +1093,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, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { @@ -1246,13 +1249,14 @@ describe("startServer feedback export wiring", () => { } }); - it.each(["retained cleanup", "workspace finalization", "native run projection", "native result commit", "native recovery notices", "external conversation settlement"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { + it.each(["retained cleanup", "workspace finalization", "environment settlement", "native run projection", "native result commit", "native recovery notices", "external conversation settlement"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); let releaseCleanup!: () => void; const ready = new Promise(resolve => { releaseCleanup = resolve; }); const stop = vi.fn(); const register = owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork : owner === "workspace finalization" ? heartbeatServiceMock.registerNativeWorkspaceFinalizationWork + : owner === "environment settlement" ? heartbeatServiceMock.registerNativeEnvironmentSettlementWork : owner === "native run projection" ? heartbeatServiceMock.registerNativeRunProjectionWork : owner === "native result commit" ? heartbeatServiceMock.registerNativeResultCommitWork : owner === "native recovery notices" ? heartbeatServiceMock.registerNativeRecoveryNoticeWork @@ -1262,7 +1266,8 @@ describe("startServer feedback export wiring", () => { else if (owner === "native recovery notices") heartbeatServiceMock.registerNativeRecoveryNoticeWork.mockReturnValueOnce({ ready }); else if (owner === "external conversation settlement") heartbeatServiceMock.registerExternalConversationSettlementWork.mockReturnValueOnce({ ready }); else (owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork - : heartbeatServiceMock.registerNativeWorkspaceFinalizationWork).mockReturnValueOnce({ ready, stop }); + : owner === "workspace finalization" ? heartbeatServiceMock.registerNativeWorkspaceFinalizationWork + : heartbeatServiceMock.registerNativeEnvironmentSettlementWork).mockReturnValueOnce({ ready, stop }); const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(() => 1 as unknown as ReturnType); let started = false; const starting = startServer().then(server => { started = true; return server; }); diff --git a/server/src/index.ts b/server/src/index.ts index 377e777d805..ee3da8b6e60 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1856,6 +1856,12 @@ async function startServerWithDatabaseTeardown( trackHeartbeatSchedulerWork(worker.ready); } + if (config.heartbeatSchedulerEnabled) { + const worker = heartbeat.registerNativeEnvironmentSettlementWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); + + trackHeartbeatSchedulerWork(worker.ready); + } if (config.heartbeatSchedulerEnabled) { const worker = heartbeat.registerNativeRunProjectionWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); @@ -2020,7 +2026,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, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index da9883281bb..1b450c94681 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -4,6 +4,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; export const DELIVERY_QUEUES = { externalConversationSettlement: "external-conversation-settlement", nativeRecoveryNotices: "native-recovery-notices", + nativeEnvironmentSettlement: "native-environment-settlement", nativeRunProjection: "native-run-projection", nativeResultCommit: "native-result-commit", nativeWorkspaceFinalization: "native-workspace-finalization", diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index 003288bb769..a1a69f922ae 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -1,14 +1,16 @@ +import { notifyNativeEnvironmentSettlement, type NativeEnvironmentSettlementScope } from "./native-environment-settlement-notifications.js"; import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyEnvironmentCleanup(transaction: object): Promise { +export async function notifyEnvironmentCleanup(transaction: object, scope?: NativeEnvironmentSettlementScope): Promise { + await notifyNativeEnvironmentSettlement(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); } -export function writeEnvironmentCleanup(db: Db, write: (tx: Db) => PromiseLike): Promise { +export function writeEnvironmentCleanup(db: Db, write: (tx: Db) => PromiseLike, scope?: NativeEnvironmentSettlementScope): Promise { return db.transaction(async tx => { - await notifyEnvironmentCleanup(tx); + await notifyEnvironmentCleanup(tx, scope); return write(tx as unknown as Db); }); } diff --git a/server/src/services/environment-lease-release-state.ts b/server/src/services/environment-lease-release-state.ts new file mode 100644 index 00000000000..f1626e75e67 --- /dev/null +++ b/server/src/services/environment-lease-release-state.ts @@ -0,0 +1,31 @@ +import { and, eq, lte, or, sql } from "drizzle-orm"; +import { environmentLeases, type Db } from "@paperclipai/db"; +import type { ProviderResourceDisposition } from "./environment-runtime.js"; +import { ENVIRONMENT_CLEANUP_RETRY_MS, pendingCleanupDeadlineSql } from "./environment-cleanup-state.js"; + + + +export function pendingRunLeaseReleaseCondition(destroy: boolean) { + return or(eq(environmentLeases.status, "active"), + and(eq(environmentLeases.status, "pending_cleanup"), + sql`(${environmentLeases.metadata} ? 'nativeWorkspaceExportResume' or ${environmentLeases.metadata} ? 'sandboxStopAndRetain')`), + destroy ? and(eq(environmentLeases.status, "retained"), eq(environmentLeases.cleanupStatus, "failed")) : undefined); +} + +export function runLeaseReleaseDeadlineSql(now = Date.now()) { + return sql`case when ${environmentLeases.status} = 'pending_cleanup' then greatest( + extract(epoch from ${environmentLeases.updatedAt}) * 1000 + ${ENVIRONMENT_CLEANUP_RETRY_MS}, + ${pendingCleanupDeadlineSql(now)}) else 0 end`; +} + +export function dueRunLeaseReleaseCondition(now = Date.now()) { + return lte(runLeaseReleaseDeadlineSql(now), now); +} + +export async function nextRunLeaseReleaseAt(db: Db, companyId: string, runId: string, + disposition: ProviderResourceDisposition): Promise { + const [row] = await db.select({ due: sql`min(${runLeaseReleaseDeadlineSql()})` }).from(environmentLeases) + .where(and(eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, runId), + pendingRunLeaseReleaseCondition(disposition === "destroy"))); + return row?.due == null ? null : Number(row.due); +} diff --git a/server/src/services/environment-run-orchestrator.ts b/server/src/services/environment-run-orchestrator.ts index 0d0cc2e4ba0..be76237d3ed 100644 --- a/server/src/services/environment-run-orchestrator.ts +++ b/server/src/services/environment-run-orchestrator.ts @@ -678,6 +678,8 @@ export function environmentRunOrchestrator( failureReason?: string; /** Explicit Stop during adapter startup; never used for ordinary cleanup. */ cancelActiveWork?: boolean; + + respectCleanupDeadlines?: boolean; /** Explicit paperclip_runner resource lifecycle. Omitted for legacy adapters. */ providerResourceDisposition?: ProviderResourceDisposition; nativeLifecycleTelemetry?: { @@ -695,13 +697,15 @@ export function environmentRunOrchestrator( let releasedLeases: EnvironmentRuntimeLeaseRecord[]; try { - releasedLeases = await environmentRuntime.releaseRunLeases( + const releaseArgs: Parameters = [ input.heartbeatRunId, status, (leaseId, error) => result.errors.push({ leaseId, error }), input.providerResourceDisposition, - ...(input.cancelActiveWork ? [true] as const : []), - ); + ]; + if (input.cancelActiveWork) releaseArgs[4] = true; + if (input.respectCleanupDeadlines) releaseArgs[5] = { respectCleanupDeadlines: true }; + releasedLeases = await environmentRuntime.releaseRunLeases(...releaseArgs); } catch (err) { result.errors.push({ leaseId: "*", error: err }); return result; diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 8503a30a49e..8db16c3209d 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1,3 +1,4 @@ +import { dueRunLeaseReleaseCondition, pendingRunLeaseReleaseCondition } from "./environment-lease-release-state.js"; import { flushDeferredEnvironmentCleanup, registerDeferredEnvironmentCleanup, writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { executeEnvironmentTask, pluginEnvironmentSupportsTasks, settleDispatchedEnvironmentTask } from "./environment-task-runtime.js"; import type { PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; @@ -4019,10 +4020,11 @@ export function environmentRuntimeService( return driver; } - async function destroyScopedReusableSandboxLease(input: { + async function destroyClaimedSandboxLease(input: { environment: Environment; leaseRow: typeof environmentLeases.$inferSelect; scopeCondition?: ReturnType; + runId?: string; failureReason: string; }): Promise { if (await nativeWorkspaceExportPendingForLease(db, input.leaseRow)) return null; @@ -4047,16 +4049,19 @@ export function environmentRuntimeService( eq(environmentLeases.id, input.leaseRow.id), eq(environmentLeases.companyId, input.leaseRow.companyId), eq(environmentLeases.environmentId, input.environment.id), - eq(environmentLeases.leasePolicy, "reuse_by_environment"), + input.runId ? eq(environmentLeases.heartbeatRunId, input.runId) + : eq(environmentLeases.leasePolicy, "reuse_by_environment"), eq(environmentLeases.status, input.leaseRow.status), input.scopeCondition, sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') != 'true'`, - sql`NOT EXISTS ( + + + input.runId ? undefined : sql`NOT EXISTS ( SELECT 1 FROM ${heartbeatRuns} WHERE ${heartbeatRuns.id} = ${environmentLeases.heartbeatRunId} AND ${heartbeatRuns.status} IN ('queued', 'scheduled_retry', 'running') )`, - )).returning()); + )).returning(), { companyId: input.leaseRow.companyId, runId: input.leaseRow.heartbeatRunId ?? undefined }); if (!claimed) return null; const lease = toEnvironmentLeaseSnapshot(claimed); let renewalInFlight = false; @@ -4078,12 +4083,14 @@ export function environmentRuntimeService( try { const driver = getDriver(getLeaseDriverKey(lease, input.environment)); if (!driver?.destroyRunLease) return lease; - return await driver.destroyRunLease({ + const released = await driver.destroyRunLease({ environment: input.environment, lease, failureReason: input.failureReason, - }) ?? lease; - } catch { + }); + return released ?? (input.runId ? null : lease); + } catch (error) { + if (input.runId) throw error; // Even a failed settlement write leaves the durable provider handle for // the sweep. Continue with later leases without logging provider errors. return lease; @@ -4095,7 +4102,7 @@ export function environmentRuntimeService( metadata: sql`${environmentLeases.metadata} || '{"pendingCleanupInFlight":false}'::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${attemptId}`, - ))).catch(() => { + )), { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }).catch(() => { logger.warn({ leaseId: lease.id }, "scoped cleanup ownership settlement failed"); }); } @@ -4199,6 +4206,7 @@ export function environmentRuntimeService( onLeaseReleaseError?: (leaseId: string, error: unknown) => void, providerResourceDisposition?: ProviderResourceDisposition, cancelActiveWork?: boolean, + releaseOptions?: { respectCleanupDeadlines?: boolean }, ): Promise { const leaseRows = await db .select() @@ -4206,11 +4214,8 @@ export function environmentRuntimeService( .where( and( eq(environmentLeases.heartbeatRunId, heartbeatRunId), - or(eq(environmentLeases.status, "active"), and(eq(environmentLeases.status, "pending_cleanup"), - sql`(${environmentLeases.metadata} ? 'nativeWorkspaceExportResume' or ${environmentLeases.metadata} ? 'sandboxStopAndRetain')`), - providerResourceDisposition === "destroy" - ? and(eq(environmentLeases.status, "retained"), eq(environmentLeases.cleanupStatus, "failed")) - : undefined), + pendingRunLeaseReleaseCondition(providerResourceDisposition === "destroy"), + releaseOptions?.respectCleanupDeadlines ? dueRunLeaseReleaseCondition() : undefined, ), ); if (leaseRows.length === 0) { @@ -4238,15 +4243,17 @@ export function environmentRuntimeService( ? await environmentsSvc.getById(leaseRow.environmentId) : null; let leaseSnapshot = toEnvironmentLeaseSnapshot(leaseRow); + let preparedStopAttemptId: unknown; if (hasNativeWorkspaceExportResume(leaseSnapshot) && providerResourceDisposition === "destroy") { const completed = await releaseCompletedNativeWorkspaceExportRetention(db, leaseSnapshot); - if (completed) leaseSnapshot = toEnvironmentLeaseSnapshot(completed); + if (completed) { leaseRow = completed; leaseSnapshot = toEnvironmentLeaseSnapshot(completed); } } if (providerResourceDisposition === "stop_and_retain" && !hasStopOnlyCleanup(leaseSnapshot) && getLeaseDriverKey(leaseSnapshot, environment) === "sandbox") { const prepared = await prepareSandboxStopAndRetain(db, leaseSnapshot); if (!prepared) continue; leaseSnapshot = toEnvironmentLeaseSnapshot(prepared); + preparedStopAttemptId = prepared.metadata?.pendingCleanupAttemptId; } if (hasStopOnlyCleanup(leaseSnapshot)) { const intentKey = stopOnlyCleanupKey(leaseSnapshot); @@ -4258,8 +4265,15 @@ export function environmentRuntimeService( }).where(and(eq(environmentLeases.id, leaseRow.id), eq(environmentLeases.heartbeatRunId, heartbeatRunId), inArray(environmentLeases.status, ["active", "pending_cleanup"]), sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') = 'false'`, + releaseOptions?.respectCleanupDeadlines ? or(dueRunLeaseReleaseCondition(), + + + typeof preparedStopAttemptId === "string" ? and( + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${preparedStopAttemptId}`, + sql`${environmentLeases.metadata}->>'pendingCleanupRetryAfterMs' = '0'`, + ) : undefined) : undefined, sql`${environmentLeases.metadata}->${intentKey} = ${JSON.stringify(leaseSnapshot.metadata?.[intentKey])}::jsonb`, - )).returning()); + )).returning(), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined }); if (!claimed) continue; const snapshot = toEnvironmentLeaseSnapshot(claimed); try { @@ -4332,9 +4346,10 @@ export function environmentRuntimeService( ); } const lease = providerResourceDisposition === "destroy" && driver?.destroyRunLease - ? await driver.destroyRunLease({ - environment, - lease: leaseSnapshot, + ? await destroyClaimedSandboxLease({ + environment, leaseRow, runId: heartbeatRunId, + scopeCondition: and(pendingRunLeaseReleaseCondition(true), + releaseOptions?.respectCleanupDeadlines ? dueRunLeaseReleaseCondition() : undefined), failureReason: "paperclip_runner_destroy_after_turn", }) : driver @@ -4495,7 +4510,7 @@ export function environmentRuntimeService( ? await environmentsSvc.getById(leaseRow.environmentId) : null; if (!environment) continue; - const lease = await destroyScopedReusableSandboxLease({ + const lease = await destroyClaimedSandboxLease({ environment, leaseRow, scopeCondition: and(...scopeConditions), @@ -4576,7 +4591,7 @@ export function environmentRuntimeService( skippedLiveRun += 1; continue; } - const lease = await destroyScopedReusableSandboxLease({ environment, leaseRow, failureReason }); + const lease = await destroyClaimedSandboxLease({ environment, leaseRow, failureReason }); if (!lease) { // Lost to a racing resume or a concurrent terminal transition — the // lease is no longer ours to destroy. diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index 020cbca36d6..8806bfca452 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -369,7 +369,7 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor const [current] = await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskRetiring":true}'::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId))) - .returning({ metadata: environmentLeases.metadata }); + .returning({ metadata: environmentLeases.metadata, heartbeatRunId: environmentLeases.heartbeatRunId }); if (current?.metadata?.runnerTaskDispatched !== true) { if (current?.metadata) await settlePluginTaskCredentialHandoff(db, { companyId: lease.companyId, leaseId: lease.id }, current.metadata, async () => { throw new Error("Undispatched task must not request credentials"); }); @@ -385,7 +385,7 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor } catch { await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", failureReason: "plugin_runner_task_cleanup_unconfirmed", releasedAt: new Date(), updatedAt: new Date(), - }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId)))); + }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId))), { companyId: lease.companyId, runId: current.heartbeatRunId ?? undefined }); throw new Error("Plugin Runner termination is unconfirmed; the saved lease requires cleanup"); } } diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index 4a3e57f803b..b6da210efa5 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1,6 +1,7 @@ +import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; -import { writeEnvironmentCleanup, notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; +import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { randomUUID } from "node:crypto"; @@ -1507,6 +1508,9 @@ export function environmentService(db: Db) { input.replacesReusableLeaseId || input.reusesReusableLeaseId ? await db.transaction(async (tx) => { + if (input.heartbeatRunId) await notifyNativeEnvironmentSettlement(tx, { + companyId: input.companyId, runId: input.heartbeatRunId, + }); if (input.assertCompanyBinding) { // Lock the environment row first. Managed reconciliation locks the // same sandbox environment rows with `for update` before it writes a @@ -1648,11 +1652,12 @@ export function environmentService(db: Db) { .returning() .then((rows) => rows[0] ?? null); }) - : await db - .insert(environmentLeases) - .values(values) - .returning() - .then((rows) => rows[0] ?? null); + : await db.transaction(async tx => { + if (input.heartbeatRunId) await notifyNativeEnvironmentSettlement(tx, { + companyId: input.companyId, runId: input.heartbeatRunId, + }); + return tx.insert(environmentLeases).values(values).returning().then(rows => rows[0] ?? null); + }); if (!row) { throw new Error("Failed to acquire environment lease"); } @@ -1675,7 +1680,7 @@ export function environmentService(db: Db) { : sql`${environmentLeases.metadata} - 'remoteExecutionTermination'`; const cleanupSettled = options?.expectedPendingCleanupAttemptId && options.cleanupStatus === "success" && status !== "pending_cleanup"; - const row = await writeEnvironmentCleanup(db, async tx => { + const row = await db.transaction(async tx => { const condition = and(eq(environmentLeases.id, id), options?.expectedPendingCleanupAttemptId ? and(eq(environmentLeases.status, "pending_cleanup"), sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.expectedPendingCleanupAttemptId}`) @@ -1685,6 +1690,7 @@ export function environmentService(db: Db) { cleanupStatus: environmentLeases.cleanupStatus, releasedAt: environmentLeases.releasedAt, }).from(environmentLeases).where(condition).for("update"); if (!previous) return null; + await notifyEnvironmentCleanup(tx, { companyId: previous.companyId, runId: previous.heartbeatRunId ?? undefined }); if (previous.heartbeatRunId && (previous.status !== status || (options?.cleanupStatus !== undefined && previous.cleanupStatus !== options.cleanupStatus) || (status !== "retained" && !previous.releasedAt))) { @@ -1764,7 +1770,7 @@ export function environmentService(db: Db) { .where(eq(environments.id, input.environmentId)) .for("update"); const environmentIdForRow = environmentRows[0]?.id ?? null; - await notifyEnvironmentCleanup(tx); + await notifyEnvironmentCleanup(tx, { companyId: input.companyId, runId: input.heartbeatRunId ?? undefined }); return tx .insert(environmentLeases) .values({ @@ -1800,8 +1806,12 @@ export function environmentService(db: Db) { id: string, metadata: Record | null, ): Promise => { - const row = await writeEnvironmentCleanup(db, tx => tx - .update(environmentLeases) + const row = await db.transaction(async tx => { + const [lease] = await tx.select({ companyId: environmentLeases.companyId, heartbeatRunId: environmentLeases.heartbeatRunId }) + .from(environmentLeases).where(eq(environmentLeases.id, id)).for("update"); + if (!lease) return null; + await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); + return tx.update(environmentLeases) .set({ metadata: withoutCleanupVerification(metadata), lastUsedAt: new Date(), @@ -1809,7 +1819,8 @@ export function environmentService(db: Db) { }) .where(eq(environmentLeases.id, id)) .returning() - .then((rows) => rows[0] ?? null)); + .then((rows) => rows[0] ?? null); + }); return row ? toEnvironmentLease(row) : null; }, @@ -1818,10 +1829,13 @@ export function environmentService(db: Db) { status: Extract = "released", ): Promise => { const now = new Date(); - const rows = await writeEnvironmentCleanup(db, async tx => { + const rows = await db.transaction(async tx => { const [run] = await tx.select({ companyId: heartbeatRuns.companyId }).from(heartbeatRuns) .where(eq(heartbeatRuns.id, heartbeatRunId)).limit(1); - if (run) await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: heartbeatRunId }); + if (run) { + await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: heartbeatRunId }); + await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: heartbeatRunId }); + } return tx.update(environmentLeases) .set({ status, diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index ff09ee63367..2f2a435982d 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyNativeFinalizationWork } from "./native-finalization-notifications.js"; @@ -192,6 +193,7 @@ export async function reconcileAbandonedExecutionControl( .where(eq(nativeRunFinalizations.runId, run.id)); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); + await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); await notifyExecutionRunWork(tx); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 42c5099a4c6..dd894aeb586 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1029,6 +1029,7 @@ export function heartbeatService( registerScheduledRetryCancellationWork, registerRetainedNativeSessionCleanupWork, registerNativeWorkspaceFinalizationWork, + registerNativeEnvironmentSettlementWork, } = createHeartbeatRecovery(db, { pluginWorkerManager: options.pluginWorkerManager, enterShutdown: () => { shutdownInProgress = true; }, @@ -4131,6 +4132,14 @@ export function heartbeatService( const state = await getSchedulingSuppression(true); return !state.suppressed || state.reason === "task_drain"; }), + registerNativeEnvironmentSettlementWork: ( + coordinator: Parameters[0], canRun: () => boolean, + ) => registerNativeEnvironmentSettlementWork(coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerNativeWorkspaceFinalizationWork: ( coordinator: Parameters[0], canRun: () => boolean, ) => registerNativeWorkspaceFinalizationWork(coordinator, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index f5e08ce8750..c5285332d9d 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,3 +1,4 @@ +import { nativeEnvironmentSettlementDisposition, nativeEnvironmentSettlementWork } from "../native-runtime/native-environment-settlement-work.js"; import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js"; import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js"; import { retainedNativeSessionCleanupWork } from "../native-runtime/retained-native-session-cleanup-work.js"; @@ -153,6 +154,7 @@ export interface HeartbeatRecoveryDependencies extends Pick Promise; acknowledgeRemoteStop: (runId: string, companyId: string) => Promise; resumeRemoteStopComments: (run: HeartbeatRun) => Promise; @@ -1571,7 +1573,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD eq(environmentLeases.status, "active"), ), ) - .returning({ id: environmentLeases.id })); + .returning({ id: environmentLeases.id }), { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); if (flipped.length > 0) recovered += 1; } @@ -1814,7 +1816,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ? 30 * 60_000 : Math.min(30 * 60_000, Math.max(30_000, backoffMs))), })}::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), - sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${claimed}`))); + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${claimed}`)), { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); if (lease.heartbeatRunId) { // Delivery failure must not revert successful provider cleanup. A new // message can still use the persisted receipt on its next admission. @@ -1887,26 +1889,32 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD return blocked; } - async function settleRecoveredNativeWorkspace(input: { + async function settleRecoveredNativeEnvironment(input: { runId: string; companyId: string; agentId: string; succeeded: boolean; - }) { + }, onlyTerminal = false) { const settledRun = await getRunWorkspaceSettlement(input.runId); - const workspaceSyncReference = readNativeWorkspaceSyncReference( - parseObject(settledRun?.runnerProfileJson).nativeWorkspaceSync, - ); + if (!settledRun || onlyTerminal && !["succeeded", "failed", "cancelled", "timed_out", "interrupted"].includes(settledRun.status)) return; await releaseEnvironmentLeasesForRun({ runId: input.runId, companyId: input.companyId, agentId: input.agentId, status: settledRun?.status, failureReason: settledRun?.error ?? undefined, - providerResourceDisposition: input.succeeded && (!parseObject(settledRun?.resultJson).workspaceExportRetry || workspaceSyncReference?.resourceDisposition === "destroy") - ? (workspaceSyncReference?.resourceDisposition ?? "stop_and_retain") - : "stop_and_retain", + providerResourceDisposition: nativeEnvironmentSettlementDisposition(settledRun, input.succeeded), + ...(onlyTerminal ? { respectCleanupDeadlines: true } : {}), }); + } + + const nativeEnvironmentSettlement = nativeEnvironmentSettlementWork(db, input => settleRecoveredNativeEnvironment(input, true)); + + async function settleRecoveredNativeWorkspace(input: { + runId: string; companyId: string; agentId: string; succeeded: boolean; + }, options?: { skipEnvironmentSettlement?: boolean }) { + if (!options?.skipEnvironmentSettlement) await settleRecoveredNativeEnvironment(input); + const settledRun = await getRun(input.runId); await releaseRuntimeServicesForRun(input.runId).catch(() => undefined); await finalizeAgentStatus( input.agentId, @@ -1967,7 +1975,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean; skipNativeEnvironmentSettlement?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -2000,7 +2008,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, - onWorkspaceSettled: settleRecoveredNativeWorkspace, + onWorkspaceSettled: input => settleRecoveredNativeWorkspace(input, { + skipEnvironmentSettlement: opts?.skipNativeEnvironmentSettlement, + }), }).catch((error) => { logger.warn( { err: error }, @@ -2612,6 +2622,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD return { registerRetainedNativeSessionCleanupWork: retainedNativeCleanup.register, registerNativeWorkspaceFinalizationWork: nativeWorkspaceFinalization.register, + registerNativeEnvironmentSettlementWork: nativeEnvironmentSettlement.register, registerScheduledRetryCancellationWork: scheduledRetryCancellation.register, sweepPendingCleanupLeases, markNativeOwnershipUnverified, diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 8159d2e04cf..f8931615322 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -361,6 +361,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont status: string | null | undefined; failureReason?: string | null; providerResourceDisposition?: ProviderResourceDisposition; + respectCleanupDeadlines?: boolean; nativeLifecycleTelemetry?: { provider: string; harness: string; @@ -466,6 +467,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont status: leaseReleaseStatusForRunStatus(status), failureReason: input.failureReason ?? undefined, providerResourceDisposition, + ...(input.respectCleanupDeadlines ? { respectCleanupDeadlines: true } : {}), nativeLifecycleTelemetry: input.nativeLifecycleTelemetry, }) .catch((err) => { @@ -726,7 +728,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont or(isNull(environmentLeases.releasedAt), inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"])), )).limit(1); if (otherOwner) continue; - await notifyEnvironmentCleanup(tx); + await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); await tx.update(environmentLeases).set({ status: "pending_cleanup", updatedAt: new Date() }) .where(eq(environmentLeases.id, lease.id)); } diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index ec632b793b0..1922364615a 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "../external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; @@ -1033,6 +1034,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl await notifyExternalAgentSettlement(tx, { companyId: previous.companyId, runId: previous.id }); if (previous?.runtimeMode === "native") await notifyNativeRecoveryNotices(tx, { companyId: previous.companyId, runId: previous.id }); if (previous?.runtimeMode === "native") await notifyNativeRunProjection(tx, { companyId: previous.companyId, runId: previous.id }); + if (previous?.runtimeMode === "native") await notifyNativeEnvironmentSettlement(tx, { companyId: previous.companyId, runId: previous.id }); if (previous && previous.status !== status) await notifyUnsafeWorkspaceExportRunChange(tx, previous); if (previous && (previous.status === "scheduled_retry" || status === "scheduled_retry")) await notifyScheduledRetryCancellation(tx, { companyId: previous.companyId, runId: previous.id }); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index e8d200a60f3..84b5840a77a 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; @@ -636,6 +637,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyExternalConversationSettlement(tx); await notifyNativeRecoveryNotices(tx); await notifyNativeRunProjection(tx); + await notifyNativeEnvironmentSettlement(tx); await notifyNativeResultCommit(tx); await notifyNativeWorkspaceFinalization(tx); } diff --git a/server/src/services/native-environment-settlement-notifications.ts b/server/src/services/native-environment-settlement-notifications.ts new file mode 100644 index 00000000000..f8dc0a966c4 --- /dev/null +++ b/server/src/services/native-environment-settlement-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeEnvironmentSettlementScope = { companyId: string; runId?: string; issueId?: string }; +export function notifyNativeEnvironmentSettlement(tx: object, scope?: NativeEnvironmentSettlementScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeEnvironmentSettlement, + scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"); +} diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index 05899ebf308..72d4cbcac49 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; @@ -14,6 +15,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { if (scope.issueId || scope.runId) { + await notifyNativeEnvironmentSettlement(transaction, scope); await notifyExternalAgentSettlement(transaction, scope); await notifyExternalConversationSettlement(transaction, scope); await notifyNativeRecoveryNotices(transaction, scope); diff --git a/server/src/services/native-runtime/native-environment-settlement-work.ts b/server/src/services/native-runtime/native-environment-settlement-work.ts new file mode 100644 index 00000000000..e957ed890c1 --- /dev/null +++ b/server/src/services/native-runtime/native-environment-settlement-work.ts @@ -0,0 +1,114 @@ +import { and, asc, eq, exists, gt, inArray, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import type { ProviderResourceDisposition } from "../environment-runtime.js"; +import { nextRunLeaseReleaseAt, pendingRunLeaseReleaseCondition } from "../environment-lease-release-state.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativeEnvironmentSettlementScope } from "../native-environment-settlement-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; + +type Run = Pick; +export function nativeEnvironmentSettlementDisposition(run: Run, succeeded: boolean): ProviderResourceDisposition { + const reference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); + return succeeded && (!run.resultJson?.workspaceExportRetry || reference?.resourceDisposition === "destroy") + ? reference?.resourceDisposition ?? "stop_and_retain" : "stop_and_retain"; +} + +type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; +type Settlement = { runId: string; companyId: string; agentId: string; succeeded: boolean }; + + + +export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settlement) => Promise) { + const retryFloors = new Map(); + const matches = (item: Entry, scope?: NativeEnvironmentSettlementScope) => !scope || item.companyId === scope.companyId + && (!scope.issueId || item.issueId === scope.issueId) && (!scope.runId || item.id === scope.runId); + function candidates(scope?: NativeEnvironmentSettlementScope, after?: string) { + return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + issueId: nativeRunFinalizations.issueId, phase: nativeRunFinalizations.phase, + runnerProfileJson: sql`case when ${heartbeatRuns.runnerProfileJson} is null then null + else jsonb_build_object('nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync') end`, + resultJson: sql`case when ${heartbeatRuns.resultJson} is null then null + else jsonb_build_object('workspaceExportRetry', ${heartbeatRuns.resultJson}->'workspaceExportRetry') end`, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.runId, nativeRunFinalizations.runId), eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), + eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), eq(nativeRunResults.schemaStatus, "accepted"))) + .where(and(eq(heartbeatRuns.runtimeMode, "native"), + inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]), + or(eq(nativeRunFinalizations.phase, "committed"), + and(eq(nativeRunFinalizations.phase, "terminal_failure"), + sql`${nativeRunFinalizations.failureDetail} ? 'workspaceFinalizeAttempt'`)), + exists(db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, heartbeatRuns.companyId), eq(environmentLeases.heartbeatRunId, heartbeatRuns.id), + pendingRunLeaseReleaseCondition(true)))), + scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, + scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, + after ? gt(heartbeatRuns.id, after) : undefined, + )).orderBy(asc(heartbeatRuns.id)).limit(100); + } + async function entry(row: Awaited>[number]): Promise { + const dueAt = await nextRunLeaseReleaseAt(db, row.companyId, row.id, + nativeEnvironmentSettlementDisposition(row, row.phase === "committed")); + if (dueAt === null) { retryFloors.delete(row.id); return null; } + return { id: row.id, companyId: row.companyId, issueId: row.issueId, + dueAt: Math.max(dueAt, retryFloors.get(row.id)?.dueAt ?? 0) }; + } + async function load(scope?: NativeEnvironmentSettlementScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (!signal?.aborted && canRun()) { + const rows = await candidates(scope, after); + for (const row of rows) { + if (signal?.aborted || !canRun()) throw new Error("Native environment discovery stopped"); + const pending = await entry(row); + if (pending) entries.push(pending); + } + if (rows.length < 100) { + const retained = new Set(entries.map(item => item.id)); + for (const item of retryFloors.values()) if (matches(item, scope) && !retained.has(item.id)) retryFloors.delete(item.id); + return entries; + } + after = rows[rows.length - 1].id; + } + throw new Error("Native environment discovery stopped"); + } + async function check(current: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Native environment settlement stopped"); + const scope = { companyId: current.companyId, issueId: current.issueId, runId: current.id }; + const [row] = await candidates(scope); + if (!row) { retryFloors.delete(current.id); return null; } + const pending = await entry(row); + if (!pending || pending.dueAt > Date.now()) return pending; + if (signal?.aborted) throw new Error("Native environment settlement stopped"); + await settle({ runId: row.id, companyId: row.companyId, agentId: row.agentId, succeeded: row.phase === "committed" }); + + + const [remaining] = await candidates(scope); + const next = remaining ? await entry(remaining) : null; + if (!next) { retryFloors.delete(current.id); return null; } + const floor = Date.now() + 30_000; + + + retryFloors.set(current.id, { ...next, dueAt: floor }); + return { ...next, dueAt: Math.max(next.dueAt, floor) }; + } + function register(coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise = async () => true) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeEnvironmentSettlement, batchSize: 1, + decodeScope: (hint: string): NativeEnvironmentSettlementScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches, + load: async (scope, signal) => await canRecover() ? load(scope, signal, canRun) : [], + check: async (item, signal) => { + if (!canRun()) throw new Error("Native environment settlement paused"); + return await canRecover() ? check(item, signal) : null; + }, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index e3052482467..e3448b840b1 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; @@ -925,6 +926,7 @@ export async function claimNativeRestartRecoveries(input: { ), ); await notifyExternalAgentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); + await notifyNativeEnvironmentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, row.run); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 7a8d6146c9e..ba2017eb9a8 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; @@ -1640,6 +1641,7 @@ export async function finalizeNativeRun(input: { if (input.projectRunStatus) await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyNativeRecoveryNotices(tx, { companyId: run.companyId, runId: run.id }); await notifyNativeRunProjection(tx, { companyId: run.companyId, runId: run.id }); + await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id }); if (input.projectRunStatus) await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 38fe83e8a07..5708272da2c 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; @@ -131,6 +132,7 @@ export async function projectCommittedRun(input: { await notifyExternalAgentSettlement(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyNativeRecoveryNotices(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyNativeRunProjection(tx, { companyId: input.run.companyId, runId: input.run.id }); + await notifyNativeEnvironmentSettlement(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyRetainedNativeSessionCleanup(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 2a58aad73a3..21cc7420294 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -12925,6 +12925,8 @@ async function createRunnerdBackendWithinSessionClaim( .select({ metadata: environmentLeases.metadata, providerLeaseId: environmentLeases.providerLeaseId, + companyId: environmentLeases.companyId, + heartbeatRunId: environmentLeases.heartbeatRunId, }) .from(environmentLeases) .where(eq(environmentLeases.id, remoteTarget.leaseId)) @@ -12959,7 +12961,7 @@ async function createRunnerdBackendWithinSessionClaim( }) .where(eq(environmentLeases.id, leaseId)) .returning({ id: environmentLeases.id }) - .then((rows) => rows[0] ?? null)); + .then((rows) => rows[0] ?? null), { companyId: leaseRow.companyId, runId: leaseRow.heartbeatRunId ?? undefined }); if (!updated) throw new Error("runner_harness_backup_lease_missing"); }; diff --git a/server/src/services/native-runtime/native-workspace-export-resume.ts b/server/src/services/native-runtime/native-workspace-export-resume.ts index c41930428d0..324d3e6d012 100644 --- a/server/src/services/native-runtime/native-workspace-export-resume.ts +++ b/server/src/services/native-runtime/native-workspace-export-resume.ts @@ -66,7 +66,7 @@ export async function preserveNativeWorkspaceExportLease(db: Db, run: typeof hea pluginId: lease.metadata?.pluginId }, pendingCleanupAttemptId: requestId, pendingCleanupInFlight: false, pendingCleanupLeaseExpiresAtMs: 0, pendingCleanupRetryAfterMs: 0, pendingCleanupRetryAttempts: 0, pendingCleanupRetryCapWarned: false }, - }).where(eq(environmentLeases.id, lease.id))); + }).where(eq(environmentLeases.id, lease.id)), { companyId: run.companyId, runId: run.id }); } /** Only verified copyback plus commitment permits the original ephemeral @@ -125,6 +125,6 @@ export async function settleNativeWorkspaceExportResume(db: Db, lease: Lease, op sql`${environmentLeases.metadata}->'nativeWorkspaceExportResume'->>'requestId' = ${marker.requestId}`, sql`${environmentLeases.metadata}->>'pluginId' = ${marker.pluginId}`, )).returning(); - }); + }, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); return row ?? null; } diff --git a/server/src/services/native-runtime/native-workspace-export-retry.ts b/server/src/services/native-runtime/native-workspace-export-retry.ts index 99e286a17ff..47de8c73cf4 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.ts @@ -94,7 +94,7 @@ export async function retryNativeWorkspaceExport(input: { // Persist cleanup authority before any provider call. A lost resume // reply or controller restart must not leave a running sandbox behind // an old stopped receipt. The pending claim fences the cleanup sweeper. - await notifyEnvironmentCleanup(tx); + await notifyEnvironmentCleanup(tx, { companyId: input.companyId, runId: input.runId }); const [lease] = await tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", releasedAt: now, failureReason: "workspace_export_resume_pending", updatedAt: now, metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined, @@ -144,7 +144,7 @@ export async function retryNativeWorkspaceExport(input: { await tx.update(issues).set({ executionRunId: current.run.id, updatedAt: now }).where(eq(issues.id, current.issue.id)); // Reclaim the same lease only for control-plane copyback. The normal // finalization settlement will stop/retain it again on success or failure. - await notifyEnvironmentCleanup(tx); + await notifyEnvironmentCleanup(tx, { companyId: input.companyId, runId: input.runId }); await tx.update(environmentLeases).set({ status: "active", releasedAt: null, cleanupStatus: null, failureReason: null, metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined, // Keep stop-only authority through copyback and commitment. A crash diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index 7ff930146ef..7e56e3500e3 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -643,14 +643,14 @@ async function persistLeaseStamp(input: { }): Promise { await input.db.transaction(async (tx) => { const lease = await tx - .select({ metadata: environmentLeases.metadata }) + .select({ metadata: environmentLeases.metadata, companyId: environmentLeases.companyId, heartbeatRunId: environmentLeases.heartbeatRunId }) .from(environmentLeases) .where(eq(environmentLeases.id, input.leaseId)) .for("update") .limit(1) .then((rows) => rows[0] ?? null); if (!lease) throw new Error("native_workspace_sync_lease_missing"); - await notifyEnvironmentCleanup(tx); + await notifyEnvironmentCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); await tx .update(environmentLeases) .set({ diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index d9b422ea913..991fa8f63b5 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; @@ -6264,6 +6265,7 @@ export function recoveryService( if (!candidate) return null; await notifyInstructionCollection(tx, { companyId: candidate.companyId, runId: candidate.id }); await notifyExternalAgentSettlement(tx, { companyId: candidate.companyId, runId: candidate.id }); + await notifyNativeEnvironmentSettlement(tx, { companyId: candidate.companyId, runId: candidate.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: candidate.companyId, runId: candidate.id }); const boundIssueId = boundRunIssueId(candidate); if (boundIssueId) { diff --git a/server/src/services/sandbox-stop-and-retain.ts b/server/src/services/sandbox-stop-and-retain.ts index 2090ac4f7bd..ab423804e38 100644 --- a/server/src/services/sandbox-stop-and-retain.ts +++ b/server/src/services/sandbox-stop-and-retain.ts @@ -116,7 +116,7 @@ async function prepareStopAndRetain(db: Db, lease: Lease, fromStatus: "active" | sql`${environmentLeases.provider} is not distinct from ${lease.provider}`, sql`${environmentLeases.providerLeaseId} is not distinct from ${lease.providerLeaseId}`, sql`${environmentLeases.metadata} is not distinct from ${lease.metadata === null ? null : JSON.stringify(lease.metadata)}::jsonb`, - )).returning()); + )).returning(), { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); return updated ?? null; } @@ -151,7 +151,7 @@ export async function settleStopOnlyCleanup(db: Db, lease: Lease, options: { att sql`${environmentLeases.metadata}->'sandboxStopAndRetain' = ${JSON.stringify(lease.metadata?.[SANDBOX_STOP_AND_RETAIN_KEY])}::jsonb`, sql`${environmentLeases.metadata}->>'pluginId' is not distinct from ${intent.pluginId}`, )).returning(); - }); + }, { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); return updated ?? null; } diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index ee185f8be91..989c74335de 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -317,7 +317,7 @@ export async function scheduleRecoveredWorkspaceCleanup(db: Db, run: typeof hear eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.providerLeaseId, ref.providerLeaseId), eq(environmentLeases.status, "released"), eq(environmentLeases.cleanupStatus, "success"), sql`${environmentLeases.metadata} = ${JSON.stringify(lease.metadata)}::jsonb`, - sql`${environmentLeases.metadata}->>'workspaceRecoveryCleanup' is null`))); + sql`${environmentLeases.metadata}->>'workspaceRecoveryCleanup' is null`)), { companyId: run.companyId, runId: run.id }); } export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedOnly = false, obligationsOnly = false, inFlightRunIds: string[] = []) { const cleanup = sql`(${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,resourceDisposition}' = 'destroy' and exists (