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 (