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 53535b60814..73a6a22c916 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2629,3 +2629,36 @@ The native finalization audit found separate historical unsafe-export recovery, Obsolete policy-review cleanup is draft [#17679](https://github.com/paperclipai/paperclip/pull/17679) at `e671c66bba6e9a42aa570826e6f3420dba199399`. All 53 checks pass, Greptile reviewed that head at 5/5, and no unresolved threads remain. It is mergeable and unmerged. Slice 45 owns automatic completion-review retirement only. It preserves direct finalizer ordering and the durable withdrawal trigger for reassessment. The worker's raw lookup propagates failures so failed discovery cannot look empty. See [the audit](2026-10-25-automatic-completion-review-sleep-audit.md). All 370 local owner/startup/consumer tests pass, as do workspace types/build. Exact-head remote gates remain required. Older stack CI/review gaps, master reconciliation and the remaining original duties are still outstanding. + +## Slice 45 accepted; slice 46 unsafe-export audit + +[#17680](https://github.com/paperclipai/paperclip/pull/17680) at +`be668fb975f119efeb4830405d4bb1ab65791f87` has 53 successful remote gates, +Greptile 5/5 at that head, and no unresolved review threads. It remains draft +and unmerged. The unsafe-export baseline passed all 51 tests on a fresh external +PostgreSQL database (`/tmp/sleep46-baseline.log`). Slice 46 must distinguish +action retirement and eligible result recovery from dormant historical failure. +The periodic backstop remains until all reactivation producers are covered. + +Slice 46 now has a scoped worker and initial producer coverage, with the periodic +backstop deliberately retained until the direct-writer audit is complete. See +[the unsafe-export audit](2026-10-25-unsafe-workspace-export-sleep-audit.md) for +work predicates, remaining producers, test evidence, and preserved failures. + + +## Slice 46 producer wiring and verification + +The unsafe-export worker now registers through heartbeat/startup and the periodic +reaper skips its historical scan. Direct task/run/action producers signal their +existing transaction boundaries. Physical-owner contention, real admin/stale-lock +release, queued-run cancellation, cancellation during ownership contention, +shutdown/restart, and multiple discovery pages are tested. The owner/startup and +consumer selections establish 626 distinct passing cases across 13 files after +running the initially skipped lifecycle cases against an explicit database. + +A measured retained-history scan justified partial index migration 0398; fresh +application, replay, and validity checks passed. The owner audit records timings, +producer coverage, fixture failures, and the supported writer boundary. Workspace +typechecks and build passed; remote acceptance remains outstanding. Origin/master was fetched at +`6336c34041`; the full stack still needs final reconciliation with master. No PR +in this stack has been merged, and the original remaining scope is unchanged. diff --git a/doc/plans/2026-10-25-unsafe-workspace-export-sleep-audit.md b/doc/plans/2026-10-25-unsafe-workspace-export-sleep-audit.md new file mode 100644 index 00000000000..6a515f6ec4d --- /dev/null +++ b/doc/plans/2026-10-25-unsafe-workspace-export-sleep-audit.md @@ -0,0 +1,361 @@ +# Historical unsafe workspace export recovery + +Current integration: 2026-10-29. Prepared parent: automatic completion-review +retirement #17680 at actual squash `73d24aa5acfcf17b3f676b4da7cd05eb70b60b99`. +Earlier draft parent: `be668fb975f119efeb4830405d4bb1ab65791f87`. +Startup registers the worker when maintenance is enabled. The periodic reaper +skips this duty. Explicit/manual recovery remains available. Fresh published-head +CI and review gates remain required; older parent scores do not prove them. + +## Problem and owner + +`native-runtime/native-workspace-export-recovery.ts` recovers old controllers' +`native_workspace_sync_out_unsafe_archive` failures. It does not retry an unsafe +archive or launch another provider turn. It can retire stale repair notices and +readmit an already accepted result after recording an explicit workspace omission. + +The old caller in `native-finalization-reconciler.ts` visits 25 terminal failure +rows per pass. Ineligible historical rows remain in that set forever. Merely +putting those same rows on a 30-second deadline would preserve idle polling. + +The new scoped worker uses the existing delivery coordinator, transaction-aware +notifications, scoped deadline helper, and scheduler. No table, provider branch, +independent timer, or host-sleep exemption is added. One partial index supports +scoped discovery without scanning completed history. The existing recovery owner +still performs all mutations and validates current state under its original +physical-owner and database-lock boundaries. The manual/startup scan still uses +that owner; extraction alone does not remove its existing behavior. + +## Work predicate + +Discovery joins the coordinator, exact native run, issue, and accepted result. +Company, issue, run, completion-contract, and result identities must agree. +The coordinator must still have the historical unsafe terminal failure. +One of two obligations must exist: + +1. A stale unsafe action for this run is active/escalated, or resolved with the + old `new_source_execution_path` note. Retire that notice even if later work + prevents result readmission. +2. The existing readmission conditions hold: no explicit action settlement, + no newer run or other queued/running run for this issue, no competing active + recovery action, the latest contract matches, the issue is blocked/in-review + and assigned to this run's agent, the run is failed/succeeded with terminal + native failure, and checkout/execution ownership is absent or belongs to it. + +A cancelled recovery action is an explicit veto. A resolved action with a null +or other non-automatic resolution note is also a veto. These are preserved. +Only the old automatic-resolution notes permit readmission. + +A coordinator lease or physical owner can delay an actual obligation. The worker +retains its retry; it does not infer that the owner stopped because time elapsed. +Dormant history has no retry. Relevant committed state changes must rediscover +it. The producer audit below records the supported local writers. + +Discovery uses 100-row keyset pages. Dispatch uses the shared fair company batch +with a five-entry limit. Exact pending work retries after 30 seconds; exceptions +use the shared bounded backoff. The check reads the exact run again before +calling the owner, then retains a deadline only if the work predicate still +holds. Successful recovery leaves an omission barrier, never a copyback receipt. + +## Producer coverage checkpoint + +Implemented: + +- The real finalization-failure writer signals before committing an unsafe + terminal failure. It does not signal unrelated failure codes. +- `issueService.update` signals changed status, assignee-agent, execution-run, + and checkout-run fields in its existing transaction. +- Recovery-action insertion, identity supersession, and relevant cause/run + identity changes signal in their existing transactions. +- Recovery-action resolution checks for a matching active action before + registering an intent. Repeated resolution of an already-settled action does + not wake this worker. Operator-guidance/attempt metadata alone does not wake it. +- The shared heartbeat run-status writer signals changed status, using the + already-loaded native issue and context issue identities. No extra lookup is + needed to identify those issues. Both distinct issue references are covered. + +Additional producer coverage: + +| State that changes eligibility | Existing owner paths now signalling before their writes | +| --- | --- | +| Task status, assignment, or stale ownership | Task update, terminal checkout/execution release, admin force release, task-tree restoration, archived-member reassignment, chat conversation settlement, queued-comment release, status-decision effects, queue dependency blocking, stale-lock recovery | +| Competing queued/running run becomes terminal or leaves that set | Lifecycle status writes, queue cancellation, dispatch suppression/stale-run cancellation, retry deferral, active-run watchdog, execution-control recovery, orphan recovery, legacy settlement, native finalizer, finalization reconciler, restart recovery, and session executor | +| Recovery action status or lineage | Recovery-action service, issue repair routes, execution/legacy recovery resolution, workspace reconstruction, safe replacement, explicit continuation | +| Historical unsafe failure first appears | Finalizer failure, interaction bridge finalization error, coordinator-store retry marking | +| Deletion removes a competing run | Agent and company deletion signal the company before deleting runs | + +Signals carry only company and optional issue IDs. Run writers use their already +loaded native issue and legacy context issue; both distinct references are sent. +Signals do not pass service contexts, provider data, or database objects through +JSON. Existing transaction owners remain authoritative. The few root writes +without a transaction now wrap only their database mutation and signal, not +provider calls. No caller gains new recovery authority. + +The writer inventory inspected issue status/assignment/ownership, run status/ +native phase/context, and action status/cause/evidence writes, including variable +patches. Metadata-only writes do not create eligibility. Monitor patches change +only monitor policy/state/deadlines. Workspace-binding patches change workspace +fields. Ordinary checkout/adoption moves to in-progress, and ordinary release +clears assignment; neither enables this accepted-result path. New-run ownership +also introduces a newer or active run, which prevents old-result readmission. +Queued-run suppression restores queued state, which remains a competing run. +Run coalescence retains the admitted task scope; ordinary context enrichment adds +comments, routing, session, or delivery metadata. Accepted result identity and +completion-contract history are immutable in supported service paths. The raw +SQL audit found no additional relevant state mutation. + +Independent SQL writers and another server process are outside the existing +process-local notification contract. Startup reconstructs committed work. This +slice adds no cross-process listener or provider wake contract. + +The heartbeat facade supplies admission state. Startup registers only in the +enabled-maintenance branch. The periodic reaper passes +`skipUnsafeWorkspaceExports`; direct/manual calls default to the old scan. +Shutdown stops the shared coordinator and joins its work. Ownership contention +keeps a concrete retry; stopped/drained execution cannot silently discard it. + +## Query cost and migration + +The measured SQL was captured from the actual worker through the PostgreSQL +client debug callback. A rolled-back fixture retained 10,000 completed native +results/coordinators and one stale unsafe notice. Without an index, the startup +query visited 10,074 result rows and took 9.827 ms. Scoped discovery removed +10,075 coordinator rows by filter and took 2.625 ms. LIMIT alone did not bound +history work. + +The partial index on `(company_id, issue_id, run_id)`, restricted to the exact +unsafe terminal-failure state, reduced these measurements to 0.105 ms and +0.046 ms. The plans selected one coordinator from the partial index. These are +local fixture timings, not deployment latency or fleet cost claims. Dormant +unsafe history still requires eligibility evaluation when its scope changes. + +Current migration `0410_freezing_kinsey_walden.sql` creates the index concurrently with +`IF NOT EXISTS` through the existing top-level migration runner. Fresh database +application, migration replay, direct SQL replay, and ready/valid index inspection +passed. The current generated snapshot set retains 0406–0410. Historical SQL and journal +entries remain intact. The index adds write/storage cost only for matching rows. + +## Evidence and preserved failures + +Baseline: 51 existing export tests passed on fresh PostgreSQL in +`/tmp/sleep46-baseline.log`. + +Initial worker run: 65 passed, four failed (`/tmp/sleep46-owner-v1.log`). The new +producer fixture incorrectly inserted a preexisting unsafe action with a +non-production fingerprint. The actual failure writer superseded it with a +cancelled record. The existing explicit-settlement veto correctly prevented +readmission. The fixture now starts before that recovery action exists. No +production veto was weakened. The failed database remains available as +`paperclip_sleep46_owner_v1_20261025_test` on the local test Postgres instance. + +The next export run passed all 70 cases (`/tmp/sleep46-owner-v2.log`). Real issue +and recovery-action producers then replaced the direct-hint fixture; all 71 +export cases passed. A combined command also ran one non-database recovery test +but skipped 88 PostgreSQL cases because their separate environment variable was +unset (`/tmp/sleep46-owner-v3.log`). Those skips are not acceptance evidence. +With its dedicated database explicitly set, all 89 recovery-action tests passed +(`/tmp/sleep46-recovery-v1.log`). + +Server typechecks passed after the worker, producer, and no-op notification +changes. The final combined run passed all 161 tests with no skips: 72 export-owner +cases and 89 recovery-action cases, including the no-op notification regression, +with separate explicit databases (`/tmp/sleep46-producers-v4.log`). + +Current tests establish empty/history quiet, explicit settlement fences, stale +notice retirement without result replay, real failure creation, outer transaction +commit/rollback, restart, drain/resume, coordinator-lease retry, failed discovery, +and real task/recovery-action reactivation. The post-wiring selection below adds startup, ownership, and direct-producer +coverage beyond this initial checkpoint. + +Post-wiring verification: + +- The first expanded selection passed 226 cases and failed one new physical-owner + assertion (`/tmp/sleep46-owner-v5.log`). It inspected the next deadline while a + worker attempt was in flight. The test now waits for that attempt to settle; + no production ownership or scheduling rule was weakened. +- The next selection passed 229 cases. After adding real queued-run cancellation + and the index migration, all 230 owner/recovery-action/startup cases passed on + fresh databases (`/tmp/sleep46-owner-v7.log`). This includes physical ownership, + task cancellation during contention, admin and stale-lock release, shutdown/ + restart, and 103 stale notices across discovery pages. +- Ten affected consumer suites passed 374 cases and skipped 22 lifecycle database + cases because embedded PostgreSQL was unavailable. The lifecycle suite now + accepts the validated explicit external-database option. Its 31 cases then all + passed, including those 22. Combined with the owner selection, this establishes + 626 distinct passing cases across 13 files. Logs: `/tmp/sleep46-consumers.log` + and `/tmp/sleep46-lifecycle.log`. +- Query-plan fixture setup first violated the existing unique active-action rule; + the fixture now settles its old active action before inserting the owed notice. + All fixture data is rolled back. The capture script initially used a missing + root runner and then the wrong module format; using the database package runner + with an ES module captured the real SQL successfully. + +Workspace typecheck and build passed. Exact-head remote CI/review remain required +before this slice is accepted. Other stack gates and migration reconciliation with +master remain outstanding. + +## Notice-retirement follow-up + +Retiring an older run's stale unsafe-export notice can make a newer accepted +result on the same issue eligible. The owner now registers an issue-scoped +notification in its existing transaction before that retirement, using the +already-loaded action rows to avoid signalling an unchanged notice. The next +pass rediscovers the issue, rather than only rechecking the older run. Existing +physical ownership and eligibility fences are unchanged. + +A real PostgreSQL regression first reproduced the newer result remaining in +terminal failure after the older notice resolved. It now proves automatic +readmission without another write or restart, preservation of both accepted +results, no extra run or provider call, and quiet after settlement. All 80 +export-owner tests and 63 startup tests passed. The first combined rerun failed +export fixture setup because the reused database retained its fixed company +prefix; the complete export suite subsequently passed on a fresh dedicated +database. Final integrated typecheck, build, and remote checks remain required. + + +## Current migration410 integration and complete acceptance + +Only this slice's unlanded 0402 migration artifacts were deferred before a +normal merge of the current automatic-review parent. There are no conflicts. +The complete original runtime/schema patch and every unrelated upstream blob +are preserved. The migration directory and client matched the exact parent +before generation. `pnpm db:generate` generated +`0410_freezing_kinsey_walden.sql`. Its concurrent, idempotent SQL is byte-identical +to the original. All prior SQL, journal entries and retained snapshot bytes +remain unchanged, with five snapshots 0406 through 0410. Nine newly added +comment lines in two files were removed under CONTRIBUTING.md. The full +non-comment TypeScript token streams are identical. The explanations above +retain the notice-retirement, omission barrier and diagnostic ownership facts. + +Source `2a13dd08a6cc52f0adab09e3df9ce31e97b746b9` passes **2188 cases across +24 complete files**, without skips. The main selection passes 2129 cases in +192.58 seconds; the independent safe-replacement suite passes 59 in 26.04 +seconds. The selection covers real export and recovery-action producers, +workspace reconstruction, explicit continuation, interaction routing, Stop, +queue admission, Slack and all 968 native-session cases. Per-file counts: + +| Complete file | Passing cases | +| --- | ---: | +| src/__tests__/issue-recovery-actions.test.ts | 89 | +| src/__tests__/native-finalization-recovery.test.ts | 32 | +| src/__tests__/native-interaction-bridge.test.ts | 7 | +| src/__tests__/native-workspace-sync.test.ts | 10 | +| src/__tests__/recovery-stale-issue-lock-sweep.test.ts | 40 | +| src/__tests__/server-startup-feedback-export.test.ts | 65 | +| src/__tests__/slack-conversation-lifecycle.test.ts | 41 | +| src/modules/run-dispatch/adapters/postgres.test.ts | 69 | +| src/modules/wake-queue/adapters/postgres.test.ts | 93 | +| src/modules/wake-queue/adapters/queued-comment-postgres.test.ts | 5 | +| src/services/execution-control-reconciliation.test.ts | 10 | +| src/services/explicit-native-continuation.test.ts | 335 | +| src/services/heartbeat/queue.test.ts | 32 | +| src/services/heartbeat/recovery.test.ts | 45 | +| src/services/heartbeat/run-control.test.ts | 51 | +| src/services/heartbeat/run-lifecycle.test.ts | 31 | +| src/services/heartbeat/upgrade-readiness.test.ts | 7 | +| src/services/legacy-execution-recovery.test.ts | 15 | +| src/services/native-runtime/native-restart-recovery.test.ts | 25 | +| src/services/native-runtime/native-run-finalizer-telemetry.test.ts | 29 | +| src/services/native-runtime/native-safe-replacement.test.ts | 59 | +| src/services/native-runtime/native-session-executor.test.ts | 968 | +| src/services/native-runtime/native-workspace-export-retry.test.ts | 80 | +| src/services/native-runtime/native-workspace-reconstruction.test.ts | 50 | + +Twenty separate fresh PostgreSQL 17 databases migrate through 0410 before +setup. Destructive fixtures do not share databases. Four embedded-only test +imports temporarily use their own external database on this IPC-exhausted +host: interaction bridge, workspace reconstruction, Slack lifecycle and +wake-queue PostgreSQL. They are restored byte for byte. All assertions and +original timing limits remain unchanged. The lifecycle suite uses the supported +external URL already added by this slice. This is qualified external-database +acceptance, not an unmodified embedded CI reproduction. Actual database build, +server types/build, module boundaries and exact-head migration order pass. + +A populated 0409-to0410 upgrade preserves all 408 prior migration receipts and +all 38 companies, appends exactly one receipt and adds none on replay. The +unsafe-export index is valid and ready on fresh and upgraded databases. + +These checks preserve all earlier failures recorded above and verify current +worker, ownership and affected-consumer behavior. Publication remains held until +the actual prerequisite merges and its ancestry is reconciled. Fresh +current-head CI and reviews remain required. These tests do not establish +deployed host/database suspension, external wake delivery or fleet savings. + + +## Prepared parent carrying published cleanup ancestry + +Normal merge `57ddb838cc93d5d8d0e7f8a0c2d1a2beed8b4394` adopts prepared parent +`790a27d28f23be0f436eb96a884dc7988ff10823`. Its source equals the prior accepted parent; +only parent documentation and ancestry change. Complete server, package, UI +and test trees remain identical to the accepted 2188-case source, including +the whole 0410 database tree. Publication remains held for the actual +automatic-review merge and fresh published-head CI and reviews. + + +## Inherited corpus repeatability fix and source-preserving acceptance + +Normal integration `8f43c9f81fac23b50997e7f0e1d0d9cf383a8109` carries the +verified prefix correction through the updated automatic-review parent. The +only changed source file relative to the accepted child is the status-authority +corpus: its two fixed prefix values now use their own company UUIDs. All +production source, all packages including the complete 0410 database tree, and +all 24 files in the accepted 2188-case selection remain byte-identical to their +accepted source. No assertion, timing limit or teardown changes. The prior +2188-case evidence retains that precise source scope. + +The additional complete 72-case status-authority corpus passes at this source +in 14.00 seconds, then passes again on the same fresh 0410 database in 13.08 +seconds. Three retained PSC/OUT/sentinel company rows remain unchanged in the +inspected ID/name/prefix fields after both runs; each invocation appends two +distinct UUID-derived prefixes. Database build and exact-head migration order +pass. This adds one independently checked file to the previous selection; the +repeat is not counted again as distinct coverage. Previous collision and +rejected-deletion failures remain recorded in the parent audit. + +The actual automatic-review prerequisite still must merge before publication. +Fresh published-head CI and review are required, with existing sleep and +external-wake limitations unchanged. + + +## Published automatic-review parent + +Normal merge `6771a6c648c6ac1e9f0f73a91710d61bf5608aaf` adopts published parent +`40f84082526d6ba405f206475c919c3b4b67ec4c`. All server, package, UI and test trees +remain identical to the accepted child, including the 2188-case runtime/test +scope, the additional 72-case corpus with its same-database repeat, and the +complete 0410 database tree. Only parent documentation and ancestry change. +Publication stays held until the actual automatic-review squash and its +ancestry reconciliation. Fresh current-head CI and reviews remain required. + + +## Inherited CI observation correction and independent master assertion + +Normal integration `4d51231abe8b68c21ce896d0d2ac16865978c8ab` adopts +published automatic-review parent `181bf800a79449961e09c995f2785dd7f8060bb3`. +The only source changes are the initial GitHub idle poll's established +five-second observation argument and master's company-skill cleanup assertion. +All production source, all packages including migration 0410, all 24 accepted +suite files and the additional 72-case corpus remain byte-identical. The +previous acceptance keeps that precise source scope; it is not rerun or +recounted here. The parent audit retains the original CI failure and controlled +comparison, including its inability to reconstruct the exact CI token timeline. + +The two native-presentation GitHub idle variants pass in 15.14 seconds on this +runtime. All six cases in the complete cleanup-removal suite pass in 8.31 seconds. +Two separate fresh PostgreSQL 17 fixtures migrate through 0410. These suites +use supported external URLs with no source adaptation. The other 1779 cases +in the large chat file are outside the focused selection and are not claimed. +This does not claim the parent's full 274-case workflow was rerun on this slice. +Publication remains held for the actual prerequisite merge and fresh CI/reviews. + + +## Actual automatic-review squash ancestry + +Parent #17680 landed as `73d24aa5acfcf17b3f676b4da7cd05eb70b60b99`. Its whole tree +equals reviewed parent `181bf800a79449961e09c995f2785dd7f8060bb3`, already present +in this child history. Normal merge `6eb13207800b85d00e74a9803278d52e7a9fd890` +resolves inspected duplicate squash-ancestry conflicts and preserves the entire +accepted child tree. Complete server, package, UI and test trees remain equal +to the accepted context source, including the entire 0410 database source. +All original upgrade and suite evidence retains its stated scope. 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 497b7a7d87c..27dc4736d5a 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -559,6 +559,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0407_greedy_star_brand.sql", "0408_short_red_skull.sql", "0409_uneven_zuras.sql", + "0410_freezing_kinsey_walden.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0410_freezing_kinsey_walden.sql b/packages/db/src/migrations/0410_freezing_kinsey_walden.sql new file mode 100644 index 00000000000..63ae03549e6 --- /dev/null +++ b/packages/db/src/migrations/0410_freezing_kinsey_walden.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "native_run_finalizations_unsafe_export_recovery_idx" ON "native_run_finalizations" USING btree ("company_id","issue_id","run_id") WHERE "native_run_finalizations"."phase" = 'terminal_failure' AND "native_run_finalizations"."failure_code" = 'native_workspace_sync_out_unsafe_archive'; diff --git a/packages/db/src/migrations/meta/0405_snapshot.json b/packages/db/src/migrations/meta/0410_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0405_snapshot.json rename to packages/db/src/migrations/meta/0410_snapshot.json index 28bd125ce11..4fb46f14289 100644 --- a/packages/db/src/migrations/meta/0405_snapshot.json +++ b/packages/db/src/migrations/meta/0410_snapshot.json @@ -1,6 +1,6 @@ { - "id": "c87ad360-f306-459b-b814-fde37327c6da", - "prevId": "d1e8fe9b-c8c9-46ed-b8e0-55bada43eeed", + "id": "c02426e4-de1a-4ce3-a541-99bb970c272a", + "prevId": "94acf3fe-50f3-4a97-a7fe-865abbc9d0a3", "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": [ @@ -43026,6 +43108,34 @@ } }, "indexes": { + "native_run_finalizations_unsafe_export_recovery_idx": { + "name": "native_run_finalizations_unsafe_export_recovery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"phase\" = 'terminal_failure' AND \"native_run_finalizations\".\"failure_code\" = 'native_workspace_sync_out_unsafe_archive'", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 469168c4ed3..9647f3b918e 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2857,6 +2857,13 @@ "when": 1793243052342, "tag": "0409_uneven_zuras", "breakpoints": true + }, + { + "idx": 410, + "version": "7", + "when": 1793244017955, + "tag": "0410_freezing_kinsey_walden", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/native_run_finalizations.ts b/packages/db/src/schema/native_run_finalizations.ts index 5fb0718208a..874c27393dc 100644 --- a/packages/db/src/schema/native_run_finalizations.ts +++ b/packages/db/src/schema/native_run_finalizations.ts @@ -50,6 +50,9 @@ export const nativeRunFinalizations = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + unsafeExportRecoveryIdx: index("native_run_finalizations_unsafe_export_recovery_idx") + .on(table.companyId, table.issueId, table.runId) + .where(sql`${table.phase} = 'terminal_failure' AND ${table.failureCode} = 'native_workspace_sync_out_unsafe_archive'`), controlDeadlineIdx: index("native_run_finalizations_control_deadline_idx") .on(table.controlDeadlineAt).where(sql`${table.controlDeadlineAt} is not null`), issueCompanyFk: foreignKey({ diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 3ed2a53a96c..95afd9989f9 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() })), + registerUnsafeWorkspaceExportWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerAutomaticCompletionReviewWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerObsoletePolicyReviewWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerCancelledWorkspacePreservationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -556,6 +557,13 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers unsafe export recovery only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerUnsafeWorkspaceExportWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerUnsafeWorkspaceExportWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers automatic completion review cleanup only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -989,7 +997,7 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 788bf90b1f9..ea4cee55de3 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.registerUnsafeWorkspaceExportWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerAutomaticCompletionReviewWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerObsoletePolicyReviewWork(deliveryWork, @@ -1960,7 +1962,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. runPeriodicRecovery(() => heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: 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 2d3d0b2f831..42267449092 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 { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; 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"; @@ -341,6 +342,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 }); + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId: input.sourceIssue.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 a0b59f04419..8531240df68 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; 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"; @@ -717,6 +718,7 @@ export function createPostgresRunDispatchAdapter( 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 notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); const [row] = await tx .update(heartbeatRuns) @@ -926,6 +928,7 @@ export function createPostgresRunDispatchAdapter( 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 notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId }); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [row] = await tx .update(heartbeatRuns) @@ -1097,6 +1100,7 @@ export function createPostgresRunDispatchAdapter( const scheduledRetryAt = new Date(now.getTime() + 30_000); const executionWait = { issueId, runId: blocker.runId, cause: blocker.cause }; await notifyScheduledRetryCancellation(tx, { companyId: run.companyId, runId: run.id }); + if (run.status === "queued") await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId }); const [waiting] = await tx.update(heartbeatRuns) .set({ status: "scheduled_retry", diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 078a3d04aa0..7b3b45f1a39 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; @@ -1121,6 +1122,11 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd ) .orderBy(asc(issues.id)); + for (const issue of candidateIssues) { + if (issue.executionRunId === run.id || issue.checkoutRunId === run.id) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: issue.id }); + } + } // Two separate updates: a retry can move `executionRunId` to a new run // while `checkoutRunId` still points at this one finishing. 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 5dcd57c3485..455c2109050 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 { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; 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"; @@ -105,6 +106,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async cancelQueueRun({ queueRunId, reason, now }) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); await notifyInstructionCollection(tx, { companyId, runId: queueRunId }); await notifyCancelledWorkspacePreservation(tx, { companyId, runId: queueRunId }); await notifyChatRunMilestones(tx, { companyId, issueId }); @@ -118,6 +120,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async clearExecutionLockAndTouchIssue({ issueId, executionRunId, updatedAt }) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); await tx .update(issues) .set({ executionRunId: null, executionAgentNameKey: null, executionLockedAt: null, updatedAt }) diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 65d44046d24..74241e50965 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../services/unsafe-workspace-export-notifications.js"; import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js"; import { completeIssueCommentReceipt } from "../services/issue-comment-receipt.js"; import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; @@ -9881,6 +9882,7 @@ export function issueRoutes( } await markExecutionReconciliation(tx as unknown as Db, workspaceRepairAction, executionReconciliation!, actor.actorId, undefined, { workspaceRepairOnly: true }); + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: lockedIssue.companyId, issueId: lockedIssue.id }); const [recoveryAction] = await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "restored", resolutionNote: resolutionNote ?? null, resolvedAt: new Date(), updatedAt: new Date(), }).where(eq(issueRecoveryActions.id, workspaceRepairAction.id)).returning(); @@ -9929,6 +9931,7 @@ export function issueRoutes( "Verified outcomes must restore this source recovery without replacing another active recovery action.", ); } + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: lockedIssue.companyId, issueId: lockedIssue.id }); const [reopened] = await tx .update(issueRecoveryActions) .set({ status: "active", outcome: null, resolvedAt: null }) diff --git a/server/src/services/access.ts b/server/src/services/access.ts index 42af549b480..c76608597cc 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js"; import { notifyChatEndpointWork } from "./chat-work-notifications.js"; @@ -751,6 +752,7 @@ export function accessService(db: Db) { }) .where(and(assignedOpenIssueWhere, eq(issues.status, "in_progress"))) .returning({ id: issues.id }); + if (assignmentPatch.assigneeAgentId) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId }); const reassigned = await tx .update(issues) .set(assignmentPatch) diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index 5b32bdd6f45..4cafdf507b1 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared"; import { persistActivity, @@ -385,6 +386,7 @@ export async function settleConversationTurn( issue.conversationState === conversationState ) return true; + if (status === "in_review") await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); await tx .update(issues) .set({ diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index 1f207367051..b78b3c5af4e 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; @@ -52,6 +53,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str await notifyDirectoryCleanup(tx, { companyId: companyId }); await notifyInstructionCollection(tx, { companyId: companyId }); await notifyCancelledWorkspacePreservation(tx, { companyId: companyId }); + await notifyUnsafeWorkspaceExportRecovery(tx, { 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/company-deletion.ts b/server/src/services/company-deletion.ts index e7a6249e9f2..f4e020bd3da 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; @@ -106,6 +107,7 @@ export async function deleteCompany(db: Db, id: string) { await notifyDirectoryCleanup(tx, { companyId: id }); await notifyInstructionCollection(tx, { companyId: id }); await notifyCancelledWorkspacePreservation(tx, { companyId: id }); + await notifyUnsafeWorkspaceExportRecovery(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 383a917a571..50152f2c127 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 = { + unsafeWorkspaceExportRecovery: "unsafe-workspace-export-recovery", automaticCompletionReviews: "automatic-completion-reviews", obsoletePolicyReviews: "obsolete-policy-reviews", cancelledWorkspacePreservation: "cancelled-workspace-preservation", diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index da1c2360f2b..b423f8205d8 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRunChange } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; @@ -187,6 +188,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 notifyUnsafeWorkspaceExportRunChange(tx, run); await notifyExecutionRunWork(tx); const [updatedRun] = await tx .update(heartbeatRuns) diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index e30808f1025..2d190156d6c 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; @@ -450,6 +451,7 @@ export async function settleUnrecoverableExecutions( .limit(25).for("update", { skipLocked: true }); for (const candidate of foldable) { if (await getConversationOwnershipBlocker(tx as unknown as Db, candidate.companyId, candidate.sourceIssueId)) continue; + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: candidate.companyId, issueId: candidate.sourceIssueId }); const [action] = await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "cancelled", @@ -585,6 +587,7 @@ export async function settleUnrecoverableExecutions( ? "Workspace repair required. Verify safe staging or repair before continuing. Saved work and approval decisions remain in force." : "Automatic recovery stopped. Recorded work is preserved; actions with unverified outcomes will not be repeated." : "Recovery closed because the task's owner, execution, or status changed. No work was replayed."; + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: task.companyId, issueId: task.id }); let nativeFailureBlock = action.evidence.nativeFailureBlock; let nativeBootstrapFailureBlock = action.evidence.nativeBootstrapFailureBlock; if (current) { diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index fce18d379f2..074ffc3bdf5 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { assertDurableChatWakeupRequest, assertDurableChatWakeupReceipt, type DurableChatWakeupRequest } from "./durable-chat-wakeup.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { canSelectFreshWorkspace, type ContinuationReadiness } from "./fresh-workspace-continuation.js"; @@ -490,8 +491,10 @@ export async function admitExplicitNativeContinuation(input: { failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "explicit_user_continuation" })}::jsonb`, updatedAt: new Date(), }).where(and(eq(nativeRunFinalizations.companyId, companyId), inArray(nativeRunFinalizations.runId, nativeSources.map(run => run.id)))); + if (actions.length) await db.transaction(async tx => { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); for (const action of actions) { - await db.update(issueRecoveryActions).set({ + await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "cancelled", resolvedAt: new Date(), updatedAt: new Date(), nextAction: "The user started a fresh conversation turn. Prior action outcomes remain recorded.", resolutionNote: "The user continued after the prior execution stopped. No action outcomes were inferred.", @@ -507,6 +510,7 @@ export async function admitExplicitNativeContinuation(input: { }, }).where(eq(issueRecoveryActions.id, action.id)); } + }); await persistActivity(db, { companyId, actorType: "user", actorId, action: "issue.execution_recovery_settled", entityType: "issue", entityId: issueId, details: { continuation: retry ? "explicit_user_retry" : "explicit_user_message", ...authorization, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c3fc549fef2..1bcecebabd9 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerUnsafeWorkspaceExportWork } from "./native-runtime/native-workspace-export-recovery.js"; import { registerAutomaticCompletionReviewWork } from "./native-runtime/automatic-completion-reviews.js"; import { registerObsoletePolicyReviewWork } from "./native-runtime/obsolete-policy-reviews.js"; import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js"; @@ -4087,6 +4088,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerUnsafeWorkspaceExportWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerUnsafeWorkspaceExportWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerAutomaticCompletionReviewWork: (coordinator: Parameters[1], canRun: () => boolean) => registerAutomaticCompletionReviewWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerObsoletePolicyReviewWork: (coordinator: Parameters[1], canRun: () => boolean) => diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index ecd913d0d16..4d93f0fade8 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery, notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; @@ -854,6 +855,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende 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 }); + await notifyUnsafeWorkspaceExportRunChange(tx, current); [terminal] = await tx .update(heartbeatRuns) .set({ @@ -977,6 +979,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyChatRunMilestones(tx, { companyId: run.companyId }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); + await notifyUnsafeWorkspaceExportRunChange(tx, run); 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(); @@ -1339,6 +1342,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende 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 }); + await notifyUnsafeWorkspaceExportRunChange(tx, run); const [cancelled] = await tx .update(heartbeatRuns) .set({ @@ -1462,6 +1466,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende 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 }); + await notifyUnsafeWorkspaceExportRunChange(tx, run); const [cancelled] = await tx .update(heartbeatRuns) .set({ @@ -1795,7 +1800,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende error: reason, }); - await db + await db.transaction(async tx => { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId }); + await tx .update(issues) .set({ executionRunId: null, @@ -1810,6 +1817,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende eq(issues.executionRunId, run.id), ), ); + }); await appendRunEvent(cancelled, { eventType: "lifecycle", @@ -3501,6 +3509,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende : "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 notifyUnsafeWorkspaceExportRunChange(tx, scheduledRun); await notifyScheduledRetryCancellation(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); await notifyChatRunMilestones(tx, { companyId: scheduledRun.companyId, issueId: issue.id }); const cancelled = await tx @@ -3641,6 +3650,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ) { await notifyInstructionCollection(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id }); + await notifyUnsafeWorkspaceExportRunChange(tx, activeExecutionRun); await notifyChatRunMilestones(tx, { companyId: activeExecutionRun.companyId, issueId: issue.id }); const cancelled = await tx .update(heartbeatRuns) @@ -3683,6 +3693,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } if (!activeExecutionRun && issue.executionRunId) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); await tx .update(issues) .set({ @@ -3880,6 +3891,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende `- Reason: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_MESSAGE}`, `- Next action: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION}`, ].join("\n"); + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id }); await tx .update(issues) diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index bca38c1583e..335532eafa1 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1947,7 +1947,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1967,6 +1967,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD await reconcileNativeFinalizations(db, undefined, { skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews, skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews, + skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/heartbeat/run-lifecycle.test.ts b/server/src/services/heartbeat/run-lifecycle.test.ts index a8863e7c8f7..1360bafe0c3 100644 --- a/server/src/services/heartbeat/run-lifecycle.test.ts +++ b/server/src/services/heartbeat/run-lifecycle.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { randomBytes } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; @@ -107,14 +108,15 @@ describe("heartbeat lifecycle module boundary", () => { }); }); -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_LIFECYCLE_TEST_DATABASE_URL"); +const support = external ? { supported: true as const } : await getEmbeddedPostgresTestSupport(); if (!support.supported) console.warn(`Skipping lifecycle module database tests: ${support.reason}`); describe.skipIf(!support.supported)("heartbeat lifecycle database wiring", () => { let database: Awaited>; let db: Db; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-lifecycle-module"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-lifecycle-module"); + db = createDb(external ?? database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end(); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index da93922cd26..00f664d4769 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; @@ -1016,13 +1017,14 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl return recordHeartbeatRunRuntimeProgress(currentRun, update, issueId); } - async function writeRunStatus(status: string, previous: Pick | null, + async function writeRunStatus(status: string, previous: Pick | null, write: (tx: Db) => Promise) { const companyId = previous?.companyId; // Register on the outer transaction before the terminal write. An unknown // commit cannot strand newly-actionable accounting after the live event. return db.transaction(async transaction => { const tx = transaction as unknown as Db; + if (previous && previous.status !== status) await notifyUnsafeWorkspaceExportRunChange(tx, previous); if (previous && (previous.status === "scheduled_retry" || status === "scheduled_retry")) await notifyScheduledRetryCancellation(tx, { companyId: previous.companyId, runId: previous.id }); if (previous) await notifyInstructionCollection(tx, { companyId: previous.companyId, runId: previous.id }); diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 04e2c19a6d1..93562d691df 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { writeExecutionControl } from "./execution-control-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { and, desc, eq, inArray, isNull, isNotNull, or, sql } from "drizzle-orm"; @@ -322,6 +323,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { const created = await db.transaction(async (tx) => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition); await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation); + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); const [superseded] = await tx .update(issueRecoveryActions) .set({ @@ -408,120 +410,132 @@ export function issueRecoveryActionService(db: DbOrTransaction) { existing.attemptCount, Math.min(nextAttemptCount, effectiveMaxAttempts), ); - const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx + const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + if ((!input.preserveExistingOwner && existing.cause !== input.cause) + || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); + } + return tx + .update(issueRecoveryActions) + .set({ + status: "escalated", + ownerType: existing.ownerType === "system" || input.ownerType === "system" ? "system" : "board", + ownerAgentId: null, + ownerUserId: null, + previousOwnerAgentId: + existing.ownerAgentId ?? existing.previousOwnerAgentId, + returnOwnerAgentId: + input.returnOwnerAgentId ?? + existing.returnOwnerAgentId ?? + existing.ownerAgentId, + evidence: { + ...(existing.evidence ?? {}), + ...(input.evidence ?? {}), + recoveryBudget: { + state: "exhausted", + attemptsUsed, + maxAttempts: effectiveMaxAttempts, + exhaustedAt: now.toISOString(), + cause: existing.cause, + fingerprint: existing.fingerprint, + }, + }, + nextAction: existing.cause === "native_workspace_finalization_owner_unverified" + ? input.nextAction + : `Automatic recovery exhausted after ${attemptsUsed}/${effectiveMaxAttempts} attempts. ` + + "Review the infrastructure failure and explicitly choose a replacement run or provider configuration.", + wakePolicy: null, + monitorPolicy: null, + attemptCount: attemptsUsed, + maxAttempts: effectiveMaxAttempts, + timeoutAt: null, + lastAttemptAt: input.lastAttemptAt ?? now, + outcome: "escalated", + resolutionNote: null, + resolvedAt: null, + updatedAt: now, + }) + .where( + and( + eq(issueRecoveryActions.id, existing.id), + inArray(issueRecoveryActions.status, [ + ...ACTIVE_RECOVERY_ACTION_STATUSES, + ]), + ), + ) + .returning(); + }); + if (!exhausted) { + return retryUpsertSourceScoped(input, retryCount); + } + return toReadModel(exhausted); + } + const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + if ((!input.preserveExistingOwner && existing.cause !== input.cause) + || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); + } + return tx .update(issueRecoveryActions) .set({ - status: "escalated", - ownerType: existing.ownerType === "system" || input.ownerType === "system" ? "system" : "board", - ownerAgentId: null, - ownerUserId: null, - previousOwnerAgentId: - existing.ownerAgentId ?? existing.previousOwnerAgentId, - returnOwnerAgentId: - input.returnOwnerAgentId ?? - existing.returnOwnerAgentId ?? - existing.ownerAgentId, - evidence: { - ...(existing.evidence ?? {}), - ...(input.evidence ?? {}), - recoveryBudget: { - state: "exhausted", - attemptsUsed, - maxAttempts: effectiveMaxAttempts, - exhaustedAt: now.toISOString(), - cause: existing.cause, - fingerprint: existing.fingerprint, - }, - }, - nextAction: existing.cause === "native_workspace_finalization_owner_unverified" - ? input.nextAction - : `Automatic recovery exhausted after ${attemptsUsed}/${effectiveMaxAttempts} attempts. ` + - "Review the infrastructure failure and explicitly choose a replacement run or provider configuration.", - wakePolicy: null, - monitorPolicy: null, - attemptCount: attemptsUsed, - maxAttempts: effectiveMaxAttempts, - timeoutAt: null, - lastAttemptAt: input.lastAttemptAt ?? now, - outcome: "escalated", - resolutionNote: null, + recoveryIssueId: input.preserveExistingOwner + ? existing.recoveryIssueId + : input.recoveryIssueId ?? null, + kind: input.preserveExistingOwner ? existing.kind : input.kind, + status: input.preserveExistingOwner ? existing.status : "active", + ownerType: input.preserveExistingOwner ? existing.ownerType : ownerType, + ownerAgentId: input.preserveExistingOwner + ? existing.ownerAgentId + : input.ownerAgentId ?? null, + ownerUserId: input.preserveExistingOwner + ? existing.ownerUserId + : input.ownerUserId ?? null, + previousOwnerAgentId: input.preserveExistingOwner + ? existing.previousOwnerAgentId + : input.previousOwnerAgentId ?? existing.previousOwnerAgentId, + returnOwnerAgentId: input.preserveExistingOwner + ? existing.returnOwnerAgentId + : input.returnOwnerAgentId ?? existing.returnOwnerAgentId, + cause: input.preserveExistingOwner ? existing.cause : input.cause, + fingerprint: input.preserveExistingOwner ? existing.fingerprint : input.fingerprint, + evidence: input.preserveExistingOwner + ? { + ...(existing.evidence ?? {}), + ...(input.evidence ?? {}), + } + : input.evidence ?? existing.evidence, + nextAction: input.preserveExistingOwner ? existing.nextAction : input.nextAction, + wakePolicy: input.preserveExistingOwner + ? existing.wakePolicy + : input.wakePolicy ?? null, + monitorPolicy: input.preserveExistingOwner + ? existing.monitorPolicy + : input.monitorPolicy ?? null, + attemptCount: nextAttemptCount, + maxAttempts: input.preserveExistingOwner + ? existing.maxAttempts + : input.maxAttempts === undefined + ? existing.maxAttempts + : input.maxAttempts, + timeoutAt: input.preserveExistingOwner + ? asDatabaseDate(existing.timeoutAt) + : input.timeoutAt ?? null, + lastAttemptAt: input.preserveExistingOwner + ? asDatabaseDate(existing.lastAttemptAt) + : input.lastAttemptAt ?? now, + outcome: input.preserveExistingOwner ? existing.outcome : null, + resolutionNote: input.preserveExistingOwner ? existing.resolutionNote : null, resolvedAt: null, updatedAt: now, }) .where( and( eq(issueRecoveryActions.id, existing.id), - inArray(issueRecoveryActions.status, [ - ...ACTIVE_RECOVERY_ACTION_STATUSES, - ]), + inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]), ), ) - .returning()); - if (!exhausted) { - return retryUpsertSourceScoped(input, retryCount); - } - return toReadModel(exhausted); - } - const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx - .update(issueRecoveryActions) - .set({ - recoveryIssueId: input.preserveExistingOwner - ? existing.recoveryIssueId - : input.recoveryIssueId ?? null, - kind: input.preserveExistingOwner ? existing.kind : input.kind, - status: input.preserveExistingOwner ? existing.status : "active", - ownerType: input.preserveExistingOwner ? existing.ownerType : ownerType, - ownerAgentId: input.preserveExistingOwner - ? existing.ownerAgentId - : input.ownerAgentId ?? null, - ownerUserId: input.preserveExistingOwner - ? existing.ownerUserId - : input.ownerUserId ?? null, - previousOwnerAgentId: input.preserveExistingOwner - ? existing.previousOwnerAgentId - : input.previousOwnerAgentId ?? existing.previousOwnerAgentId, - returnOwnerAgentId: input.preserveExistingOwner - ? existing.returnOwnerAgentId - : input.returnOwnerAgentId ?? existing.returnOwnerAgentId, - cause: input.preserveExistingOwner ? existing.cause : input.cause, - fingerprint: input.preserveExistingOwner ? existing.fingerprint : input.fingerprint, - evidence: input.preserveExistingOwner - ? { - ...(existing.evidence ?? {}), - ...(input.evidence ?? {}), - } - : input.evidence ?? existing.evidence, - nextAction: input.preserveExistingOwner ? existing.nextAction : input.nextAction, - wakePolicy: input.preserveExistingOwner - ? existing.wakePolicy - : input.wakePolicy ?? null, - monitorPolicy: input.preserveExistingOwner - ? existing.monitorPolicy - : input.monitorPolicy ?? null, - attemptCount: nextAttemptCount, - maxAttempts: input.preserveExistingOwner - ? existing.maxAttempts - : input.maxAttempts === undefined - ? existing.maxAttempts - : input.maxAttempts, - timeoutAt: input.preserveExistingOwner - ? asDatabaseDate(existing.timeoutAt) - : input.timeoutAt ?? null, - lastAttemptAt: input.preserveExistingOwner - ? asDatabaseDate(existing.lastAttemptAt) - : input.lastAttemptAt ?? now, - outcome: input.preserveExistingOwner ? existing.outcome : null, - resolutionNote: input.preserveExistingOwner ? existing.resolutionNote : null, - resolvedAt: null, - updatedAt: now, - }) - .where( - and( - eq(issueRecoveryActions.id, existing.id), - inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]), - ), - ) - .returning()); + .returning(); + }); if (!updated) { return retryUpsertSourceScoped(input, retryCount); } @@ -529,10 +543,13 @@ export function issueRecoveryActionService(db: DbOrTransaction) { } try { - const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx - .insert(issueRecoveryActions) - .values(buildInsertValues(input, ownerType, now)) - .returning()); + const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); + return tx + .insert(issueRecoveryActions) + .values(buildInsertValues(input, ownerType, now)) + .returning(); + }); return toReadModel(created!); } catch (error) { if (!isUniqueRecoveryActionConflict(error)) throw error; @@ -569,17 +586,23 @@ export function issueRecoveryActionService(db: DbOrTransaction) { predicates.push(eq(issueRecoveryActions.fingerprint, input.fingerprint)); } - const [updated] = await writeExecutionControl(dbOrTx, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx - .update(issueRecoveryActions) - .set({ - status: input.status, - outcome: input.outcome, - resolutionNote: input.resolutionNote ?? null, - resolvedAt: now, - updatedAt: now, - }) - .where(and(...predicates)) - .returning()); + const [updated] = await writeExecutionControl(dbOrTx, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + const pending = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions) + .where(and(...predicates)).limit(1); + if (!pending.length) return []; + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); + return tx + .update(issueRecoveryActions) + .set({ + status: input.status, + outcome: input.outcome, + resolutionNote: input.resolutionNote ?? null, + resolvedAt: now, + updatedAt: now, + }) + .where(and(...predicates)) + .returning(); + }); return updated ? toReadModel(updated) : null; } diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 0ac70e83077..5697e19449c 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; @@ -1003,6 +1004,7 @@ export function issueTreeControlService(db: Db) { for (const issueId of issueIdsForStatus) { await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); await notifyChatResponseAvailability(tx, { companyId, issueId }); + if (status === "blocked" || status === "in_review") await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); } const rows = await tx .update(issues) diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index a6a406a6b3e..04387d35298 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; @@ -7821,6 +7822,7 @@ export function issueService(db: Db) { if (run && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(run.status)) return false; if (isExplicitContinuationRetryClaim(issue, run)) return false; + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId }); const updated = await tx .update(issues) .set({ @@ -7894,6 +7896,7 @@ export function issueService(db: Db) { if (isExplicitContinuationRetryClaim(issue, executionRun)) return false; } + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId }); const updated = await tx .update(issues) .set({ @@ -11462,6 +11465,12 @@ export function issueService(db: Db) { if (patch.hiddenAt !== undefined || patch.harnessKind !== undefined || patch.conversationAgentId !== undefined) { await notifyApprovalWork(tx, receiptExisting.companyId); } + if ((patch.status !== undefined && patch.status !== receiptExisting.status) + || (patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId) + || (patch.executionRunId !== undefined && patch.executionRunId !== receiptExisting.executionRunId) + || (patch.checkoutRunId !== undefined && patch.checkoutRunId !== receiptExisting.checkoutRunId)) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: receiptExisting.companyId, issueId: id }); + } const updated = await tx .update(issues) .set(patch) @@ -12402,6 +12411,7 @@ export function issueService(db: Db) { const existing = await tx .select({ id: issues.id, + companyId: issues.companyId, checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId, }) @@ -12422,6 +12432,9 @@ export function issueService(db: Db) { } await notifyTerminalWorkspaceWork(tx, `issue:${id}`); + if (existing.checkoutRunId || existing.executionRunId) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id }); + } const updated = await tx .update(issues) .set(patch) diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 8252a2ecd56..22461cf38b3 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { hasRequiredWorkspaceRecovery, preserveWorkspaceRestoreRecoveryMetadata, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js"; @@ -131,6 +132,7 @@ export async function terminalizeLegacyExecution(input: { resultJson: preserveWorkspaceRestoreRecoveryMetadata(current.resultJson, input.recordRestoreFailureOnly ? { ...current.resultJson, ...patch.resultJson } : patch.resultJson), }; + if (issueId) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: current.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id }); await notifyExecutionRunWork(tx); let [updated] = await tx @@ -347,6 +349,7 @@ export async function settleInterruptedNativeBootstrap( const [settled] = await tx.update(heartbeatRuns).set({ resultJson: { ...run.resultJson, executionRecovery: { kind: "bootstrap", providerWorkStarted: false, preparationSettledAt: now.toISOString() }, }, updatedAt: now }).where(eq(heartbeatRuns.id, run.id)).returning(); + if (holds.length) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: task.companyId, issueId: task.id }); const resolved = holds.length ? await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "false_positive", resolutionNote: "The interrupted executor finished preparation and cleanup without starting a provider.", resolvedAt: now, updatedAt: now, diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 93959ef1e4a..bfab2c0025a 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 { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { readRequiredQualityReview } from "../required-quality-review.js"; @@ -519,6 +520,7 @@ export async function claimNativeSessionResumptions(input: { }).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 }); + await notifyUnsafeWorkspaceExportRunChange(tx, row.run); const [updatedRun] = await tx.update(heartbeatRuns).set({ status: "failed", finishedAt: now, @@ -639,6 +641,7 @@ export async function claimNativeSessionResumptions(input: { type NativeReconciliationOptions = { skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; + skipUnsafeWorkspaceExports?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -721,10 +724,12 @@ async function reconcileNativeFinalizationsPass( options: NativeReconciliationOptions, timing: NativeReconciliationPassTiming, ) { - timing.phase("legacy_workspace_exports"); - await recoverLegacyUnsafeWorkspaceExports(db, runIds).catch((err) => { - logger.warn({ err }, "Historical unsafe export recovery remains pending"); - }); + if (!options.skipUnsafeWorkspaceExports) { + timing.phase("legacy_workspace_exports"); + await recoverLegacyUnsafeWorkspaceExports(db, runIds).catch((err) => { + logger.warn({ err }, "Historical unsafe export recovery remains pending"); + }); + } if (!options.skipObsoletePolicyReviews) { timing.phase("obsolete_policy_reviews"); await dismissObsoleteNativePolicyReviews(db, runIds).catch((err) => { diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index 0320564677e..abe854fad6b 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import type { Db } from "@paperclipai/db"; import type { NativeInteractionResponseEnvelope } from "../../vendor/paperclip-runner/index.js"; import { and, asc, eq, inArray } from "drizzle-orm"; @@ -120,6 +121,9 @@ export async function applyNativeAttentionStatusDecision(input: { continue; } if (effect.kind === "record_finalization_error") { + if (effect.cause === "native_workspace_sync_out_unsafe_archive") { + await notifyUnsafeWorkspaceExportRecovery(tx, input); + } const [coordinator] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: effect.cause, diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 3c34f8b7b58..0b7c8bf41ed 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 { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; @@ -923,6 +924,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 }); + await notifyUnsafeWorkspaceExportRunChange(tx, row.run); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-run-coordinator-store.ts b/server/src/services/native-runtime/native-run-coordinator-store.ts index 34e45437d2f..de6955c3c44 100644 --- a/server/src/services/native-runtime/native-run-coordinator-store.ts +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { isSafeNativeChatProgressEventType } from "../safe-native-chat-progress.js"; import { notifyChatNativeProgress } from "../chat-native-progress-notifications.js"; import { createHash } from "node:crypto"; @@ -562,7 +563,11 @@ export class NativeRunCoordinatorStore { ) { throw new Error("native_finalization_retry_invalid"); } - const [updated] = await this.#db + const [updated] = await this.#db.transaction(async tx => { + if (input.failureCode === "native_workspace_sync_out_unsafe_archive") { + await notifyUnsafeWorkspaceExportRecovery(tx, this.#binding); + } + return tx .update(nativeRunFinalizations) .set({ leaseOwner: null, @@ -580,6 +585,7 @@ export class NativeRunCoordinatorStore { ), ) .returning({ runId: nativeRunFinalizations.runId }); + }); if (!updated) throw new Error("native_finalization_lease_lost"); } } diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 6647b68b8e2..ed1c10648ce 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,3 +1,5 @@ +import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; +import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { readRequiredQualityReview } from "../required-quality-review.js"; @@ -380,6 +382,9 @@ async function recordRetryableFailure(input: { : "native_finalization_retry_exhausted" : input.failureCode; if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); + if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); + } await notifyExecutionRunWork(tx); await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); await tx @@ -414,6 +419,7 @@ async function recordRetryableFailure(input: { } 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 }); + if (projectsTerminalStatus) await notifyUnsafeWorkspaceExportRunChange(tx, input.run); const [updatedRun] = await tx .update(heartbeatRuns) .set({ @@ -604,7 +610,7 @@ function recoveredExecutionFailureMetadata() { async function projectCommittedRun(input: { db: Db; - run: Pick; + run: Pick; coordinator: typeof nativeRunFinalizations.$inferSelect; }) { if (!input.coordinator.resultId) return; @@ -673,6 +679,7 @@ async function projectCommittedRun(input: { 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 notifyUnsafeWorkspaceExportRunChange(tx, input.run); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); await notifyExecutionRunWork(tx); await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); @@ -1164,6 +1171,8 @@ export async function finalizeNativeRun(input: { const admittedRun = await input.db .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, + nativeIssueId: heartbeatRuns.nativeIssueId, + contextSnapshot: sql>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->>'issueId')`, nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) @@ -1693,6 +1702,7 @@ export async function finalizeNativeRun(input: { 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 notifyUnsafeWorkspaceExportRunChange(tx, run); 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-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index 03109f56def..78b889b169a 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { logger } from "../../middleware/logger.js"; import { createHash, randomUUID } from "node:crypto"; @@ -466,6 +467,9 @@ export async function reconcileSafeNativeReplacements( eq(issueRecoveryActions.cause, current.failureCode!), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, )).for("update"); + if (holds.some(hold => hold.status === "active" || hold.status === "escalated")) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: task.id }); + } for (const hold of holds) { await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "handed_back", resolvedAt: now, updatedAt: now, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 467098f6138..7ec1a87225e 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 { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; 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"; @@ -9822,6 +9823,7 @@ async function executePaperclipNativeSessionWithinScope( .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 }); + if (!ownershipUnverified) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId }); let [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-workspace-export-recovery.ts b/server/src/services/native-runtime/native-workspace-export-recovery.ts index 1c97e2ea2e7..e64bc4037c4 100644 --- a/server/src/services/native-runtime/native-workspace-export-recovery.ts +++ b/server/src/services/native-runtime/native-workspace-export-recovery.ts @@ -1,4 +1,9 @@ -import { and, asc, desc, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; +import { alias } from "drizzle-orm/pg-core"; +import { notifyUnsafeWorkspaceExportRecovery, type UnsafeWorkspaceExportScope } from "../unsafe-workspace-export-notifications.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { and, asc, desc, eq, exists, gt, inArray, isNull, ne, notExists, or, sql } from "drizzle-orm"; import { completionContracts, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { publishLiveEvent } from "../live-events.js"; import { buildHeartbeatRunStatusLiveEventPayload } from "../heartbeat-run-status-payload.js"; @@ -25,87 +30,174 @@ export async function recoverLegacyUnsafeWorkspaceExports(db: Db, runIds?: strin } for (const candidate of candidates) { if (!runIds?.length) scanCursors.set(db, candidate.runId); - const owned = await withNativeWorkspaceFinalizationOwnership({ db, ...candidate }, async ownership => { - await ownership.assertHeld(); - return db.transaction(async tx => { - const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId))).for("update").limit(1); - const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( - eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, candidate.runId), - )).for("update").limit(1); - if (!coordinator || coordinator.phase !== "terminal_failure" || coordinator.failureCode !== UNSAFE_EXPORT - || !coordinator.resultId || coordinator.leaseOwner || coordinator.issueId !== issue?.id) return null; - const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, candidate.runId))).for("update").limit(1); - if (!issue || !run || run.runtimeMode !== "native" || run.nativeIssueId !== issue.id) return null; - const [result] = await tx.select().from(nativeRunResults).where(and( - eq(nativeRunResults.id, coordinator.resultId), eq(nativeRunResults.companyId, run.companyId), - eq(nativeRunResults.runId, run.id), eq(nativeRunResults.issueId, issue.id), - eq(nativeRunResults.completionContractId, run.completionContractId ?? "00000000-0000-0000-0000-000000000000"), - eq(nativeRunResults.schemaStatus, "accepted"), - )).limit(1); - if (!result) return null; - const priorActions = await tx.select().from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), - eq(issueRecoveryActions.cause, UNSAFE_EXPORT), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, - )); - const explicitlySettled = priorActions.some(action => action.status === "cancelled" - || (action.status === "resolved" && !["new_source_execution_path", "unsafe_export_automatic_recovery"].includes(action.resolutionNote ?? ""))); - const now = new Date(); - // A stale unsafe notice is never an instruction to repair or start a - // provider turn, including when newer work prevents replay of this run. - await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "restored", resolvedAt: now, - resolutionNote: "unsafe_export_automatic_recovery", nextAction: "", wakePolicy: null, updatedAt: now, - }).where(and(eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), - eq(issueRecoveryActions.cause, UNSAFE_EXPORT), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, - or(inArray(issueRecoveryActions.status, ["active", "escalated"]), - and(eq(issueRecoveryActions.status, "resolved"), eq(issueRecoveryActions.resolutionNote, "new_source_execution_path"))), - )); - const [newerRun] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id), - or(eq(heartbeatRuns.nativeIssueId, issue.id), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), - or(gt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, ["queued", "running"])), - )).limit(1); - const [otherAction] = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), - inArray(issueRecoveryActions.status, ["active", "escalated"]), - )).limit(1); - const [latestContract] = await tx.select({ id: completionContracts.id }).from(completionContracts).where(and( - eq(completionContracts.companyId, run.companyId), eq(completionContracts.issueId, issue.id), - )).orderBy(desc(completionContracts.revision)).limit(1); - const eligible = !explicitlySettled && !newerRun && !otherAction && latestContract?.id === result.completionContractId - && ["blocked", "in_review"].includes(issue.status) && issue.assigneeAgentId === run.agentId - && ["failed", "succeeded"].includes(run.status) && run.nativePhase === "terminal_failure" - && (!issue.executionRunId || issue.executionRunId === run.id) - && (!issue.checkoutRunId || issue.checkoutRunId === run.id); - let updatedRun: typeof heartbeatRuns.$inferSelect | undefined; - if (eligible) { - await ownership.assertHeld(); - // Publish the omission barrier atomically with re-admission. This is - // explicitly not a receipt claiming that files were copied to the host. - await tx.insert(workspaceOperations).values({ companyId: run.companyId, heartbeatRunId: run.id, issueId: issue.id, - phase: "workspace_finalize", status: "succeeded", exitCode: 0, finishedAt: now, - metadata: { owningService: "native_workspace_finalizer", workspaceSync: { omitted: true, reason: "restore_unsafe_archive", legacy: true } }, - }); - await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, - failureDetail: { workspaceFinalizeAttempt: 0, legacyUnsafeExportOmitted: { resultId: result.id, recoveredAt: now.toISOString() } }, updatedAt: now, - }).where(eq(nativeRunFinalizations.runId, run.id)); - [updatedRun] = await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null, - nativePhase: "result_accepted", nativePhaseUpdatedAt: now, - resultJson: { ...run.resultJson, finalizationPhase: "result_accepted", failureCode: null, originalFailureCode: null, nextAttemptAt: null }, updatedAt: now, - }).where(eq(heartbeatRuns.id, run.id)).returning(); - await tx.update(issues).set({ executionRunId: run.id, updatedAt: now }).where(eq(issues.id, issue.id)); - await appendHeartbeatRunEvent(tx as unknown as Db, { companyId: run.companyId, agentId: run.agentId, runId: run.id, - eventType: "workspace_export_omitted", stream: "system", level: "info", - message: "Historical unsafe workspace export omitted; continuing with the accepted result.", - payload: { reason: "restore_unsafe_archive", legacy: true }, - }); - } - // Normal finalization publishes task status. Unsafe-export diagnostics - // belong only in run logs, not activity history or task notifications. - return updatedRun ?? null; - }); - }); - if (owned.acquired && owned.value) publishLiveEvent({ companyId: owned.value.companyId, - type: "heartbeat.run.status", payload: buildHeartbeatRunStatusLiveEventPayload(owned.value), - }); + await recoverCandidate(db, candidate); } } + +async function recoverCandidate(db: Db, candidate: { runId: string; companyId: string; issueId: string }) { + const owned = await withNativeWorkspaceFinalizationOwnership({ db, ...candidate }, async ownership => { + await ownership.assertHeld(); + return db.transaction(async tx => { + const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId))).for("update").limit(1); + const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, candidate.runId), + )).for("update").limit(1); + if (!coordinator || coordinator.phase !== "terminal_failure" || coordinator.failureCode !== UNSAFE_EXPORT + || !coordinator.resultId || coordinator.leaseOwner || coordinator.issueId !== issue?.id) return null; + const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, candidate.runId))).for("update").limit(1); + if (!issue || !run || run.runtimeMode !== "native" || run.nativeIssueId !== issue.id) return null; + const [result] = await tx.select().from(nativeRunResults).where(and( + eq(nativeRunResults.id, coordinator.resultId), eq(nativeRunResults.companyId, run.companyId), + eq(nativeRunResults.runId, run.id), eq(nativeRunResults.issueId, issue.id), + eq(nativeRunResults.completionContractId, run.completionContractId ?? "00000000-0000-0000-0000-000000000000"), + eq(nativeRunResults.schemaStatus, "accepted"), + )).limit(1); + if (!result) return null; + const priorActions = await tx.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), + eq(issueRecoveryActions.cause, UNSAFE_EXPORT), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, + )); + const explicitlySettled = priorActions.some(action => action.status === "cancelled" + || (action.status === "resolved" && !["new_source_execution_path", "unsafe_export_automatic_recovery"].includes(action.resolutionNote ?? ""))); + const now = new Date(); + + + if (priorActions.some(action => ["active", "escalated"].includes(action.status) + || (action.status === "resolved" && action.resolutionNote === "new_source_execution_path"))) { + + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: issue.id }); + } + await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "restored", resolvedAt: now, + resolutionNote: "unsafe_export_automatic_recovery", nextAction: "", wakePolicy: null, updatedAt: now, + }).where(and(eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), + eq(issueRecoveryActions.cause, UNSAFE_EXPORT), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, + or(inArray(issueRecoveryActions.status, ["active", "escalated"]), + and(eq(issueRecoveryActions.status, "resolved"), eq(issueRecoveryActions.resolutionNote, "new_source_execution_path"))), + )); + const [newerRun] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id), + or(eq(heartbeatRuns.nativeIssueId, issue.id), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), + or(gt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, ["queued", "running"])), + )).limit(1); + const [otherAction] = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), + inArray(issueRecoveryActions.status, ["active", "escalated"]), + )).limit(1); + const [latestContract] = await tx.select({ id: completionContracts.id }).from(completionContracts).where(and( + eq(completionContracts.companyId, run.companyId), eq(completionContracts.issueId, issue.id), + )).orderBy(desc(completionContracts.revision)).limit(1); + const eligible = !explicitlySettled && !newerRun && !otherAction && latestContract?.id === result.completionContractId + && ["blocked", "in_review"].includes(issue.status) && issue.assigneeAgentId === run.agentId + && ["failed", "succeeded"].includes(run.status) && run.nativePhase === "terminal_failure" + && (!issue.executionRunId || issue.executionRunId === run.id) + && (!issue.checkoutRunId || issue.checkoutRunId === run.id); + let updatedRun: typeof heartbeatRuns.$inferSelect | undefined; + if (eligible) { + await ownership.assertHeld(); + + + await tx.insert(workspaceOperations).values({ companyId: run.companyId, heartbeatRunId: run.id, issueId: issue.id, + phase: "workspace_finalize", status: "succeeded", exitCode: 0, finishedAt: now, + metadata: { owningService: "native_workspace_finalizer", workspaceSync: { omitted: true, reason: "restore_unsafe_archive", legacy: true } }, + }); + await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, + failureDetail: { workspaceFinalizeAttempt: 0, legacyUnsafeExportOmitted: { resultId: result.id, recoveredAt: now.toISOString() } }, updatedAt: now, + }).where(eq(nativeRunFinalizations.runId, run.id)); + [updatedRun] = await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null, + nativePhase: "result_accepted", nativePhaseUpdatedAt: now, + resultJson: { ...run.resultJson, finalizationPhase: "result_accepted", failureCode: null, originalFailureCode: null, nextAttemptAt: null }, updatedAt: now, + }).where(eq(heartbeatRuns.id, run.id)).returning(); + await tx.update(issues).set({ executionRunId: run.id, updatedAt: now }).where(eq(issues.id, issue.id)); + await appendHeartbeatRunEvent(tx as unknown as Db, { companyId: run.companyId, agentId: run.agentId, runId: run.id, + eventType: "workspace_export_omitted", stream: "system", level: "info", + message: "Historical unsafe workspace export omitted; continuing with the accepted result.", + payload: { reason: "restore_unsafe_archive", legacy: true }, + }); + } + + + return updatedRun ?? null; + }); + }); + if (owned.acquired && owned.value) publishLiveEvent({ companyId: owned.value.companyId, + type: "heartbeat.run.status", payload: buildHeartbeatRunStatusLiveEventPayload(owned.value), + }); +} + +function findOwedUnsafeExports(db: Db, scope?: UnsafeWorkspaceExportScope, after?: string, runId?: string) { + const action = alias(issueRecoveryActions, "unsafe_export_action"); + const otherRun = alias(heartbeatRuns, "unsafe_export_other_run"); + const contract = alias(completionContracts, "unsafe_export_contract"); + const sameActionRun = and(eq(action.companyId, nativeRunFinalizations.companyId), + eq(action.sourceIssueId, nativeRunFinalizations.issueId), eq(action.cause, UNSAFE_EXPORT), + sql`${action.evidence}->>'runId' = ${nativeRunFinalizations.runId}::text`); + const staleAction = db.select({ id: action.id }).from(action).where(and(sameActionRun, + or(inArray(action.status, ["active", "escalated"]), + and(eq(action.status, "resolved"), eq(action.resolutionNote, "new_source_execution_path"))))); + const explicitlySettled = db.select({ id: action.id }).from(action).where(and(sameActionRun, + or(eq(action.status, "cancelled"), and(eq(action.status, "resolved"), + sql`coalesce(${action.resolutionNote}, '') not in ('new_source_execution_path', 'unsafe_export_automatic_recovery')`)))); + const competingAction = db.select({ id: action.id }).from(action).where(and( + eq(action.companyId, nativeRunFinalizations.companyId), eq(action.sourceIssueId, nativeRunFinalizations.issueId), + inArray(action.status, ["active", "escalated"]), + or(ne(action.cause, UNSAFE_EXPORT), sql`${action.evidence}->>'runId' is distinct from ${nativeRunFinalizations.runId}::text`))); + const newerRun = db.select({ id: otherRun.id }).from(otherRun).where(and( + eq(otherRun.companyId, heartbeatRuns.companyId), ne(otherRun.id, heartbeatRuns.id), + or(eq(otherRun.nativeIssueId, issues.id), sql`${otherRun.contextSnapshot}->>'issueId' = ${issues.id}::text`), + or(gt(otherRun.createdAt, heartbeatRuns.createdAt), inArray(otherRun.status, ["queued", "running"])))); + const latestContract = db.select({ id: contract.id }).from(contract).where(and( + eq(contract.companyId, nativeRunFinalizations.companyId), eq(contract.issueId, nativeRunFinalizations.issueId), + )).orderBy(desc(contract.revision)).limit(1); + return db.select({ runId: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, issueId: nativeRunFinalizations.issueId }) + .from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), + eq(heartbeatRuns.id, nativeRunFinalizations.runId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(issues, and(eq(issues.companyId, nativeRunFinalizations.companyId), eq(issues.id, nativeRunFinalizations.issueId))) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.companyId, heartbeatRuns.companyId), eq(nativeRunResults.runId, heartbeatRuns.id), + eq(nativeRunResults.issueId, issues.id), eq(nativeRunResults.completionContractId, heartbeatRuns.completionContractId), + eq(nativeRunResults.schemaStatus, "accepted"))) + .where(and(eq(nativeRunFinalizations.phase, "terminal_failure"), eq(nativeRunFinalizations.failureCode, UNSAFE_EXPORT), + eq(heartbeatRuns.runtimeMode, "native"), + scope ? and(eq(nativeRunFinalizations.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined, + after ? gt(nativeRunFinalizations.runId, after) : undefined, + runId ? eq(nativeRunFinalizations.runId, runId) : undefined, + or(exists(staleAction), and( + notExists(explicitlySettled), notExists(competingAction), notExists(newerRun), + eq(nativeRunResults.completionContractId, latestContract), + inArray(issues.status, ["blocked", "in_review"]), eq(issues.assigneeAgentId, heartbeatRuns.agentId), + inArray(heartbeatRuns.status, ["failed", "succeeded"]), eq(heartbeatRuns.nativePhase, "terminal_failure"), + or(isNull(issues.executionRunId), eq(issues.executionRunId, heartbeatRuns.id)), + or(isNull(issues.checkoutRunId), eq(issues.checkoutRunId, heartbeatRuns.id)), + )), + )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); +} + +export function registerUnsafeWorkspaceExportWork(db: Db, coordinator: ReturnType, canRun: () => boolean) { + type Entry = UnsafeWorkspaceExportScope & { id: string; dueAt: number }; + async function load(scope?: UnsafeWorkspaceExportScope, signal?: AbortSignal): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await findOwedUnsafeExports(db, scope, after); + for (const row of rows) entries.push({ id: row.runId, companyId: row.companyId, issueId: row.issueId, dueAt: 0 }); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].runId; + } + throw new Error("Unsafe export recovery discovery stopped"); + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.unsafeWorkspaceExportRecovery, batchSize: 5, + decodeScope: (hint: string): UnsafeWorkspaceExportScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: UnsafeWorkspaceExportScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId), + load, + check: async (entry: Entry, signal?: AbortSignal) => { + const [candidate] = await findOwedUnsafeExports(db, entry, undefined, entry.id); + if (!canRun() || signal?.aborted) throw new Error("Unsafe export recovery stopped"); + if (!candidate) return null; + await recoverCandidate(db, candidate); + const pending = await findOwedUnsafeExports(db, entry, undefined, entry.id); + return pending.length ? { ...entry, dueAt: Date.now() + 30_000 } : null; + }, + }, canRun); +} diff --git a/server/src/services/native-runtime/native-workspace-export-retry.test.ts b/server/src/services/native-runtime/native-workspace-export-retry.test.ts index d8e664fba01..5b4ef55853c 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.test.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.test.ts @@ -1,29 +1,37 @@ +import { heartbeatService } from "../heartbeat.js"; +import { issueService } from "../issues.js"; +import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../task-admission.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; -import { beforeAll, afterAll, describe, expect, it, vi } from "vitest"; -import { eq } from "drizzle-orm"; +import { beforeAll, afterAll, beforeEach, afterEach, describe, expect, it, vi } from "vitest"; +import { and, eq } from "drizzle-orm"; import { agents, companies, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, agentWakeupRequests, activityLog } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { remoteTerminationReceipt } from "../remote-execution-termination.js"; import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js"; const probe = vi.hoisted(() => vi.fn()); vi.mock("../environment-execution-target.js", () => ({ resolveEnvironmentExecutionTarget: async () => ({ kind: "remote", transport: "sandbox", remoteCwd: "/work", runner: { execute: probe } }) })); -import { recoverLegacyUnsafeWorkspaceExports } from "./native-workspace-export-recovery.js"; +import { recoverLegacyUnsafeWorkspaceExports, registerUnsafeWorkspaceExportWork } from "./native-workspace-export-recovery.js"; import { recordNativeFinalizationFailure } from "./native-run-finalizer.js"; import { recoveryService } from "../recovery/service.js"; import { retryNativeWorkspaceExport } from "./native-workspace-export-retry.js"; import { releaseCompletedNativeWorkspaceExportRetention } from "./native-workspace-export-resume.js"; describe("board retry of accepted workspace export", () => { - let temporary: Awaited>; + let temporary: Awaited> | undefined; let db: ReturnType; const companyId = randomUUID(), agentId = randomUUID(), environmentId = randomUUID(); beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("workspace-export-retry-"); db = createDb(temporary.connectionString); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXPORT_RETRY_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("workspace-export-retry-"); + db = createDb(external ?? temporary!.connectionString); await db.insert(companies).values({ id: companyId, name: "Export repair", issuePrefix: "EXP" }); await db.insert(agents).values({ id: agentId, companyId, name: "Exporter", adapterType: "paperclip_runner" }); await db.insert(environments).values({ id: environmentId, name: `Retained ${environmentId}`, driver: "sandbox" }); }, 30_000); - afterAll(async () => { await temporary.cleanup(); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); async function seed(cause = "native_workspace_sync_out_retry_exhausted") { probe.mockReset().mockResolvedValue({ exitCode: 0, timedOut: false }); const issueId = randomUUID(), runId = randomUUID(), resultId = randomUUID(), contractId = randomUUID(), leaseId = randomUUID(), actionId = randomUUID(), providerLeaseId = randomUUID(); @@ -288,4 +296,263 @@ describe("board retry of accepted workspace export", () => { expect((f.request.environmentRuntime as { resumeRunLease: ReturnType }).resumeRunLease).not.toHaveBeenCalled(); }); + describe("unsafe-export scoped recovery", () => { + const unsafe = "native_workspace_sync_out_unsafe_archive"; + let coordinator: ReturnType | undefined; + let admitted = true; + const errors: unknown[] = []; + beforeEach(async () => { + await db.update(nativeRunFinalizations).set({ failureCode: "unrelated_test_history" }) + .where(and(eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.failureCode, unsafe))); + admitted = true; errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + expect(errors).toEqual([]); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + await registerUnsafeWorkspaceExportWork(db, coordinator, () => admitted).ready; + } + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 15_000 }); + } + async function quiet() { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); } + finally { select.mockRestore(); transaction.mockRestore(); } + } + const phase = async (runId: string) => (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase; + async function seedBeforeFailure() { + const f = await seed(unsafe); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + await db.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ status: "running", nativePhase: "result_accepted" }).where(eq(heartbeatRuns.id, f.runId)); + return f; + } + function failExport(f: Awaited>, owner = db) { + return recordNativeFinalizationFailure({ db: owner, runId: f.runId, error: new Error(unsafe), + failureScope: "workspace", projectRunStatus: true, permanent: true }); + } + function gate() { + let release!: () => void; + const promise = new Promise(resolve => { release = resolve; }); + return { promise, release }; + } + + it("stays quiet with no matching work", async () => { + await seed(); await start(); await flush(); await quiet(); expect(probe).not.toHaveBeenCalled(); + }); + it.each(["newer_run", "older_active_run", "competing_action", "cancelled", "changed_contract", "explicit_settlement", "cancelled_action", "null_settlement", "rejected"])("does not poll dormant history: %s", async kind => { + const f = await seed(unsafe); + await db.update(issueRecoveryActions).set({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }).where(eq(issueRecoveryActions.id, f.actionId)); + if (["newer_run", "older_active_run"].includes(kind)) await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: f.issueId, + status: kind === "older_active_run" ? "running" : "failed", createdAt: new Date(Date.now() + (kind === "older_active_run" ? -60_000 : 60_000)) }); + if (kind === "competing_action") await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", ownerType: "board", cause: "other_repair", fingerprint: randomUUID(), nextAction: "Keep this repair", evidence: {} }); + if (kind === "cancelled") await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); + if (kind === "changed_contract") { + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, f.issueId)); + await db.insert(completionContracts).values({ ...contract, id: randomUUID(), revision: contract.revision + 1, canonicalSha256: randomUUID() }); + } + if (["explicit_settlement", "null_settlement"].includes(kind)) await db.update(issueRecoveryActions).set({ resolutionNote: kind === "null_settlement" ? null : "operator_resolved" }).where(eq(issueRecoveryActions.id, f.actionId)); + if (kind === "cancelled_action") await db.update(issueRecoveryActions).set({ status: "cancelled" }).where(eq(issueRecoveryActions.id, f.actionId)); + if (kind === "rejected") await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, f.resultId)); + await start(); await flush(); expect(await phase(f.runId)).toBe("terminal_failure"); await quiet(); + expect(probe).not.toHaveBeenCalled(); + }); + it("retires a stale action even when a newer run prevents result readmission", async () => { + const f = await seed(unsafe); + await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: f.issueId, status: "failed", createdAt: new Date(Date.now() + 60_000) }); + await start(); await flush(); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0]).toMatchObject({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }); + expect(await phase(f.runId)).toBe("terminal_failure"); await quiet(); + }); + it("discovers a newer accepted result after retiring an older unsafe notice", async () => { + const older = await seed(unsafe); + const [oldRun] = await db.update(heartbeatRuns).set({ createdAt: new Date(Date.now() - 60_000) }) + .where(eq(heartbeatRuns.id, older.runId)).returning(); + const [oldResult] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, older.resultId)); + const runId = randomUUID(), resultId = randomUUID(); + await db.insert(heartbeatRuns).values({ ...oldRun, id: runId, createdAt: new Date() }); + await db.insert(nativeRunResults).values({ ...oldResult, id: resultId, runId, + serverFingerprint: resultId, canonicalSha256: resultId }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId: older.issueId, + phase: "terminal_failure", resultId, failureCode: unsafe }); + const accepted = await db.select().from(nativeRunResults).where(eq(nativeRunResults.issueId, older.issueId)); + + + + await start(); await flush(); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, older.actionId)))[0]) + .toMatchObject({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }); + expect(await phase(older.runId)).toBe("terminal_failure"); + expect(await phase(runId)).toBe("result_accepted"); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.issueId, older.issueId))).toEqual(accepted); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, older.issueId))).toHaveLength(2); + expect(probe).not.toHaveBeenCalled(); + await quiet(); + }); + it("wakes from a real finalization failure and readmits only the accepted result", async () => { + const f = await seedBeforeFailure(); await start(); await flush(); await quiet(); + await failExport(f); await flush(); expect(await phase(f.runId)).toBe("result_accepted"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toHaveLength(1); + expect(probe).not.toHaveBeenCalled(); await quiet(); + }); + it.each(["commit", "rollback"] as const)("waits for the outer failure transaction to %s", async outcome => { + const f = await seedBeforeFailure(); await start(); await flush(); + const entered = gate(), release = gate(); + const writing = db.transaction(async tx => { + await failExport(f, tx as unknown as typeof db); entered.release(); await release.promise; + if (outcome === "rollback") throw new Error("rollback unsafe export"); + }).then(() => null, error => error); + await entered.promise; await vi.advanceTimersByTimeAsync(2000); + expect(await phase(f.runId)).toBe("result_accepted"); release.release(); + const error = await writing; + if (outcome === "rollback") expect(error).toMatchObject({ message: "rollback unsafe export" }); + else expect(error).toBeNull(); + await flush(); expect(await phase(f.runId)).toBe("result_accepted"); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run.status).toBe("running"); await quiet(); + }); + it.each(["restart", "drain"])("recovers saved work after %s", async mode => { + const f = await seedBeforeFailure(); + if (mode === "drain") { await start(); await flush(); admitted = false; } + await failExport(f); + if (mode === "restart") await start(); + else { await quiet(); expect(await phase(f.runId)).toBe("terminal_failure"); admitted = true; await vi.advanceTimersByTimeAsync(30_000); } + await flush(); expect(await phase(f.runId)).toBe("result_accepted"); await quiet(); + }); + it("retains a concrete coordinator lease until its owner releases it", async () => { + const f = await seed(unsafe); + await db.update(nativeRunFinalizations).set({ leaseOwner: "still-owned", leaseExpiresAt: new Date(Date.now() + 300_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); expect(await phase(f.runId)).toBe("terminal_failure"); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await vi.advanceTimersByTimeAsync(30_000); await flush(); expect(await phase(f.runId)).toBe("result_accepted"); await quiet(); + }); + it("waits for physical workspace ownership without stealing or discarding the obligation", async () => { + const f = await seed(unsafe); + const owned = await withNativeWorkspaceFinalizationOwnership(f, async () => { + await start(); + await vi.advanceTimersByTimeAsync(60_000); + expect(await phase(f.runId)).toBe("terminal_failure"); + await vi.waitFor(() => { + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + }); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("active"); + }); + expect(owned.acquired).toBe(true); + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(await phase(f.runId)).toBe("result_accepted"); + expect(probe).not.toHaveBeenCalled(); await quiet(); + }); + it.each(["admin", "stale_lock_recovery"])("reactivates dormant history after %s releases an older run's locks", async mode => { + const f = await seed(unsafe); + await db.update(issueRecoveryActions).set({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }).where(eq(issueRecoveryActions.id, f.actionId)); + const [older] = await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: f.issueId, + status: "failed", createdAt: new Date(Date.now() - 60_000) }).returning(); + await db.update(issues).set({ checkoutRunId: older.id, executionRunId: older.id }).where(eq(issues.id, f.issueId)); + await start(); await flush(); + expect(await phase(f.runId)).toBe("terminal_failure"); await quiet(); + if (mode === "admin") await issueService(db).adminForceRelease(f.issueId); + else expect((await recoveryService(db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks()).issueIds).toContain(f.issueId); + await flush(); expect(await phase(f.runId)).toBe("result_accepted"); + expect(probe).not.toHaveBeenCalled(); await quiet(); + }); + it("stops pending retries and reconstructs them on restart", async () => { + const f = await seed(unsafe); + await db.update(nativeRunFinalizations).set({ leaseOwner: "still-owned", leaseExpiresAt: new Date(Date.now() + 300_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await coordinator!.stop(); coordinator = undefined; + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await quiet(); expect(await phase(f.runId)).toBe("terminal_failure"); + await start(); await flush(); expect(await phase(f.runId)).toBe("result_accepted"); await quiet(); + }); + it("revalidates task cancellation while a physical owner delays recovery", async () => { + const f = await seed(unsafe); + await withNativeWorkspaceFinalizationOwnership(f, async () => { + await start(); + await issueService(db).update(f.issueId, { status: "cancelled" }); + }); + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(await phase(f.runId)).toBe("terminal_failure"); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].status).toBe("resolved"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed"); + expect(probe).not.toHaveBeenCalled(); await quiet(); + }); + it("discovers and drains more than one page of stale notices", async () => { + const fixtures = []; + for (let index = 0; index < 103; index++) { + const f = await seed(unsafe); + await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); + fixtures.push(f); + } + await start(); await flush(); + for (const f of fixtures) { + expect(await phase(f.runId)).toBe("terminal_failure"); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0]).toMatchObject({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }); + } + expect(probe).not.toHaveBeenCalled(); await quiet(); + }, 30_000); + it("reactivates dormant history when the real cancellation path settles an older queued run", async () => { + const f = await seed(unsafe); + await db.update(issueRecoveryActions).set({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }).where(eq(issueRecoveryActions.id, f.actionId)); + const [older] = await db.insert(heartbeatRuns).values({ companyId, agentId, contextSnapshot: { issueId: f.issueId }, + status: "queued", runtimeMode: "legacy", createdAt: new Date(Date.now() - 60_000) }).returning(); + await start(); await flush(); expect(await phase(f.runId)).toBe("terminal_failure"); await quiet(); + await heartbeatService(db).cancelRun(older.id); + await flush(); expect(await phase(f.runId)).toBe("result_accepted"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, older.id)))[0].status).toBe("cancelled"); + expect(probe).not.toHaveBeenCalled(); await quiet(); + }); + it("retries failed discovery after admission resumes", async () => { + const f = await seed(unsafe); + const read = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("injected unsafe discovery failure"); }); + await start(); await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); read.mockRestore(); + errors.length = 0; admitted = false; await quiet(); admitted = true; + await vi.advanceTimersByTimeAsync(30_000); await flush(); expect(await phase(f.runId)).toBe("result_accepted"); await quiet(); + }); + it("does not wake dormant history for unchanged action metadata or repeated resolution", async () => { + const f = await seed(unsafe); + await db.update(issueRecoveryActions).set({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }).where(eq(issueRecoveryActions.id, f.actionId)); + const actionInput = { companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog" as const, + ownerType: "board" as const, cause: "other_repair", fingerprint: randomUUID(), nextAction: "Wait for repair" }; + const service = issueRecoveryActionService(db); + const other = await service.upsertSourceScoped(actionInput); + await start(); await flush(); await quiet(); + await service.upsertSourceScoped({ ...actionInput, nextAction: "Updated operator guidance" }); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + const resolution = { companyId, sourceIssueId: f.issueId, actionId: other.id, + status: "resolved" as const, outcome: "restored" as const, resolutionNote: "Other repair finished" }; + await service.resolveActiveForIssue(resolution); await flush(); + expect(await phase(f.runId)).toBe("result_accepted"); + await service.resolveActiveForIssue(resolution); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("reactivates dormant history from the real recovery-action resolution", async () => { + const f = await seed(unsafe); + await db.update(issueRecoveryActions).set({ status: "resolved", resolutionNote: "unsafe_export_automatic_recovery" }).where(eq(issueRecoveryActions.id, f.actionId)); + const [other] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", ownerType: "board", cause: "other_repair", fingerprint: randomUUID(), nextAction: "Wait for repair" }).returning(); + await start(); await flush(); await quiet(); + await issueRecoveryActionService(db).resolveActiveForIssue({ companyId, sourceIssueId: f.issueId, actionId: other.id, + status: "resolved", outcome: "restored", resolutionNote: "Other repair finished" }); + await flush(); expect(await phase(f.runId)).toBe("result_accepted"); await quiet(); + }); + it("reactivates dormant history from a real task update", async () => { + const f = await seed(unsafe); + await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); + await start(); await flush(); await quiet(); + await issueService(db).update(f.issueId, { status: "blocked" }); + await flush(); expect(await phase(f.runId)).toBe("result_accepted"); await quiet(); + }); + }); + }); diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.ts b/server/src/services/native-runtime/native-workspace-reconstruction.ts index 07191d7b411..b8ed3709fc3 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { and, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm"; import { agentWakeupRequests, assets, environmentLeases, heartbeatRuns, issueAttachments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { reconstructWorkspaceSchema, type ExecutionReconciliation } from "@paperclipai/shared"; @@ -177,6 +178,7 @@ export async function reconstructNativeWorkspace(input: { ...(source.interveningExecutions.length ? { interveningExecutions: source.interveningExecutions } : {}) } }; await markExecutionReconciliation(db, source.action, decision, input.actorId); const now = new Date(); + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.issueId }); await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "handed_back", resolvedAt: now, updatedAt: now, resolutionNote: "workspace_reconstruction_authorized", nextAction: "Continue from saved sources. The original unexported edits are lost; verify the selected attachments before rebuilding. Do not repeat completed external actions or reset the original work limits.", evidence: sql`${issueRecoveryActions.evidence} || ${JSON.stringify({ workspaceReconstructionRequest: requestSha256 })}::jsonb`, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index f2c780837e1..1606994752e 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 { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js"; import { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -1072,6 +1073,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 notifyUnsafeWorkspaceExportRecovery(input.tx, { companyId: input.companyId, issueId: input.issue.id }); await notifyDeliveryWork(input.tx, DELIVERY_QUEUES.accounting, input.companyId); const [run] = await input.tx .update(heartbeatRuns) @@ -1503,6 +1505,7 @@ async function materializeDecisionEffect(input: { }; } if (effect.kind === "release_checkout") { + await notifyUnsafeWorkspaceExportRecovery(input.tx, { companyId: input.companyId, issueId: input.issue.id }); const [released] = await input.tx .update(issues) .set({ diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index fa02e7a09af..3870dbb66c0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; @@ -6253,6 +6254,7 @@ export function recoveryService( await notifyCancelledWorkspacePreservation(tx, { companyId: candidate.companyId, runId: candidate.id }); const boundIssueId = boundRunIssueId(candidate); if (boundIssueId) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: boundIssueId }); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: boundIssueId }); } return tx @@ -6449,7 +6451,9 @@ export function recoveryService( continue; } - const updated = await db + const updated = await db.transaction(async tx => { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); + return tx .update(issues) .set({ checkoutRunId: null, @@ -6471,6 +6475,7 @@ export function recoveryService( ) .returning({ id: issues.id }) .then((rows) => rows[0] ?? null); + }); if (!updated) continue; diff --git a/server/src/services/slack-conversation-lifecycle.ts b/server/src/services/slack-conversation-lifecycle.ts index fdd6064ccd5..e89b09a11ef 100644 --- a/server/src/services/slack-conversation-lifecycle.ts +++ b/server/src/services/slack-conversation-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { and, desc, eq, inArray, sql } from "drizzle-orm"; import { chatConversations, chatEndpoints, heartbeatRuns, issueComments, issues, type Db } from "@paperclipai/db"; import { persistActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; @@ -90,6 +91,7 @@ export async function settleExternalConversation(db: Db, companyId: string, issu .where(and(eq(chatConversations.id, binding.conversation.id), eq(chatConversations.companyId, companyId), eq(chatConversations.state, "active"))) .returning({ id: chatConversations.id }); if (!changed.length) return false; + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); await tx.update(issues).set({ status: "in_review", updatedAt: new Date() }) .where(and(eq(issues.id, issueId), eq(issues.companyId, companyId))); activity = (await persistActivity(tx as unknown as Db, { diff --git a/server/src/services/unsafe-workspace-export-notifications.ts b/server/src/services/unsafe-workspace-export-notifications.ts new file mode 100644 index 00000000000..adcf9932be4 --- /dev/null +++ b/server/src/services/unsafe-workspace-export-notifications.ts @@ -0,0 +1,16 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type UnsafeWorkspaceExportScope = { companyId: string; issueId?: string }; + +export async function notifyUnsafeWorkspaceExportRecovery(transaction: object, scope: UnsafeWorkspaceExportScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.unsafeWorkspaceExportRecovery, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +} + +export async function notifyUnsafeWorkspaceExportRunChange(transaction: object, run: { + companyId: string; nativeIssueId: string | null; contextSnapshot: Record | null; +}) { + const contextIssueId = run.contextSnapshot?.issueId; + const issueIds = new Set([run.nativeIssueId, typeof contextIssueId === "string" ? contextIssueId : null]); + for (const issueId of issueIds) if (issueId) await notifyUnsafeWorkspaceExportRecovery(transaction, { companyId: run.companyId, issueId }); +}