From 33567f91551d60053722c013f0d9b402c1db29e6 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 21:00:54 -0500 Subject: [PATCH] fix(server): schedule cancelled workspace preservation Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 16 ++ ...lled-workspace-preservation-sleep-audit.md | 147 ++++++++++ packages/db/src/client.ts | 1 + .../db/src/migrations/0397_lumpy_talos.sql | 5 + ...{0392_snapshot.json => 0397_snapshot.json} | 130 ++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/heartbeat_runs.ts | 6 + .../server-startup-feedback-export.test.ts | 10 +- server/src/index.ts | 4 +- .../active-run-watchdog/adapters/postgres.ts | 2 + .../modules/run-dispatch/adapters/postgres.ts | 4 + .../adapters/queued-comment-postgres.ts | 2 + server/src/services/agent-deletion.ts | 2 + ...ed-workspace-preservation-notifications.ts | 7 + server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + .../execution-control-reconciliation.ts | 2 + server/src/services/heartbeat.ts | 9 + server/src/services/heartbeat/queue.ts | 7 + server/src/services/heartbeat/recovery.ts | 4 +- .../src/services/heartbeat/run-lifecycle.ts | 2 + server/src/services/instance-settings.ts | 2 + .../native-failed-workspace-recovery.test.ts | 253 +++++++++++++++++- .../native-failed-workspace-recovery.ts | 77 +++++- .../native-finalization-reconciler.ts | 2 + .../native-runtime/native-restart-recovery.ts | 2 + .../native-runtime/native-run-finalizer.ts | 4 + .../native-runtime/native-session-executor.ts | 3 + .../native-runtime/native-workspace-sync.ts | 4 +- .../status-decision-committer.ts | 2 + server/src/services/recovery/service.ts | 2 + 31 files changed, 708 insertions(+), 13 deletions(-) create mode 100644 doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md create mode 100644 packages/db/src/migrations/0397_lumpy_talos.sql rename packages/db/src/migrations/meta/{0392_snapshot.json => 0397_snapshot.json} (99%) create mode 100644 server/src/services/cancelled-workspace-preservation-notifications.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 03539adf873..76350fdb43a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2150,3 +2150,19 @@ coverage passed 101 separately. Exact-head remote acceptance is still required. Other native recovery, dispatch, retention, integration and aggregate-tick duties remain in scope. The older CI/review gates recorded above are not discharged by this slice's local tests. + +## Slice 42 acceptance and slice 43 cancelled workspace preservation + +Stopped instruction collection PR #17674 at +`6dc58115cdebb988a63bb45dce4521f5bb5013f6` has all 53 checks successful, +Greptile 5/5 at that head and zero unresolved threads. It remains draft and open. + +Slice 43 moves cancelled native workspace preservation out of periodic reaping. +Its owner and producer contract is recorded in +[the preservation audit](2026-10-25-cancelled-workspace-preservation-sleep-audit.md). +It preserves known file debt separately from coordinator/lease readiness and +keeps teardown with the existing stop-only worker. Exact-head remote acceptance +is still required. The full remaining stack and older unresolved gates remain +in scope. + +Slice 43 final local acceptance: 120 owner/startup cases plus 209 affected consumer cases pass, without skips. Workspace typecheck/build, final server types, migration application/replay and index-plan checks passed. The negative fixture FK rejection and its correction remain in the owner audit. Remote CI and review are still required. diff --git a/doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md b/doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md new file mode 100644 index 00000000000..56ec3a5b0d1 --- /dev/null +++ b/doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md @@ -0,0 +1,147 @@ +# Cancelled native workspace preservation + +Date: 2026-10-25. Parent: stopped instruction collection PR #17674 at +`6dc58115cdebb988a63bb45dce4521f5bb5013f6`. + +## One reason and its owner + +A native run can acknowledge user Stop while its prepared workspace still needs +a durable file-recovery hold. The periodic orphan reaper previously scanned for +these cancellations on every pass. This slice migrates that discovery to the +existing transaction signals, delivery coordinator and scoped deadline worker. +It adds no table and no independent timer. + +The owner remains `native-runtime/native-failed-workspace-recovery.ts`. +`preserveNativeFailedWorkspace` retains the task/coordinator/run lock order, +source identity checks, original lease pinning, no-replay recovery card and +activity record. It performs database work only. The stop-only cleanup owner +still owns provider teardown. Recording a file hold does not prove termination +and never grants permission to destroy the allocation or replay the run. + +This is not only historical data migration. Current foreground cancellation +tries retention before lease release. If it loses a task lock or database write, +it leaves the source active and needs the same retry mechanism. + +## Obligation, discovery and completion + +Discovery reads native cancelled runs whose prepared workspace reference and +acknowledged cancellation suggest unfinished preservation. It excludes a valid +existing workspace-recovery receipt paired with a recognized restore-failure +code. A schema marker alone is insufficient to declare preservation complete. +The owner then validates the full existing acknowledgement: same run, company, +native issue, user actor, run-only scope, cancellation reason and audit IDs. +A missing native issue, invalid reference, foreign receipt or admitted +workspace-less execution creates no preservation entry. + +Discovery pages 100 rows at a time by run ID, reads the fields needed for these +checks, and retains only IDs/company IDs/deadlines in the scheduler. Fair batches +attempt at most five entries. Cost scales with actual candidates, including +unusual prepared history that must be classified; it is not constant time. + +The new pending predicate is deliberately separate from immediate dispatch +eligibility. A valid cancelled prepared workspace can still owe preservation +while a controller owns its coordinator or stop-only proof is not ready. Those +rows remain scheduled instead of depending on another global scan to discover +when they become actionable. Accepted results belong to native finalization; +this worker does not take over their workspace obligation. The audited runtime +writers do not erase an accepted coordinator result ID to create this duty. + +A check rereads the run and exact company/native-issue coordinator. It delegates +only at terminal_failure, without a coordinator owner or accepted result, and +when a lease meets the prior scan's status/policy eligibility. The retention +owner rereads under its locks and verifies the exact reference lease, provider +allocation, workspace and remote directory. Recovery requires the current run +still be cancelled with the exact acknowledgement and the coordinator still be +terminal_failure. These extra checks close stale-discovery races without +changing the foreground preterminal API. + +After delegation, the worker rereads the durable run receipt. A valid recorded +hold, finalized reference or removed obligation disarms it. Actual unfinished +preservation retains a 30-second retry, including missing proof, missing source +or a still-owned coordinator. Database/owner failures use the shared bounded +failure backoff and do not starve other companies. These are retries for known +file debt, not probes of idle configuration. Retry timing is process-local; +restart reconstructs the durable obligation and tries it again. No unrelated +native-finalizer business deadline is overwritten. + +## Producer and shared-boundary audit + +Cancellation acknowledgement, terminal status and workspace preparation can be +separate transactions. All three must reactivate the worker. Signals register +before the relevant write and settle with the existing outer transaction. + +| Producer | Boundary | +| --- | --- | +| Heartbeat run lifecycle | Existing run-status transaction, scoped to company/run | +| Heartbeat queue, run dispatch, queued-comment and watchdog adapters | Existing terminal-write transaction, scoped to the affected run | +| Native finalizer, restart, finalization reconciliation, session failure and status-decision committer | Existing terminal projection transaction | +| Native cancellation acknowledgement | Existing audit/acknowledgement transaction before resultJson is updated | +| Native workspace reference | Existing locked reference transaction; includes preparation and finalization | +| Execution-control and task recovery | Existing eligible-run terminal-write transaction | +| Agent/company deletion | Company-scoped invalidation before cascades | +| Worktree execution settings | Global invalidation in the existing settings transaction | + +The status signals can cause a scoped inspection even if that run has no relevant +workspace. They add no recurring work. The existing transaction identity is +shared across topics; no provider/file call is wrapped in a new transaction. +Notifications and receipt interpretation stay with their respective owners. + +Coordinator ownership and lease proof changes do not need another broad polling +lane or new global signal: the run's explicit file debt is already registered +and retains its retry until it can be pinned. The retention recorder does not +signal its own partial failure back into immediate discovery, which would reset +the retry and recreate a tight loop. The due check observes its successful +receipt. Foreground completion may be noticed on that already-known retry. + +The supported writer model remains one process with instrumented writes. +Independent SQL writers and additional replicas require explicit integration. +Admitted execution binding is immutable; a different workspace contract belongs +to another admitted run. Do not infer coverage for arbitrary metadata rewrites. + +## Lifecycle, compatibility and index + +The heartbeat facade retains worktree-execution and task-drain admission. +Startup is registered only when heartbeat maintenance is enabled. The existing +startup/manual preservation entry point remains available; periodic reaping skips +only this migrated duty. Drain/standby retain changes without passive SQL. +Shutdown joins admitted owner transactions and removes the registration. + +Migration 0397 adds a partial index on (id, company_id) using the same coarse +pending predicate. It excludes the recorded completed-preservation history. +It is concurrent, replay-safe, registered in the existing nontransactional +migration runner, and accompanied by the five newest generated snapshots. + +## Verification + +The unchanged owner baseline passed 48 tests before implementation. The first +worker/startup selection passed 114 tests. Adding stale-acknowledgement, phase, +status and pagination cases raised that selection to 118 passing tests. Final +negative binding selection passed all 120 owner/startup tests (64 owner and 56 startup). Together with the 209 consumer cases, this gives 329 passing tests with no skips. + +Seven affected consumer suites passed 209 tests: Stop/run control, run dispatch, +queued comments, heartbeat queue/recovery, native finalizer and workspace sync. +These tests use real PostgreSQL where persistence matters and controlled runtime +or filesystem fixtures for external effects. Workspace typecheck and build +passed; final server types passed after the binding cases. + +Worker cases cover active/completed/workspace-less quiet, the real run-status +producer, outer rollback/commit, restart, drain/standby, another coordinator +owner, late exact stop-only proof, failed owner transactions, shutdown joining, +103 pending sources, stale candidate races and malformed bindings. Assertions +include lease policies, durable recovery receipts, recovery-card count, preserved +task/run state, no idle SQL and no remaining deadline. + +Fresh migration application, runner replay, direct SQL replay and migration +numbering/safety checks passed. A rolled-back query-plan fixture added 10,000 +completed rows plus one pending row. Discovery used the partial index, returned +one row with two shared buffer hits plus one local seed-table hit, and took +0.012 ms locally. This is index-selection evidence, not production latency. + +One test typecheck failed because an invokability mock widened literal true to +boolean. The mock now retains its literal type; the subsequent workspace check +passed. The first negative binding selection had 119 passing tests and one fixture failure: it attempted to clear a run issue binding while a coordinator still referenced it. PostgreSQL correctly rejected that state with 23503. The fixture now deletes its coordinator before clearing the binding, preserving the foreign key. The fresh final selection passed all 120 tests. Logs are retained at `/tmp/sleep43-owners-v4.log` and `/tmp/sleep43-owners-v5.log`. No failed selection was discarded. All test databases are disposable PostgreSQL 17 instances on this host; +this host's embedded harness was already proven unavailable in the parent slice. + +Exact-head remote CI and review remain required. This slice does not establish +Cloud alarm coverage, host/database suspension or fleet savings, and it does +not finish other native finalization/recovery duties. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index dc9dd75a9b6..bafd5212d92 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -522,6 +522,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0394_breezy_mongu.sql", "0395_colorful_darkstar.sql", "0396_late_cerise.sql", + "0397_lumpy_talos.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0397_lumpy_talos.sql b/packages/db/src/migrations/0397_lumpy_talos.sql new file mode 100644 index 00000000000..a010cb1a0bb --- /dev/null +++ b/packages/db/src/migrations/0397_lumpy_talos.sql @@ -0,0 +1,5 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_cancelled_workspace_preservation_idx" ON "heartbeat_runs" USING btree ("id","company_id") WHERE "heartbeat_runs"."runtime_mode" = 'native' AND "heartbeat_runs"."status" = 'cancelled' + AND "heartbeat_runs"."runner_profile_json"->'nativeWorkspaceSync'->>'state' = 'prepared' + AND "heartbeat_runs"."result_json"->'nativeCancellation'->>'dispatchState' = 'acknowledged' + AND ("heartbeat_runs"."result_json"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1' + AND "heartbeat_runs"."result_json"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0392_snapshot.json b/packages/db/src/migrations/meta/0397_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0392_snapshot.json rename to packages/db/src/migrations/meta/0397_snapshot.json index 8c7686e1e80..82e64ad9fb6 100644 --- a/packages/db/src/migrations/meta/0392_snapshot.json +++ b/packages/db/src/migrations/meta/0397_snapshot.json @@ -1,6 +1,6 @@ { - "id": "02276337-d910-42c8-a559-11facc2e8da6", - "prevId": "5f2e1d87-0147-430a-9bdf-a8df947725f1", + "id": "cdc2d88c-11cd-4c24-888d-50ec0c9f98cc", + "prevId": "27bec2e7-b637-4317-aa3c-a457bcc709fe", "version": "7", "dialect": "postgresql", "tables": { @@ -2637,6 +2637,50 @@ } }, "indexes": { + "agent_instruction_copies_collection_idx": { + "name": "agent_instruction_copies_collection_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "NOT (coalesce(\"agent_instruction_working_copies\".\"receipt\", '{}'::jsonb) ? 'retainedByRunId') AND (\n (\"agent_instruction_working_copies\".\"attempts\" < 3 AND (\"agent_instruction_working_copies\".\"state\" IN ('prepared', 'pending_collection', 'unchanged_turn', 'warm_saved')\n OR (\"agent_instruction_working_copies\".\"state\" = 'preparing' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1')))\n OR (\"agent_instruction_working_copies\".\"state\" = 'unavailable' AND \"agent_instruction_working_copies\".\"process_stopped_at\" IS NULL\n AND \"agent_instruction_working_copies\".\"location\" LIKE 'remote:%' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1'))", + "concurrently": false, + "method": "btree", + "with": {} + }, + "agent_instruction_copies_cleanup_idx": { + "name": "agent_instruction_copies_cleanup_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"agent_instruction_working_copies\".\"state\" IN ('saved', 'unchanged', 'resolved', 'unavailable')\n AND \"agent_instruction_working_copies\".\"process_stopped_at\" IS NOT NULL\n AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1'\n AND coalesce(\"agent_instruction_working_copies\".\"receipt\"->>'retainedByRunId', '') = ''\n AND \"agent_instruction_working_copies\".\"receipt\"->>'cleanupPending' IS DISTINCT FROM 'false'\n AND (\"agent_instruction_working_copies\".\"receipt\" ? 'baseline' OR \"agent_instruction_working_copies\".\"receipt\"->>'cleanupPending' = 'true')", + "concurrently": false, + "method": "btree", + "with": {} + }, "agent_instruction_copies_pending_idx": { "name": "agent_instruction_copies_pending_idx", "columns": [ @@ -9617,6 +9661,34 @@ "method": "btree", "with": {} }, + "chat_actions_github_stale_head_idx": { + "name": "chat_actions_github_stale_head_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"payload\"->'event'->>'deliveryId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_review_check'\n and \"chat_actions\".\"status\" = 'cancelled' and \"chat_actions\".\"result\"->>'code' = 'stale_head'", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_actions_github_check_work_idx": { "name": "chat_actions_github_check_work_idx", "columns": [ @@ -13137,6 +13209,22 @@ } }, "indexes": { + "chat_github_reviews_projection_work_idx": { + "name": "chat_github_reviews_projection_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_github_reviews\".\"state\" in ('queued', 'running')\n and \"chat_github_reviews\".\"assessment\" is null\n and \"chat_github_reviews\".\"delivery_id\" ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_github_reviews_delivery_uq": { "name": "chat_github_reviews_delivery_uq", "columns": [ @@ -33583,6 +33671,28 @@ } }, "indexes": { + "heartbeat_runs_cancelled_workspace_preservation_idx": { + "name": "heartbeat_runs_cancelled_workspace_preservation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" = 'cancelled'\n AND \"heartbeat_runs\".\"runner_profile_json\"->'nativeWorkspaceSync'->>'state' = 'prepared'\n AND \"heartbeat_runs\".\"result_json\"->'nativeCancellation'->>'dispatchState' = 'acknowledged'\n AND (\"heartbeat_runs\".\"result_json\"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'\n AND \"heartbeat_runs\".\"result_json\"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_retained_recovery_adoption_idx": { "name": "heartbeat_runs_retained_recovery_adoption_idx", "columns": [ @@ -33599,6 +33709,22 @@ "method": "btree", "with": {} }, + "heartbeat_runs_scheduled_retry_cancellation_idx": { + "name": "heartbeat_runs_scheduled_retry_cancellation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"status\" = 'scheduled_retry' and jsonb_typeof(\"heartbeat_runs\".\"result_json\"->'startupCancellation') = 'object'", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 89e39f4cf72..98e028905fa 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2766,6 +2766,13 @@ "when": 1792976876772, "tag": "0396_late_cerise", "breakpoints": true + }, + { + "idx": 397, + "version": "7", + "when": 1792978930564, + "tag": "0397_lumpy_talos", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 287ca78b7fb..99f06a955dc 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -111,6 +111,12 @@ export const heartbeatRuns = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + cancelledWorkspacePreservationIdx: index("heartbeat_runs_cancelled_workspace_preservation_idx").on(table.id, table.companyId) + .where(sql`${table.runtimeMode} = 'native' AND ${table.status} = 'cancelled' + AND ${table.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared' + AND ${table.resultJson}->'nativeCancellation'->>'dispatchState' = 'acknowledged' + AND (${table.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1' + AND ${table.resultJson}->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE`), retainedRecoveryAdoptionIdx: index("heartbeat_runs_retained_recovery_adoption_idx") .on(table.id).where(sql`${table.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1' and not (coalesce(${table.resultJson}, '{}'::jsonb) ? 'systemRecoveryAdoption') diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 688dc59b8d4..5d8d9a2477f 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -108,6 +108,7 @@ const { reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerCancelledWorkspacePreservationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerInstructionDirectoryCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerCapturedInstructionCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })), @@ -542,6 +543,13 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers cancelled workspace preservation only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerCancelledWorkspacePreservationWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerCancelledWorkspacePreservationWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers stopped instruction collection only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -920,7 +928,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 }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 64b49e464ad..04ef2379325 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1777,6 +1777,8 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + if (config.heartbeatSchedulerEnabled) await heartbeat.registerCancelledWorkspacePreservationWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerStoppedInstructionCollectionWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; @@ -1907,7 +1909,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. trackHeartbeatSchedulerWork(heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 26c2a9cd517..2d3d0b2f831 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; import { and, asc, desc, eq, gt, gte, inArray, ne, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -339,6 +340,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch const transactionResult = await db.transaction(async (tx) => { await notifyInstructionCollection(tx, { companyId, runId: input.run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id }); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index aef69b49971..a0b59f04419 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js"; import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; @@ -669,6 +670,7 @@ export function createPostgresRunDispatchAdapter( | { applied: false } > { await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); + await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId }); await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: issueId ?? undefined }); const [row] = await tx @@ -713,6 +715,7 @@ export function createPostgresRunDispatchAdapter( input: CancelSuppressedRetryInput, ): Promise { await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); + await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId }); await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); const [row] = await tx @@ -921,6 +924,7 @@ export function createPostgresRunDispatchAdapter( providerDispatchNotStarted = false, ): Promise { await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [row] = await tx diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index 01f0ef27acb..5dcd57c3485 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; @@ -105,6 +106,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu async cancelQueueRun({ queueRunId, reason, now }) { await notifyInstructionCollection(tx, { companyId, runId: queueRunId }); + await notifyCancelledWorkspacePreservation(tx, { companyId, runId: queueRunId }); await notifyChatRunMilestones(tx, { companyId, issueId }); const row = await tx .update(heartbeatRuns) diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index 0ed1a03ae40..1f207367051 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; @@ -50,6 +51,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str await notifyInstructionCommit(tx, { companyId: companyId }); await notifyDirectoryCleanup(tx, { companyId: companyId }); await notifyInstructionCollection(tx, { companyId: companyId }); + await notifyCancelledWorkspacePreservation(tx, { companyId: companyId }); await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, id)); await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, id)); await tx.delete(agentApiKeys).where(eq(agentApiKeys.agentId, id)); diff --git a/server/src/services/cancelled-workspace-preservation-notifications.ts b/server/src/services/cancelled-workspace-preservation-notifications.ts new file mode 100644 index 00000000000..9294c30d7b9 --- /dev/null +++ b/server/src/services/cancelled-workspace-preservation-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type CancelledWorkspacePreservationScope = { companyId: string; runId?: string }; + +export async function notifyCancelledWorkspacePreservation(transaction: object, scope?: CancelledWorkspacePreservationScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.cancelledWorkspacePreservation, scope ? JSON.stringify(scope) : "*"); +} diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index d37910772d4..8d41410a4ef 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; @@ -102,6 +103,7 @@ export async function deleteCompany(db: Db, id: string) { await notifyInstructionCommit(tx, { companyId: id }); await notifyDirectoryCleanup(tx, { companyId: id }); await notifyInstructionCollection(tx, { companyId: id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: id }); await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, id)); await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, id)); await tx.delete(agentApiKeys).where(eq(agentApiKeys.companyId, id)); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index ba3eade9d8d..fe48aec9b10 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + cancelledWorkspacePreservation: "cancelled-workspace-preservation", instructionCollection: "instruction-collection", directoryCleanup: "directory-cleanup", instructionCommit: "instruction-commit", diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 1d61ae329b1..ce5441fd687 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { randomUUID } from "node:crypto"; @@ -184,6 +185,7 @@ export async function reconcileAbandonedExecutionControl( }) .where(eq(nativeRunFinalizations.runId, run.id)); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyExecutionRunWork(tx); const [updatedRun] = await tx .update(heartbeatRuns) diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 335e547604e..a0fb2885e6f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { writeExecutionControl } from "./execution-control-notifications.js"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; @@ -4062,6 +4063,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerCancelledWorkspacePreservationWork: ( + coordinator: Parameters[1], canRun: () => boolean, + ) => registerCancelledWorkspacePreservationWork(db, coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerStoppedInstructionCollectionWork: ( coordinator: Parameters[0], canRun: () => boolean, ) => instructionCopies.registerStoppedCollectionWork(coordinator, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 53a37472c5d..fd8db6d1c91 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatResponseAvailability } from "../chat-response-availability.js"; @@ -850,6 +851,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende : "Automatic continuation source could not be verified before provider admission. Review the task and send a fresh request; this attempt will not automatically retry."; await notifyChatRunMilestones(tx, { companyId: current.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: current.companyId, runId: current.id }); [terminal] = await tx .update(heartbeatRuns) .set({ @@ -972,6 +974,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (!currentRun) return null; await notifyChatRunMilestones(tx, { companyId: run.companyId }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); const [row] = await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), errorCode: "case_handoff_superseded", error: "This case handoff was cancelled or superseded.", updatedAt: new Date() }) .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "queued"))).returning(); @@ -1333,6 +1336,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende "Automatic continuation stopped by the committed chat conversation close. Send a fresh request in chat or on the Board."; await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); const [cancelled] = await tx .update(heartbeatRuns) .set({ @@ -1455,6 +1459,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const reason = "Queued messages were discarded before dispatch"; await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); const [cancelled] = await tx .update(heartbeatRuns) .set({ @@ -3447,6 +3452,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ? "Cancelled because the issue was cancelled before the scheduled retry became due" : "Cancelled because the issue was reassigned before the scheduled retry became due"; await notifyInstructionCollection(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); await notifyScheduledRetryCancellation(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); await notifyChatRunMilestones(tx, { companyId: scheduledRun.companyId, issueId: issue.id }); const cancelled = await tx @@ -3586,6 +3592,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende activeExecutionRun.agentId !== issue.assigneeAgentId ) { await notifyInstructionCollection(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id }); await notifyChatRunMilestones(tx, { companyId: activeExecutionRun.companyId, issueId: issue.id }); const cancelled = await tx .update(heartbeatRuns) diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 7e1f5b8e133..37036be05ec 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1915,8 +1915,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean }) { - await preserveCancelledNativeWorkspaces(db); + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean }) { + if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); if (!opts?.skipScheduledRetryCancellation) await scheduledRetryCancellation.sweep(); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index fbab2bb7d23..7d1b48991ce 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; @@ -1024,6 +1025,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl if (previous && (previous.status === "scheduled_retry" || status === "scheduled_retry")) await notifyScheduledRetryCancellation(tx, { companyId: previous.companyId, runId: previous.id }); if (previous) await notifyInstructionCollection(tx, { companyId: previous.companyId, runId: previous.id }); + if (previous) await notifyCancelledWorkspacePreservation(tx, { companyId: previous.companyId, runId: previous.id }); if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId); await notifyExecutionRunWork(tx); if (companyId) await notifyChatRunMilestones(tx, { companyId }); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 2d54950c51b..46f531d4712 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; @@ -624,6 +625,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyInstructionCommit(tx); await notifyDirectoryCleanup(tx); await notifyInstructionCollection(tx); + await notifyCancelledWorkspacePreservation(tx); } if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx); await notifyToolConnectionHealth(tx, "*"); diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts index 22570efdb0f..f43616e099f 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts @@ -1,17 +1,23 @@ +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../task-admission.js"; +import { createHeartbeatLifecycle } from "../heartbeat/run-lifecycle.js"; +import { createHeartbeatRunState } from "../heartbeat/run-state.js"; +import { createHeartbeatRunPreparation } from "../heartbeat/run-preparation.js"; import { randomUUID } from "node:crypto"; -import { and, eq } from "drizzle-orm"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { and, eq, inArray } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { activityLog, agents, companies, createDb, environmentLeases, environments, externalAgentHolds, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { getExecutionBlocker } from "../execution-blocker.js"; import { markExecutionReconciliation, validateExecutionReconciliation } from "../execution-recovery-resolution.js"; import { prepareSandboxStopAndRetain, settleStopOnlyCleanup } from "../sandbox-stop-and-retain.js"; -import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js"; +import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, registerCancelledWorkspacePreservationWork } from "./native-failed-workspace-recovery.js"; import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; -const externalUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL"); const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("failed native workspace retention", () => { let db: ReturnType; @@ -51,6 +57,245 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes const readLease = async (id: string) => (await db.select().from(environmentLeases).where(eq(environmentLeases.id, id)))[0]!; const actions = async (issueId: string) => db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)); + describe("cancelled workspace preservation worker", () => { + const fixtureCompanies = new Set(); + afterEach(async () => { + vi.useRealTimers(); vi.restoreAllMocks(); + if (fixtureCompanies.size) await db.update(heartbeatRuns).set({ status: "failed" }) + .where(inArray(heartbeatRuns.companyId, [...fixtureCompanies])); + fixtureCompanies.clear(); + }); + async function prepared() { + const fixture = await seed(); fixtureCompanies.add(fixture.companyId); return fixture; + } + const acknowledgement = (f: Awaited>) => ({ + cancelledByActorType: "user", cancelledByUserId: "operator", nativeCancellation: { + schema: "paperclip.native-cancellation.v1", runId: f.runId, companyId: f.companyId, issueId: f.issueId, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID(), + }, + }); + const clock = () => vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const coordinator = (canRun = () => true, canReconcile = () => true) => + createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile, onError: () => {} }); + const lifecycle = (owner = db) => createHeartbeatLifecycle(owner, { + ...createHeartbeatRunState(owner), ...createHeartbeatRunPreparation(owner), + issuesSvc: { addComment: vi.fn(), listReviewAttention: vi.fn() }, + recovery: { escalateStrandedAssignedIssue: vi.fn() }, + companySkills: { completeTestRunForIssue: vi.fn() }, + budgets: { getInvocationBlock: vi.fn(async () => null) }, + treeControlSvc: { getActivePauseHoldGate: vi.fn(async () => null) }, + enqueueWakeup: vi.fn(), getCurrentUserRedactionOptions: vi.fn(async () => ({})), + getAgentInvokability: vi.fn(async () => ({ invokable: true as const })), + emitTerminalAgentTaskRun: vi.fn(), budgetHooks: {}, + }); + async function cancel(f: Awaited>, owner = db) { + await lifecycle(owner).setRunStatus(f.runId, "cancelled", { resultJson: acknowledgement(f), finishedAt: new Date() }); + } + async function pinned(f: Awaited>) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(100); + expect((await readRun(f.runId)).resultJson).toMatchObject({ + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }, + }); + }, { timeout: 5000 }); + expect((await readLease(f.lease.id)).leasePolicy).toBe("retain_on_failure"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_progress"); + expect((await readRun(f.runId)).status).toBe("cancelled"); + expect(await actions(f.issueId)).toHaveLength(1); + } + it("stays quiet for active runs, completed preservation, and admitted workspace-less sessions", async () => { + const active = await prepared(), settled = await prepared(), external = await prepared(); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, active.runId)); + await cancel(settled); await preserveNativeFailedWorkspace(db, await readRun(settled.runId), { allowPreterminal: true }); + await bindExternalInput(external, "muse"); await cancel(external); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + expect((await readLease(active.lease.id)).status).toBe("active"); + expect((await readLease(external.lease.id)).status).toBe("active"); + } finally { await worker.stop(); } + }); + it.each(["missing issue", "foreign company"] as const)("does not schedule an unbound cancellation receipt: %s", async mismatch => { + const f = await prepared(); const result = acknowledgement(f); + if (mismatch === "missing issue") await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ status: "cancelled", + ...(mismatch === "missing issue" ? { nativeIssueId: null } : {}), + resultJson: { ...result, nativeCancellation: { ...result.nativeCancellation, + ...(mismatch === "missing issue" ? { issueId: null } : { companyId: randomUUID() }), + } }, + }).where(eq(heartbeatRuns.id, f.runId)); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + expect((await readLease(f.lease.id)).status).toBe("active"); + } finally { await worker.stop(); } + }); + it("reactivates from the real run-status producer and becomes quiet after recording the hold", async () => { + const f = await prepared(); clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect(worker.nextWakeAt()).toBeNull(); + await cancel(f); await pinned(f); + await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }); + it.each(["restart", "drain", "standby"] as const)("retains preservation across %s", async mode => { + const f = await prepared(); if (mode === "restart") await cancel(f); + let admitted = mode === "restart"; + clock(); const worker = coordinator(() => admitted, () => admitted || mode !== "standby"); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => admitted, async () => true).ready; + if (mode !== "restart") { + await cancel(f); await vi.advanceTimersByTimeAsync(1000); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect((await readLease(f.lease.id)).status).toBe("active"); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); + } + await pinned(f); + } finally { await worker.stop(); } + }); + it("waits for the outer cancellation commit and discards a rolled-back transition", async () => { + const f = await prepared(); clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + await expect(db.transaction(async tx => { + await cancel(f, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(1000); + expect((await readLease(f.lease.id)).status).toBe("active"); + throw new Error("rollback cancellation"); + })).rejects.toThrow("rollback cancellation"); + await vi.advanceTimersByTimeAsync(1000); + expect((await readRun(f.runId)).status).toBe("failed"); + await db.transaction(async tx => { + await cancel(f, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(1000); + expect((await readLease(f.lease.id)).status).toBe("active"); + }); + await pinned(f); + } finally { await worker.stop(); } + }); + it("preserves owed work while another controller owns finalization", async () => { + const f = await prepared(); await cancel(f); + await db.update(nativeRunFinalizations).set({ leaseOwner: "other-controller" }).where(eq(nativeRunFinalizations.runId, f.runId)); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect((await readLease(f.lease.id)).status).toBe("active"); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(20_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await db.update(nativeRunFinalizations).set({ leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await vi.advanceTimersByTimeAsync(10_000); await pinned(f); + } finally { await worker.stop(); } + }); + it("waits for exact stop-only proof before retaining an already-released source", async () => { + const f = await prepared(); await cancel(f); + const stopping = await prepareSandboxStopAndRetain(db, f.lease); + if (!stopping) throw new Error("Expected stop-only source"); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect((await readRun(f.runId)).resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(20_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await settleStopOnlyCleanup(db, stopping, { attemptId: String(stopping.metadata?.pendingCleanupAttemptId), + receipt: { providerLeaseId: f.lease.providerLeaseId, state: "stopped" } }); + await vi.advanceTimersByTimeAsync(10_000); await pinned(f); + expect((await readLease(f.lease.id)).status).toBe("released"); + } finally { await worker.stop(); } + }); + it("retries failed owner transactions without blocking another cancelled source", async () => { + const first = await prepared(), second = await prepared(); await cancel(first); await cancel(second); + clock(); const worker = coordinator(); + const transactions = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("preservation database unavailable")); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + transactions.mockRestore(); + const policies = [(await readLease(first.lease.id)).leasePolicy, (await readLease(second.lease.id)).leasePolicy]; + expect(policies.filter(policy => policy === "retain_on_failure")).toHaveLength(1); + await vi.advanceTimersByTimeAsync(30_000); await pinned(first); await pinned(second); + } finally { transactions.mockRestore(); await worker.stop(); } + }); + it.each(["acknowledgement", "phase", "status"] as const)("rejects changed %s at the locked preservation boundary", async changed => { + const f = await prepared(); await cancel(f); clock(); const worker = coordinator(); + const transaction = db.transaction.bind(db); let changedOnce = false; + const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => { + if (!changedOnce) { + changedOnce = true; + if (changed === "phase") await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + else if (changed === "status") await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.runId)); + else { + const result = acknowledgement(f); + result.nativeCancellation.dispatchState = "pending"; + await db.update(heartbeatRuns).set({ resultJson: result }).where(eq(heartbeatRuns.id, f.runId)); + } + } + return transaction(callback, options); + }) as typeof db.transaction); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + transactions.mockRestore(); expect(changedOnce).toBe(true); + expect((await readLease(f.lease.id)).status).toBe("active"); + expect((await readRun(f.runId)).resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + if (changed === "phase") { + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("observed"); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await vi.advanceTimersByTimeAsync(30_000); + } else await cancel(f); + await pinned(f); + } finally { transactions.mockRestore(); await worker.stop(); } + }); + it("reconstructs more than one page and pins all sources in bounded batches", async () => { + const fixtures = []; + for (let index = 0; index < 103; index++) { + const f = await prepared(); await cancel(f); fixtures.push(f); + } + clock(); const worker = coordinator(); + const retained = () => db.select({ id: environmentLeases.id }).from(environmentLeases) + .where(and(inArray(environmentLeases.companyId, [...fixtureCompanies]), eq(environmentLeases.leasePolicy, "retain_on_failure"))); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect((await retained()).length).toBeLessThanOrEqual(5); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await retained()).toHaveLength(103); + }, { timeout: 20_000 }); + for (const f of [fixtures[0], fixtures[102]]) await pinned(f); + await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }, 60_000); + it("joins an admitted preservation transaction during shutdown", async () => { + const f = await prepared(); await cancel(f); clock(); const worker = coordinator(); + const transaction = db.transaction.bind(db); + let entered!: () => void, release!: () => void; + const reached = new Promise(resolve => { entered = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const transactions = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + entered(); await held; return transaction(callback, options); + }) as typeof db.transaction); + let stopping: Promise | undefined; + try { + const registration = registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true); + await reached; let complete = false; + stopping = worker.stop().then(() => { complete = true; }); + await new Promise(resolve => setImmediate(resolve)); expect(complete).toBe(false); + release(); await stopping; await registration.ready; await pinned(f); + const restarted = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, restarted, () => true, async () => true).ready; + expect(restarted.nextWakeAt()).toBeNull(); + } finally { await restarted.stop(); } + } finally { release(); transactions.mockRestore(); await stopping; await worker.stop(); } + }); + }); + it.each(["acknowledged", "warm", "pending", "foreign", "owned"])("pins historical cancelled work only after an exact Stop acknowledgement: %s", async kind => { const f = await seed(); if (kind === "warm") await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment" }).where(eq(environmentLeases.id, f.lease.id)); diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index c8f31024920..1e9d1721b9f 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -1,3 +1,7 @@ +import type { CancelledWorkspacePreservationScope } from "../cancelled-workspace-preservation-notifications.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; @@ -10,7 +14,7 @@ import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finaliz import { hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; import { logger } from "../../middleware/logger.js"; -export async function preserveNativeFailedWorkspace(db: Db, selected: typeof heartbeatRuns.$inferSelect, options: { allowPreterminal?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) { +export async function preserveNativeFailedWorkspace(db: Db, selected: typeof heartbeatRuns.$inferSelect, options: { allowPreterminal?: boolean; requireCancelledStop?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) { if (readNativeWorkspaceSyncReference(selected.runnerProfileJson?.nativeWorkspaceSync)?.state !== "prepared") return; const issueId = selected.nativeIssueId ?? selected.issueId ?? selected.contextSnapshot?.issueId; if (typeof issueId !== "string") return; @@ -24,6 +28,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea eq(nativeRunFinalizations.runId, selected.id), eq(nativeRunFinalizations.companyId, selected.companyId), eq(nativeRunFinalizations.issueId, issueId), )).for("update"); + if (options.requireCancelledStop && coordinator?.phase !== "terminal_failure") return; const claim = options.restartRecovery; const reference = readNativeWorkspaceSyncReference(selected.runnerProfileJson?.nativeWorkspaceSync)!; const ownsClaim = Boolean(options.allowPreterminal && coordinator?.phase === "observed" && claim @@ -39,6 +44,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea eq(heartbeatRuns.id, selected.id), eq(heartbeatRuns.companyId, selected.companyId), )).for("update"); if (!run || run.runtimeMode !== "native" || (!options.allowPreterminal && run.status !== "failed")) return; + if (options.requireCancelledStop && (run.status !== "cancelled" || !hasAcknowledgedNativeStopIntent(run))) return; const currentReference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); if (!currentReference || currentReference.state !== "prepared" || (claim?.kind === "reattach_remote_runner" && (claim.remote.providerLeaseId !== currentReference.providerLeaseId @@ -110,7 +116,7 @@ export async function preserveCancelledNativeWorkspaces(db: Db) { cancelledWorkspaceCursors.set(db, candidate.runId); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, candidate.runId)); if (run && hasAcknowledgedNativeStopIntent(run)) { - try { await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true }); } + try { await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); } catch (error) { if (error instanceof NativeWorkspaceFinalizationBusyError) continue; let current: unknown = error; @@ -126,3 +132,70 @@ export async function preserveCancelledNativeWorkspaces(db: Db) { } } } + +function pendingCancelledWorkspace() { + return sql`${heartbeatRuns.runtimeMode} = 'native' AND ${heartbeatRuns.status} = 'cancelled' + AND ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared' + AND ${heartbeatRuns.resultJson}->'nativeCancellation'->>'dispatchState' = 'acknowledged' + AND (${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1' + AND ${heartbeatRuns.resultJson}->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE`; +} + +export function registerCancelledWorkspacePreservationWork(db: Db, + coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise) { + type Entry = { id: string; companyId: string; dueAt: number }; + type Candidate = Pick; + const eligible = (run: Candidate) => run.nativeIssueId !== null && hasAcknowledgedNativeStopIntent(run) + && readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync)?.state === "prepared" + && !hasRequiredWorkspaceRecovery(run.resultJson) && !nativeRunHasNoWorkspaceAccess(run); + const scopeCondition = (scope?: CancelledWorkspacePreservationScope) => scope ? and( + eq(heartbeatRuns.companyId, scope.companyId), scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined) : undefined; + async function load(scope: CancelledWorkspacePreservationScope | undefined, signal?: AbortSignal) { + if (!await canRecover()) return []; + const entries: Entry[] = []; + let cursor: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, + driverKind: heartbeatRuns.driverKind, status: heartbeatRuns.status, resultJson: heartbeatRuns.resultJson, + runnerProfileJson: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and(pendingCancelledWorkspace(), scopeCondition(scope), + cursor ? gt(heartbeatRuns.id, cursor) : undefined)).orderBy(asc(heartbeatRuns.id)).limit(100); + for (const row of rows) if (eligible(row)) entries.push({ id: row.id, companyId: row.companyId, dueAt: 0 }); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + throw new Error("Cancelled workspace preservation reconstruction stopped"); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + if (!await canRecover()) return null; + const [run] = await db.select().from(heartbeatRuns) + .where(and(pendingCancelledWorkspace(), scopeCondition({ companyId: entry.companyId, runId: entry.id }))); + if (!run || !eligible(run)) return null; + if (!canRun() || signal?.aborted) throw new Error("Cancelled workspace preservation stopped"); + const [finalization] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.companyId, run.companyId), + eq(nativeRunFinalizations.issueId, run.nativeIssueId!), + )); + if (finalization?.resultId) return null; + if (finalization?.phase !== "terminal_failure" || finalization.leaseOwner) + return { ...entry, dueAt: Date.now() + 30_000 }; + const [source] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + or(eq(environmentLeases.status, "active"), and(inArray(environmentLeases.status, ["pending_cleanup", "released"]), + eq(environmentLeases.leasePolicy, "ephemeral"))), + )).limit(1); + if (source) await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); + const [current] = await db.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))); + return current && current.status === "cancelled" && eligible(current) ? { ...entry, dueAt: Date.now() + 30_000 } : null; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.cancelledWorkspacePreservation, batchSize: 5, + decodeScope: (hint: string): CancelledWorkspacePreservationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: CancelledWorkspacePreservationScope) => !scope || + entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId), + load, check, + }, canRun); +} diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index c63ff046a86..2c022c527fe 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import type { WorkspaceWorkers } from "../plugin-workspaces.js"; import { recoverLegacyUnsafeWorkspaceExports } from "./native-workspace-export-recovery.js"; @@ -511,6 +512,7 @@ export async function claimNativeSessionResumptions(input: { updatedAt: now, }).where(eq(nativeRunFinalizations.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 }); const [updatedRun] = await tx.update(heartbeatRuns).set({ status: "failed", finishedAt: now, diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 07c81beb603..cf5f5767369 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 { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; @@ -891,6 +892,7 @@ export async function claimNativeRestartRecoveries(input: { ), ); await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id }); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 955dfd779a8..8769df752eb 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 { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { pendingNativeGovernance } from "./native-governance.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; @@ -410,6 +411,7 @@ async function recordRetryableFailure(input: { await preserveNativeWorkspaceExportLease(tx as unknown as Db, input.run, input.coordinator.resultId); } if (projectsTerminalStatus) await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id }); + if (projectsTerminalStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id }); const [updatedRun] = await tx .update(heartbeatRuns) .set({ @@ -668,6 +670,7 @@ async function projectCommittedRun(input: { const [updatedRun] = await input.db.transaction(async transaction => { const tx = transaction as unknown as Db; await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); await notifyExecutionRunWork(tx); await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); @@ -1605,6 +1608,7 @@ export async function finalizeNativeRun(input: { const [updatedRun] = await input.db.transaction(async transaction => { const tx = transaction as unknown as Db; if (input.projectRunStatus) await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); await notifyExecutionRunWork(tx); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 1a1ed567e9d..4c413466224 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; @@ -7412,6 +7413,7 @@ export async function cancelNativeSession( const acknowledgementAuditId = activity.activity?.id ?? null; if (!acknowledgementAuditId) throw new Error("native_cancellation_ack_audit_missing"); + await notifyCancelledWorkspacePreservation(tx, { companyId: cancellationContext.companyId, runId }); const cancellationWrite = await tx .update(heartbeatRuns) .set({ @@ -9596,6 +9598,7 @@ async function executePaperclipNativeSessionWithinScope( .where(eq(heartbeatRuns.id, input.execution.binding.runId)) .for("update"); if (!ownershipUnverified) await notifyInstructionCollection(tx, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId }); + if (!ownershipUnverified) await notifyCancelledWorkspacePreservation(tx, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId }); let [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index bcc3573e4e3..dc8499c6483 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { emitStartupReuse, type StartupReuseReason } from "@paperclipai/adapter-utils/startup-reuse"; @@ -614,13 +615,14 @@ async function persistRunReference( ): Promise { await db.transaction(async (tx) => { const run = await tx - .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) + .select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)) .for("update") .limit(1) .then((rows) => rows[0] ?? null); if (!run) throw new Error("native_workspace_sync_run_missing"); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId }); await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 26a54b0e4f4..900087cac69 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; @@ -1065,6 +1066,7 @@ async function materializeDecisionEffect(input: { ); } await notifyInstructionCollection(input.tx, { companyId: input.companyId, runId: input.runId }); + await notifyCancelledWorkspacePreservation(input.tx, { companyId: input.companyId, runId: input.runId }); await notifyDeliveryWork(input.tx, DELIVERY_QUEUES.accounting, input.companyId); const [run] = await input.tx .update(heartbeatRuns) diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index c4f93391546..e560ece9edb 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { boundRunIssueId } from "../run-task-binding.js"; @@ -6214,6 +6215,7 @@ export function recoveryService( const [candidate] = await tx.select().from(heartbeatRuns).where(eligible).for("update"); if (!candidate) return null; await notifyInstructionCollection(tx, { companyId: candidate.companyId, runId: candidate.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: candidate.companyId, runId: candidate.id }); const boundIssueId = boundRunIssueId(candidate); if (boundIssueId) { await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: boundIssueId });