fix(server): schedule cancelled workspace preservation (#17677)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - A cancelled native run may still owe preservation of prepared
workspace files.
> - The existing recovery receipt and source lease record that
obligation.
> - The orphan reaper still searches for these cancellations on every
pass.
> - This PR registers committed cancellations with the shared deadline
worker.
> - The existing retention owner keeps file pinning and ownership
checks.
> - Idle and completed history no longer causes this preservation scan.

## Linked Issues or Issue Description

Refs #17674 and #17145. Stopped instruction collection is merged into
master. It owns file-preservation discovery only. The existing
environment cleanup worker continues to own stop-only teardown.

**What needs to be improved?**

`preserveCancelledNativeWorkspaces` runs from every orphan-reaper pass.
It queries even when no cancelled native workspace needs a file-recovery
hold. Current foreground Stop can also defer retention after a lock or
database failure, so startup-only recovery would be insufficient.

**Why is this improvement important?**

Empty queries keep idle tenants active. Removing them must not lose user
files, weaken Stop ownership, replay cancelled work, or turn a retained
allocation into destructive cleanup.

**Proposed solution**

Register the existing durable obligation from committed run-status,
cancellation-acknowledgement and workspace-reference writes. Reconstruct
it at startup. Keep retries while actual preservation remains
unresolved. Disarm after the existing owner records the hold. Related-PR
search and ROADMAP.md review found no duplicate of this scheduling
migration.

## What Changed

- Add a company/run-scoped preservation worker with 100-row keyset
reconstruction and fair batches of five.
- Separate pending file debt from immediate coordinator/lease readiness.
Retain a 30-second retry for known unfinished preservation and shared
failure backoff for errors.
- Signal the existing status, acknowledgement, workspace-reference,
deletion and settings transactions. Add no table or independent timer.
- Keep exact source checks, no-workspace contracts, task/coordinator/run
lock order, retention receipt and no-replay recovery card with their
owners.
- Recheck cancellation acknowledgement, run status and coordinator phase
under the retention locks. Preserve the existing foreground preterminal
API.
- Keep startup/manual recovery; skip only this duty in periodic reaping.
Preserve drain/standby and shutdown behavior.
- Add concurrent, replay-safe partial index migration 0409 and retain
the five newest snapshots.
- Add the producer and lifecycle audit in
`doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md`.

## Verification

- Current source `21b46a8e4293ca670235117458b984493a181759`: all 1472
cases in 16 complete files pass without skips. The owner passes 84 cases
in 14.93 seconds. Fifteen consumer files pass 1388 cases in 133.46
seconds, including all 968 native-session cases and Stop, dispatch,
restart, workspace-sync and upgrade-readiness coverage.
- Twelve separate fresh PostgreSQL 17 databases migrated through 0409
before setup. The embedded-only lifecycle fixture temporarily used its
own external database on this IPC-exhausted host and was restored byte
for byte. Assertions and timing limits stayed unchanged. This qualifies
external-database behavior; it does not claim an unmodified embedded CI
run.
- Database build, server typecheck/build, module boundaries and
exact-head migration order pass. The complete original runtime/schema
patch and unrelated current-parent blobs are preserved. Original
concurrent index SQL, all earlier executable history and retained
snapshot bytes are unchanged; five snapshots 0405 through 0409 remain.
- A populated 0408-to0409 upgrade preserves all 407 prior receipts and
38 companies. Exactly one receipt is appended; replay adds none. The
preservation index is valid and ready on fresh and upgraded databases.
- Actual parent `82529741418e138d9d623bf13d2fa22c1ff79d36` is merged
through a normal ancestry reconciliation. Its complete tree equals the
reviewed parent. The entire accepted child tree and all source hashes
are preserved. The audit retains earlier failures, qualifications and
per-file counts. Published head
`b34f0589847f33754ecde9413fd36bf007353d2c` has Greptile 5/5 and all 57
checks passing or skipped. No Ponytail review is present. These checks
do not establish deployed host/database suspension or Cloud alarm
delivery.

## Risks

- Signals cover instrumented single-process writers. Independent SQL
writers and additional replicas need integration.
- Status signals may inspect a run without relevant file debt once. They
do not create recurring work. Discovery and memory still scale with
actual candidates.
- Missing ownership or stop evidence remains pending. Active retries are
intentional while a concrete file obligation is unresolved.
- Retry timing is process-local; startup reconstructs durable debt. The
worker does not overwrite another native-finalizer deadline.
- Admitted database operations may finish after drain begins. Shutdown
joins them. Provider teardown remains with the existing cleanup owner.
- The partial index adds storage and write cost. Its concurrent
migration uses the existing interrupted-index repair path.
- Cloud alarm coverage, sleep eligibility and host suspension are
unchanged. This proves worker behavior, not fleet savings.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning, shell tools, and PostgreSQL
tests. The runtime does not expose a more precise model ID or
context-window size for this turn.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -2601,3 +2601,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.

doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md

@@ -0,0 +1,260 @@
+# Cancelled native workspace preservation
+
+Date: 2026-10-25; current integration: 2026-10-29. Parent: stopped instruction
+collection PR #17674 at `d176d731f083c7edd2ede688cb901d1e66d8f1d8`.
+The original draft parent was `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.
+
+Current migration 0409 adds a partial index (the original draft used 0397) 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.
+
+
+## Current parent integration and acceptance (2026-10-29)
+
+The existing slice adopts the updated stopped-collection parent by normal merge.
+The initial automatic proof rejected three source overlaps, rather than choosing
+a side. The resolved heartbeat facade retains both upgrade-readiness and worker
+registration imports. The recovery owner retains the upstream narrow projected
+input and plugin-task fence, with this slice's `requireCancelledStop` option and
+locked-row acknowledgement check. Both the exact original worker test block and
+the exact upstream superseded-retry test block are retained. Unrelated upstream
+blobs are preserved. A cached-diff whitespace warning belonged to an unchanged
+upstream routine-status file; the owned diff against the current parent is clean.
+
+Earlier source `8deb3941850e4c07744caceaccf30ad0f53abefd` passes **1472 cases across 16 complete files,
+with no skips**. The owner passes 84 cases, including the preserved superseded-
+retry coverage. Fifteen consumer files pass 1388 cases, including the complete
+968-case native session executor. Per-file counts follow:
+
+| Complete file | Passing cases |
+| --- | ---: |
+| src/services/native-runtime/native-failed-workspace-recovery.test.ts | 84 |
+| src/services/native-runtime/native-session-executor.test.ts | 968 |
+| src/services/heartbeat/run-control.test.ts | 51 |
+| src/services/heartbeat/run-lifecycle.test.ts | 31 |
+| src/services/heartbeat/queue.test.ts | 32 |
+| src/modules/run-dispatch/adapters/postgres.test.ts | 69 |
+| src/services/heartbeat/recovery.test.ts | 45 |
+| src/services/native-runtime/native-run-finalizer-telemetry.test.ts | 29 |
+| src/__tests__/server-startup-feedback-export.test.ts | 59 |
+| src/__tests__/native-finalization-recovery.test.ts | 32 |
+| src/services/execution-control-reconciliation.test.ts | 10 |
+| src/modules/wake-queue/adapters/queued-comment-postgres.test.ts | 5 |
+| src/__tests__/native-workspace-sync.test.ts | 10 |
+| src/services/heartbeat/upgrade-readiness.test.ts | 7 |
+| src/services/native-runtime/native-restart-recovery.test.ts | 25 |
+| src/services/legacy-execution-recovery.test.ts | 15 |
+
+Eleven distinct fresh PostgreSQL databases were created and migrated before
+the fixtures. The preservation owner and native-session suite have different
+databases despite sharing the same environment-variable name. This avoids
+interpreting shared fixture cleanup as a product failure. Database build, server
+typecheck/build, module boundaries, target migration order, and all eleven fresh
+0407 migrations pass. Existing embedded fixtures remain real.
+
+Only this slice's unpublished 0401 artifacts were deferred before adopting parent
+history, then generated through `pnpm db:generate` as `0407_flaky_blob.sql`.
+The concurrent idempotent index SQL is byte-identical to the original. All prior
+SQL and journal entries and retained parent snapshots are unchanged; five newest
+snapshots 0403-0407 remain. The historical snapshot overlap uses exact parent
+bytes. No new code comments are added relative to the current parent.
+
+A populated 0406 database upgrades to 0407 with all 405 prior receipts identical
+and exactly one appended. Replay appends none. The preservation index is valid
+and ready on both fresh and upgraded databases. These checks establish combined
+worker, ownership and migration behavior, not deployed host/database suspension
+or external wake delivery. Actual landed parent ancestry and fresh published-head
+review and CI remain required. Prior failures above retain their original scope.
+
+
+## Current stopped-collection parent and migration409 acceptance
+
+Only this slice's unlanded 0407 migration artifacts were deferred before the
+normal merge of published parent `d176d731f083c7edd2ede688cb901d1e66d8f1d8`.
+The merge has no conflicts. All original runtime and schema edits remain
+identical, and every unrelated upstream blob is preserved. This includes the
+new quality-review boundary, bounded retry cancellation, and independent
+sidebar fix. The migration directory and client matched the exact parent
+before generation. The first path preflight used rename detection and failed
+before mutation; explicit no-renames verifies the five expected artifacts.
+
+`pnpm db:generate` produced `0409_uneven_zuras.sql`. Its concurrent, replay-safe
+index SQL is byte-identical to the original. All prior SQL, journal entries and
+retained snapshot bytes remain unchanged. The newest five snapshots are 0405
+through 0409. No new code comments are added.
+
+Source `21b46a8e4293ca670235117458b984493a181759` passes all **1472 cases in
+16 complete files**, without skips: the 84-case owner in 14.93 seconds and
+1388 consumer cases in 133.46 seconds. This includes the full 968-case native
+session suite and existing Stop, restart, dispatch, workspace-sync and
+upgrade-readiness consumers. Twelve separate fresh PostgreSQL 17 databases
+migrated through 0409 isolate destructive fixtures. The embedded-only lifecycle
+fixture temporarily uses its own external database on this IPC-exhausted host
+and is restored byte for byte. Assertions and timing limits stay unchanged.
+This is qualified external-database acceptance, not an unmodified embedded CI
+run. Actual database build, server types/build, boundaries and exact-head
+migration-order checks pass.
+
+A populated 0408-to0409 upgrade preserves all 407 prior migration receipts and
+all 38 companies, appends exactly one receipt and adds none on replay. The
+preservation index is valid and ready on fresh and upgraded databases.
+
+Publication remains held until the parent actually merges. An explicit-base
+planning probe predicts squash-ancestry conflicts; it is not an actual merge
+proof. Its first sandboxed attempt could not write Git objects and was not
+valid evidence. The checked successful probe records the expected conflicts.
+The actual landed-parent ancestry must be reconciled and complete owned and
+unrelated upstream changes proved again before publication. Fresh current-head
+CI and reviews remain required. Earlier failures and qualifications stay recorded.
+These tests do not establish deployed host/database suspension or Cloud wakes.
+
+
+## Actual stopped-collection squash ancestry
+
+Parent #17674 merged as `82529741418e138d9d623bf13d2fa22c1ff79d36`. Its
+entire tree equals reviewed parent `d176d731f083c7edd2ede688cb901d1e66d8f1d8`,
+which is already an ancestor of the accepted child. Normal merge
+`28536423228b67541503872da6bd98b9b4c5dfef` resolves the inspected duplicate
+squash-ancestry conflicts. Its entire resulting tree equals the accepted child.
+Complete server, package, test and UI trees remain identical to the 1472-case
+source, including the whole 0409 database tree. No assertions or timing limits
+change. Fresh published-head CI and reviews remain required before merge.

packages/db/src/client.ts

@@ -558,6 +558,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
   "0406_zippy_war_machine.sql",
   "0407_greedy_star_brand.sql",
   "0408_short_red_skull.sql",
+  "0409_uneven_zuras.sql",
 
 
   "0384_funny_adam_warlock.sql",

packages/db/src/migrations/0409_uneven_zuras.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;

packages/db/src/migrations/meta/0409_snapshot.json

@@ -1,6 +1,6 @@
 {
-  "id": "d1e8fe9b-c8c9-46ed-b8e0-55bada43eeed",
-  "prevId": "e21e38de-5a82-473f-b81e-e96eff5ef4c8",
+  "id": "94acf3fe-50f3-4a97-a7fe-865abbc9d0a3",
+  "prevId": "6bd67ad8-0d80-4163-873b-8b8eacf32bb2",
   "version": "7",
   "dialect": "postgresql",
   "tables": {
@@ -2665,6 +2665,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": [
@@ -33727,6 +33771,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": [
@@ -33743,6 +33809,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": [
@@ -39111,6 +39193,140 @@
       "checkConstraints": {},
       "isRLSEnabled": false
     },
+    "public.issue_thread_interaction_idempotency_keys": {
+      "name": "issue_thread_interaction_idempotency_keys",
+      "schema": "",
+      "columns": {
+        "id": {
+          "name": "id",
+          "type": "uuid",
+          "primaryKey": true,
+          "notNull": true,
+          "default": "gen_random_uuid()"
+        },
+        "company_id": {
+          "name": "company_id",
+          "type": "uuid",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "issue_id": {
+          "name": "issue_id",
+          "type": "uuid",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "interaction_id": {
+          "name": "interaction_id",
+          "type": "uuid",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "idempotency_key": {
+          "name": "idempotency_key",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "created_at": {
+          "name": "created_at",
+          "type": "timestamp with time zone",
+          "primaryKey": false,
+          "notNull": true,
+          "default": "now()"
+        }
+      },
+      "indexes": {
+        "issue_thread_interaction_idempotency_keys_scope_key_uq": {
+          "name": "issue_thread_interaction_idempotency_keys_scope_key_uq",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "issue_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "idempotency_key",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": true,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "issue_thread_interaction_idempotency_keys_interaction_idx": {
+          "name": "issue_thread_interaction_idempotency_keys_interaction_idx",
+          "columns": [
+            {
+              "expression": "interaction_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        }
+      },
+      "foreignKeys": {
+        "issue_thread_interaction_idempotency_keys_company_id_companies_id_fk": {
+          "name": "issue_thread_interaction_idempotency_keys_company_id_companies_id_fk",
+          "tableFrom": "issue_thread_interaction_idempotency_keys",
+          "tableTo": "companies",
+          "columnsFrom": [
+            "company_id"
+          ],
+          "columnsTo": [
+            "id"
+          ],
+          "onDelete": "cascade",
+          "onUpdate": "no action"
+        },
+        "issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk": {
+          "name": "issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk",
+          "tableFrom": "issue_thread_interaction_idempotency_keys",
+          "tableTo": "issues",
+          "columnsFrom": [
+            "issue_id"
+          ],
+          "columnsTo": [
+            "id"
+          ],
+          "onDelete": "cascade",
+          "onUpdate": "no action"
+        },
+        "issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk": {
+          "name": "issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk",
+          "tableFrom": "issue_thread_interaction_idempotency_keys",
+          "tableTo": "issue_thread_interactions",
+          "columnsFrom": [
+            "interaction_id"
+          ],
+          "columnsTo": [
+            "id"
+          ],
+          "onDelete": "cascade",
+          "onUpdate": "no action"
+        }
+      },
+      "compositePrimaryKeys": {},
+      "uniqueConstraints": {},
+      "policies": {},
+      "checkConstraints": {},
+      "isRLSEnabled": false
+    },
     "public.issue_thread_interactions": {
       "name": "issue_thread_interactions",
       "schema": "",

packages/db/src/migrations/meta/_journal.json

@@ -2850,6 +2850,13 @@
       "when": 1793240590542,
       "tag": "0408_short_red_skull",
       "breakpoints": true
+    },
+    {
+      "idx": 409,
+      "version": "7",
+      "when": 1793243052342,
+      "tag": "0409_uneven_zuras",
+      "breakpoints": true
     }
   ]
 }

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')

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -116,6 +116,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: [] })),
@@ -553,6 +554,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();
@@ -965,7 +973,7 @@ describe("startServer feedback export wiring", () => {
       intervalCallback?.();
       await new Promise<void>((resolve) => setImmediate(resolve));
       expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
-      expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, 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 {

server/src/index.ts

@@ -1824,6 +1824,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;
 
@@ -1954,7 +1956,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 })
+            .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true })
             .then(() => heartbeat.promoteDueScheduledRetries())
             .then(async (promotion) => {
               await heartbeat.resumeQueuedRuns();

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({

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<CancelSuppressedRetryResult> {
     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<CancelStaleQueuedRunOutcome> {
       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

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)

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));

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) : "*");
+}

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";
@@ -104,6 +105,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));

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",

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";
@@ -185,6 +186,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)

server/src/services/heartbeat.ts

@@ -1,3 +1,4 @@
+import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js";
 import { readUpgradeReadiness } from "./heartbeat/upgrade-readiness.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { writeExecutionControl } from "./execution-control-notifications.js";
@@ -4084,6 +4085,14 @@ export function heartbeatService(
     triggerIssueMonitor,
 
     reportRunActivity: clearDetachedRunWarning,
+    registerCancelledWorkspacePreservationWork: (
+      coordinator: Parameters<typeof registerCancelledWorkspacePreservationWork>[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<typeof instructionCopies.registerStoppedCollectionWork>[0], canRun: () => boolean,
     ) => instructionCopies.registerStoppedCollectionWork(coordinator,

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";
@@ -852,6 +853,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({
@@ -974,6 +976,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();
@@ -1335,6 +1338,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({
@@ -1457,6 +1461,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({
@@ -3495,6 +3500,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
@@ -3634,6 +3640,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)

server/src/services/heartbeat/recovery.ts

@@ -1947,8 +1947,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();

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";
@@ -1025,6 +1026,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 });

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, "*");

server/src/services/native-runtime/native-failed-workspace-recovery.test.ts

@@ -1,25 +1,34 @@
+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 { activityLog, agents, companies, createDb, environmentLeases, environments, externalAgentHolds, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
+import { and, eq, inArray } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+import { activityLog, agents, companies, createDb, applyPendingMigrations, 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, settleUnrecoverableExecutions, validateExecutionReconciliation } from "../execution-recovery-resolution.js";
 import { prepareSandboxStopAndRetain, settleStopOnlyCleanup } from "../sandbox-stop-and-retain.js";
-import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, preserveOrphanedNativeWorkspace } from "./native-failed-workspace-recovery.js";
+import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, preserveOrphanedNativeWorkspace, 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";
 import { claimNativeRestartRecoveries } from "./native-restart-recovery.js";
 import { claimNativeSessionResumptions } from "./native-finalization-reconciler.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<typeof createDb>;
   let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
   beforeAll(async () => {
-    if (externalUrl) db = createDb(externalUrl);
+    if (externalUrl) {
+      await applyPendingMigrations(externalUrl);
+      db = createDb(externalUrl);
+    }
     else {
       temporary = await startEmbeddedPostgresTestDatabase("native-failed-workspace-");
       db = createDb(temporary.connectionString);
@@ -53,6 +62,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<string>();
+    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<ReturnType<typeof seed>>) => ({
+      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<ReturnType<typeof seed>>, owner = db) {
+      await lifecycle(owner).setRunStatus(f.runId, "cancelled", { resultJson: acknowledgement(f), finishedAt: new Date() });
+    }
+    async function pinned(f: Awaited<ReturnType<typeof seed>>) {
+      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<void>(resolve => { entered = resolve; });
+      const held = new Promise<void>(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<void> | undefined;
+      try {
+        const registration = registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true);
+        await reached; let complete = false;
+        stopping = worker.stop().then(() => { complete = true; });
+        await new Promise<void>(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(); }
+    });
+  });
+
   async function supersededRetry() {
     const f = await seed();
     await db.update(heartbeatRuns).set({ nativePhase: "retryable_failure", startedAt: new Date(1), finishedAt: new Date(2) })

server/src/services/native-runtime/native-failed-workspace-recovery.ts

@@ -1,3 +1,6 @@
+import type { CancelledWorkspacePreservationScope } from "../cancelled-workspace-preservation-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";
@@ -13,7 +16,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notificati
 
 export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeof heartbeatRuns.$inferSelect,
   "id" | "companyId" | "nativeIssueId" | "issueId" | "contextSnapshot" | "runnerProfileJson">,
-  options: { allowPreterminal?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) {
+  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;
@@ -27,6 +30,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
       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
@@ -42,6 +46,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
       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.kind === "plugin_task" || claim.remote.providerLeaseId !== currentReference.providerLeaseId
@@ -144,7 +149,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;
         const code = retentionContentionCode(error);
@@ -154,3 +159,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<typeof createDeliveryWorkCoordinator>, canRun: () => boolean,
+  canRecover: () => Promise<boolean>) {
+  type Entry = { id: string; companyId: string; dueAt: number };
+  type Candidate = Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "issueId" | "nativeIssueId"
+    | "runtimeMode" | "driverKind" | "status" | "resultJson" | "runnerProfileJson">;
+  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<Entry | null> {
+    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);
+}

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 { readRequiredQualityReview } from "../required-quality-review.js";
 import type { WorkspaceWorkers } from "../plugin-workspaces.js";
@@ -517,6 +518,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,

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 { hasNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
@@ -921,6 +922,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({

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 { readRequiredQualityReview } from "../required-quality-review.js";
 import { pendingNativeGovernance } from "./native-governance.js";
@@ -412,6 +413,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({
@@ -670,6 +672,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 });
@@ -1689,6 +1692,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 });

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 { providerPackWorkspaceId, recordAdmittedRemoteProviderPack, resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
 import { type ProviderPackInstallation, verifyRemoteProviderPack as verifyRemoteProviderPackIdentity } from "./remote-provider-pack-verification.js";
@@ -7655,6 +7656,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({
@@ -9819,6 +9821,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({

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<void> {
   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({

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";
@@ -1068,6 +1069,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)

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";
@@ -6249,6 +6250,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 });