diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 9ac4abb3b2b..ab2d7165048 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/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. diff --git a/doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md b/doc/plans/2026-10-25-cancelled-workspace-preservation-sleep-audit.md new file mode 100644 index 00000000000..cf28f908554 --- /dev/null +++ b/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. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index c327dfe02a1..497b7a7d87c 100644 --- a/packages/db/src/client.ts +++ b/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", diff --git a/packages/db/src/migrations/0409_uneven_zuras.sql b/packages/db/src/migrations/0409_uneven_zuras.sql new file mode 100644 index 00000000000..a010cb1a0bb --- /dev/null +++ b/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; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0404_snapshot.json b/packages/db/src/migrations/meta/0409_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0404_snapshot.json rename to packages/db/src/migrations/meta/0409_snapshot.json index 9f351bf180e..54b5f4d70ba 100644 --- a/packages/db/src/migrations/meta/0404_snapshot.json +++ b/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": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index d193710eb2c..469168c4ed3 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/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 } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 287ca78b7fb..99f06a955dc 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -111,6 +111,12 @@ export const heartbeatRuns = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + cancelledWorkspacePreservationIdx: index("heartbeat_runs_cancelled_workspace_preservation_idx").on(table.id, table.companyId) + .where(sql`${table.runtimeMode} = 'native' AND ${table.status} = 'cancelled' + AND ${table.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared' + AND ${table.resultJson}->'nativeCancellation'->>'dispatchState' = 'acknowledged' + AND (${table.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1' + AND ${table.resultJson}->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE`), retainedRecoveryAdoptionIdx: index("heartbeat_runs_retained_recovery_adoption_idx") .on(table.id).where(sql`${table.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1' and not (coalesce(${table.resultJson}, '{}'::jsonb) ? 'systemRecoveryAdoption') diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 4aba603e719..b06b38085a5 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/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((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 96e027f6834..56b877c0a9a 100644 --- a/server/src/index.ts +++ b/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(); diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 26c2a9cd517..2d3d0b2f831 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; import { and, asc, desc, eq, gt, gte, inArray, ne, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -339,6 +340,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch const transactionResult = await db.transaction(async (tx) => { await notifyInstructionCollection(tx, { companyId, runId: input.run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id }); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index aef69b49971..a0b59f04419 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js"; import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; @@ -669,6 +670,7 @@ export function createPostgresRunDispatchAdapter( | { applied: false } > { await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); + await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId }); await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: issueId ?? undefined }); const [row] = await tx @@ -713,6 +715,7 @@ export function createPostgresRunDispatchAdapter( input: CancelSuppressedRetryInput, ): Promise { await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); + await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId }); await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); const [row] = await tx @@ -921,6 +924,7 @@ export function createPostgresRunDispatchAdapter( providerDispatchNotStarted = false, ): Promise { await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [row] = await tx diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index 01f0ef27acb..5dcd57c3485 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; @@ -105,6 +106,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu async cancelQueueRun({ queueRunId, reason, now }) { await notifyInstructionCollection(tx, { companyId, runId: queueRunId }); + await notifyCancelledWorkspacePreservation(tx, { companyId, runId: queueRunId }); await notifyChatRunMilestones(tx, { companyId, issueId }); const row = await tx .update(heartbeatRuns) diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index 0ed1a03ae40..1f207367051 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; @@ -50,6 +51,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str await notifyInstructionCommit(tx, { companyId: companyId }); await notifyDirectoryCleanup(tx, { companyId: companyId }); await notifyInstructionCollection(tx, { companyId: companyId }); + await notifyCancelledWorkspacePreservation(tx, { companyId: companyId }); await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, id)); await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, id)); await tx.delete(agentApiKeys).where(eq(agentApiKeys.agentId, id)); diff --git a/server/src/services/cancelled-workspace-preservation-notifications.ts b/server/src/services/cancelled-workspace-preservation-notifications.ts new file mode 100644 index 00000000000..9294c30d7b9 --- /dev/null +++ b/server/src/services/cancelled-workspace-preservation-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type CancelledWorkspacePreservationScope = { companyId: string; runId?: string }; + +export async function notifyCancelledWorkspacePreservation(transaction: object, scope?: CancelledWorkspacePreservationScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.cancelledWorkspacePreservation, scope ? JSON.stringify(scope) : "*"); +} diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 0ccd4499d77..e7a6249e9f2 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; @@ -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)); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index ae7f4a22c36..ff596f86a79 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + cancelledWorkspacePreservation: "cancelled-workspace-preservation", instructionCollection: "instruction-collection", directoryCleanup: "directory-cleanup", instructionCommit: "instruction-commit", diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 1c4ad3d0885..da1c2360f2b 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { randomUUID } from "node:crypto"; @@ -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) diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f56ad8b529c..5df7f4e1408 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js"; import { 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[1], canRun: () => boolean, + ) => registerCancelledWorkspacePreservationWork(db, coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerStoppedInstructionCollectionWork: ( coordinator: Parameters[0], canRun: () => boolean, ) => instructionCopies.registerStoppedCollectionWork(coordinator, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index f3c2366c219..ecd913d0d16 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatResponseAvailability } from "../chat-response-availability.js"; @@ -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) diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 28fc3077f09..350568dd3a5 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/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(); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 8729a0f2ab8..da93922cd26 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; @@ -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 }); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 2d54950c51b..46f531d4712 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; import { notifyInstructionCommit } from "./instruction-commit-notifications.js"; @@ -624,6 +625,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyInstructionCommit(tx); await notifyDirectoryCleanup(tx); await notifyInstructionCollection(tx); + await notifyCancelledWorkspacePreservation(tx); } if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx); await notifyToolConnectionHealth(tx, "*"); diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts index eb9425f715e..0b740cee829 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts @@ -1,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; let temporary: Awaited>; 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(); + afterEach(async () => { + vi.useRealTimers(); vi.restoreAllMocks(); + if (fixtureCompanies.size) await db.update(heartbeatRuns).set({ status: "failed" }) + .where(inArray(heartbeatRuns.companyId, [...fixtureCompanies])); + fixtureCompanies.clear(); + }); + async function prepared() { + const fixture = await seed(); fixtureCompanies.add(fixture.companyId); return fixture; + } + const acknowledgement = (f: Awaited>) => ({ + cancelledByActorType: "user", cancelledByUserId: "operator", nativeCancellation: { + schema: "paperclip.native-cancellation.v1", runId: f.runId, companyId: f.companyId, issueId: f.issueId, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID(), + }, + }); + const clock = () => vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const coordinator = (canRun = () => true, canReconcile = () => true) => + createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile, onError: () => {} }); + const lifecycle = (owner = db) => createHeartbeatLifecycle(owner, { + ...createHeartbeatRunState(owner), ...createHeartbeatRunPreparation(owner), + issuesSvc: { addComment: vi.fn(), listReviewAttention: vi.fn() }, + recovery: { escalateStrandedAssignedIssue: vi.fn() }, + companySkills: { completeTestRunForIssue: vi.fn() }, + budgets: { getInvocationBlock: vi.fn(async () => null) }, + treeControlSvc: { getActivePauseHoldGate: vi.fn(async () => null) }, + enqueueWakeup: vi.fn(), getCurrentUserRedactionOptions: vi.fn(async () => ({})), + getAgentInvokability: vi.fn(async () => ({ invokable: true as const })), + emitTerminalAgentTaskRun: vi.fn(), budgetHooks: {}, + }); + async function cancel(f: Awaited>, owner = db) { + await lifecycle(owner).setRunStatus(f.runId, "cancelled", { resultJson: acknowledgement(f), finishedAt: new Date() }); + } + async function pinned(f: Awaited>) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(100); + expect((await readRun(f.runId)).resultJson).toMatchObject({ + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }, + }); + }, { timeout: 5000 }); + expect((await readLease(f.lease.id)).leasePolicy).toBe("retain_on_failure"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_progress"); + expect((await readRun(f.runId)).status).toBe("cancelled"); + expect(await actions(f.issueId)).toHaveLength(1); + } + it("stays quiet for active runs, completed preservation, and admitted workspace-less sessions", async () => { + const active = await prepared(), settled = await prepared(), external = await prepared(); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, active.runId)); + await cancel(settled); await preserveNativeFailedWorkspace(db, await readRun(settled.runId), { allowPreterminal: true }); + await bindExternalInput(external, "muse"); await cancel(external); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + expect((await readLease(active.lease.id)).status).toBe("active"); + expect((await readLease(external.lease.id)).status).toBe("active"); + } finally { await worker.stop(); } + }); + it.each(["missing issue", "foreign company"] as const)("does not schedule an unbound cancellation receipt: %s", async mismatch => { + const f = await prepared(); const result = acknowledgement(f); + if (mismatch === "missing issue") await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ status: "cancelled", + ...(mismatch === "missing issue" ? { nativeIssueId: null } : {}), + resultJson: { ...result, nativeCancellation: { ...result.nativeCancellation, + ...(mismatch === "missing issue" ? { issueId: null } : { companyId: randomUUID() }), + } }, + }).where(eq(heartbeatRuns.id, f.runId)); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + expect((await readLease(f.lease.id)).status).toBe("active"); + } finally { await worker.stop(); } + }); + it("reactivates from the real run-status producer and becomes quiet after recording the hold", async () => { + const f = await prepared(); clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect(worker.nextWakeAt()).toBeNull(); + await cancel(f); await pinned(f); + await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }); + it.each(["restart", "drain", "standby"] as const)("retains preservation across %s", async mode => { + const f = await prepared(); if (mode === "restart") await cancel(f); + let admitted = mode === "restart"; + clock(); const worker = coordinator(() => admitted, () => admitted || mode !== "standby"); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => admitted, async () => true).ready; + if (mode !== "restart") { + await cancel(f); await vi.advanceTimersByTimeAsync(1000); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect((await readLease(f.lease.id)).status).toBe("active"); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); + } + await pinned(f); + } finally { await worker.stop(); } + }); + it("waits for the outer cancellation commit and discards a rolled-back transition", async () => { + const f = await prepared(); clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + await expect(db.transaction(async tx => { + await cancel(f, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(1000); + expect((await readLease(f.lease.id)).status).toBe("active"); + throw new Error("rollback cancellation"); + })).rejects.toThrow("rollback cancellation"); + await vi.advanceTimersByTimeAsync(1000); + expect((await readRun(f.runId)).status).toBe("failed"); + await db.transaction(async tx => { + await cancel(f, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(1000); + expect((await readLease(f.lease.id)).status).toBe("active"); + }); + await pinned(f); + } finally { await worker.stop(); } + }); + it("preserves owed work while another controller owns finalization", async () => { + const f = await prepared(); await cancel(f); + await db.update(nativeRunFinalizations).set({ leaseOwner: "other-controller" }).where(eq(nativeRunFinalizations.runId, f.runId)); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect((await readLease(f.lease.id)).status).toBe("active"); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(20_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await db.update(nativeRunFinalizations).set({ leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await vi.advanceTimersByTimeAsync(10_000); await pinned(f); + } finally { await worker.stop(); } + }); + it("waits for exact stop-only proof before retaining an already-released source", async () => { + const f = await prepared(); await cancel(f); + const stopping = await prepareSandboxStopAndRetain(db, f.lease); + if (!stopping) throw new Error("Expected stop-only source"); + clock(); const worker = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect((await readRun(f.runId)).resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(20_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await settleStopOnlyCleanup(db, stopping, { attemptId: String(stopping.metadata?.pendingCleanupAttemptId), + receipt: { providerLeaseId: f.lease.providerLeaseId, state: "stopped" } }); + await vi.advanceTimersByTimeAsync(10_000); await pinned(f); + expect((await readLease(f.lease.id)).status).toBe("released"); + } finally { await worker.stop(); } + }); + it("retries failed owner transactions without blocking another cancelled source", async () => { + const first = await prepared(), second = await prepared(); await cancel(first); await cancel(second); + clock(); const worker = coordinator(); + const transactions = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("preservation database unavailable")); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + transactions.mockRestore(); + const policies = [(await readLease(first.lease.id)).leasePolicy, (await readLease(second.lease.id)).leasePolicy]; + expect(policies.filter(policy => policy === "retain_on_failure")).toHaveLength(1); + await vi.advanceTimersByTimeAsync(30_000); await pinned(first); await pinned(second); + } finally { transactions.mockRestore(); await worker.stop(); } + }); + it.each(["acknowledgement", "phase", "status"] as const)("rejects changed %s at the locked preservation boundary", async changed => { + const f = await prepared(); await cancel(f); clock(); const worker = coordinator(); + const transaction = db.transaction.bind(db); let changedOnce = false; + const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => { + if (!changedOnce) { + changedOnce = true; + if (changed === "phase") await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + else if (changed === "status") await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.runId)); + else { + const result = acknowledgement(f); + result.nativeCancellation.dispatchState = "pending"; + await db.update(heartbeatRuns).set({ resultJson: result }).where(eq(heartbeatRuns.id, f.runId)); + } + } + return transaction(callback, options); + }) as typeof db.transaction); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + transactions.mockRestore(); expect(changedOnce).toBe(true); + expect((await readLease(f.lease.id)).status).toBe("active"); + expect((await readRun(f.runId)).resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + if (changed === "phase") { + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("observed"); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await vi.advanceTimersByTimeAsync(30_000); + } else await cancel(f); + await pinned(f); + } finally { transactions.mockRestore(); await worker.stop(); } + }); + it("reconstructs more than one page and pins all sources in bounded batches", async () => { + const fixtures = []; + for (let index = 0; index < 103; index++) { + const f = await prepared(); await cancel(f); fixtures.push(f); + } + clock(); const worker = coordinator(); + const retained = () => db.select({ id: environmentLeases.id }).from(environmentLeases) + .where(and(inArray(environmentLeases.companyId, [...fixtureCompanies]), eq(environmentLeases.leasePolicy, "retain_on_failure"))); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect((await retained()).length).toBeLessThanOrEqual(5); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await retained()).toHaveLength(103); + }, { timeout: 20_000 }); + for (const f of [fixtures[0], fixtures[102]]) await pinned(f); + await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }, 60_000); + it("joins an admitted preservation transaction during shutdown", async () => { + const f = await prepared(); await cancel(f); clock(); const worker = coordinator(); + const transaction = db.transaction.bind(db); + let entered!: () => void, release!: () => void; + const reached = new Promise(resolve => { entered = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const transactions = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + entered(); await held; return transaction(callback, options); + }) as typeof db.transaction); + let stopping: Promise | undefined; + try { + const registration = registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true); + await reached; let complete = false; + stopping = worker.stop().then(() => { complete = true; }); + await new Promise(resolve => setImmediate(resolve)); expect(complete).toBe(false); + release(); await stopping; await registration.ready; await pinned(f); + const restarted = coordinator(); + try { + await registerCancelledWorkspacePreservationWork(db, restarted, () => true, async () => true).ready; + expect(restarted.nextWakeAt()).toBeNull(); + } finally { await restarted.stop(); } + } finally { release(); transactions.mockRestore(); await stopping; await worker.stop(); } + }); + }); + async function supersededRetry() { const f = await seed(); await db.update(heartbeatRuns).set({ nativePhase: "retryable_failure", startedAt: new Date(1), finishedAt: new Date(2) }) diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index a311bc96328..b93c94aa897 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -1,3 +1,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, - 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'nativeWorkspaceSync'->>'state' = 'prepared' + AND ${heartbeatRuns.resultJson}->'nativeCancellation'->>'dispatchState' = 'acknowledged' + AND (${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1' + AND ${heartbeatRuns.resultJson}->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE`; +} + +export function registerCancelledWorkspacePreservationWork(db: Db, + coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise) { + type Entry = { id: string; companyId: string; dueAt: number }; + type Candidate = Pick; + const eligible = (run: Candidate) => run.nativeIssueId !== null && hasAcknowledgedNativeStopIntent(run) + && readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync)?.state === "prepared" + && !hasRequiredWorkspaceRecovery(run.resultJson) && !nativeRunHasNoWorkspaceAccess(run); + const scopeCondition = (scope?: CancelledWorkspacePreservationScope) => scope ? and( + eq(heartbeatRuns.companyId, scope.companyId), scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined) : undefined; + async function load(scope: CancelledWorkspacePreservationScope | undefined, signal?: AbortSignal) { + if (!await canRecover()) return []; + const entries: Entry[] = []; + let cursor: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, + driverKind: heartbeatRuns.driverKind, status: heartbeatRuns.status, resultJson: heartbeatRuns.resultJson, + runnerProfileJson: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and(pendingCancelledWorkspace(), scopeCondition(scope), + cursor ? gt(heartbeatRuns.id, cursor) : undefined)).orderBy(asc(heartbeatRuns.id)).limit(100); + for (const row of rows) if (eligible(row)) entries.push({ id: row.id, companyId: row.companyId, dueAt: 0 }); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + throw new Error("Cancelled workspace preservation reconstruction stopped"); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + if (!await canRecover()) return null; + const [run] = await db.select().from(heartbeatRuns) + .where(and(pendingCancelledWorkspace(), scopeCondition({ companyId: entry.companyId, runId: entry.id }))); + if (!run || !eligible(run)) return null; + if (!canRun() || signal?.aborted) throw new Error("Cancelled workspace preservation stopped"); + const [finalization] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.companyId, run.companyId), + eq(nativeRunFinalizations.issueId, run.nativeIssueId!), + )); + if (finalization?.resultId) return null; + if (finalization?.phase !== "terminal_failure" || finalization.leaseOwner) + return { ...entry, dueAt: Date.now() + 30_000 }; + const [source] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + or(eq(environmentLeases.status, "active"), and(inArray(environmentLeases.status, ["pending_cleanup", "released"]), + eq(environmentLeases.leasePolicy, "ephemeral"))), + )).limit(1); + if (source) await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); + const [current] = await db.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))); + return current && current.status === "cancelled" && eligible(current) ? { ...entry, dueAt: Date.now() + 30_000 } : null; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.cancelledWorkspacePreservation, batchSize: 5, + decodeScope: (hint: string): CancelledWorkspacePreservationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: CancelledWorkspacePreservationScope) => !scope || + entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId), + load, check, + }, canRun); +} diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 9505af6f60f..df7d36fa7b0 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { 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, diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 1325a45d8c4..3c34f8b7b58 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { 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({ diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 4fc1345e17d..6647b68b8e2 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { 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 }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index d40029b85a4..467098f6138 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { 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({ diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index af87f40d413..7ff930146ef 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { emitStartupReuse, type StartupReuseReason } from "@paperclipai/adapter-utils/startup-reuse"; @@ -614,13 +615,14 @@ async function persistRunReference( ): Promise { await db.transaction(async (tx) => { const run = await tx - .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) + .select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)) .for("update") .limit(1) .then((rows) => rows[0] ?? null); if (!run) throw new Error("native_workspace_sync_run_missing"); + await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId }); await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 157be7c0abe..980b797152c 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; @@ -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) diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index e8f708ed602..fa02e7a09af 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { boundRunIssueId } from "../run-task-binding.js"; @@ -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 });