From a019d00a393b57f531afe8dd8ab0b8a1820d1c4c Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 04:40:01 -0500 Subject: [PATCH] fix(server): wake native workspace finalization from committed work (#17734) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs can accept a result before workspace copy-back finishes. > - The periodic orphan pass repeatedly tries this recovery, including cases that require operator action. > - Copy-back needs exact ownership evidence. A generic timer cannot grant that authority. > - This pull request wakes workspace recovery from committed changes and local owner completion. > - Saved retries use the existing scheduler. Empty work, completed barriers, and operator holds become quiet. ## Linked Issues or Issue Description Refs #17717. This is slice 52 of the recurring-server sleep stack, targeting master after parent #17717 merged at `4c56702e06c698cc10d4a86514669aba0f4d9a9b`. Related work: #17709 (status reassessment), #17687 (unsafe export recovery), and #16478 (workspace repair holds). This PR covers native workspace copy-back and validation. Result commit/projection and resource settlement remain separate recovery duties. It does not claim full-instance sleep. ## What Changed - Register a company/run/task-scoped workspace worker. Reconstruct accepted, unfinished work from existing records. Use saved retry and lease deadlines. - Notify from both accepted-result writers (control-plane and PRP coordinator), workspace receipts, coordinator changes, operator confirmation, retained workspace repair, and relevant task updates. Notifications follow the outer transaction. - Distinguish active local ownership, busy external ownership, pending Cloud evidence, operator-required evidence, and changed bindings. Only concrete external ownership/evidence waits receive a scoped retry. - Preserve advisory locks, signed stop proof, exact binding checks, physical callback joins, and existing failure policy. Join physical operations during shutdown. - Respect the existing heartbeat execution-suppression policy before discovery and execution; wake suppressed work on settings activation. - Skip physical workspace recovery in the enabled periodic orphan pass. Keep direct/startup reconciliation and the remaining result/cleanup recovery paths. - Keep routine company-only finalization hints quiet. Ordinary work requires issue/run scope; explicit settings and owner-release events keep their company-wide contract. - Preserve the parent's entire Runner/package tree, profile 80 and all historical identities. Add the dated obligation, producer, lifecycle and limitation audit. ## Verification - Current local source: `4059b24e3ea5c623be54ade18ec9b6ab78eb0289`, prepared by a normal merge of the current parent. The sole manual conflict keeps ordinary workspace fanout inside the existing issue/run guard. All 1,198 parent-only, 15 owned-only and 10 automatic shared paths are preserved. The complete package tree matches the parent byte-for-byte. - Controlled regression: the original workspace-only unscoped fanout performs three candidate SELECTs for routine company hints. The same zero-query assertion passes with scoped fanout. A real accepted result then produces one successful workspace settlement and returns to quiet. The failed baseline remains in the audit. - The first complete local group passes **1,693 cases in 14 files**, including all 19 workspace-worker cases. It covers both accepted-result producers, physical ownership, export retry/failure, finalization recovery, arbiter corpus, issue recovery actions, control-plane port, workspace recovery, heartbeat recovery, startup, retained cleanup and native execution. - Separate process-recovery and external-chat response files qualify the changed recovery/settlement paths without sharing their general test database with other files. Both pass in full: 410 process-recovery cases in 162.79 seconds and 273 external-chat cases in 53.79 seconds. The total is **2,376 distinct passes in 16 complete files**, with no skips. Normal merge `5c46c52873158529736b8349e4d44658865108ee` adopts the parent squash with this entire tested child tree unchanged. - Production server typechecking, build, module boundaries, migration order and source-profile checks pass. Thirteen dedicated PostgreSQL 17 fixtures use the unchanged migration 413 schema. Embedded-only paths use the existing recovery helper with fresh per-call fixtures. All temporary imports are restored byte-exact. No test assertion or timeout is relaxed. - Runner validation belongs to the unchanged parent package tree: 130 passing installation/compatibility/transport cases, 22 installed ACPX/Grok contract cases and 26 native-provider Rust cases. One existing opt-in closed Pi snapshot case is skipped because its distribution root is not configured. These are source-scoped checks, with no live provider qualification claim. - The owner audit at `doc/plans/2026-10-26-native-workspace-finalization-sleep-audit.md` preserves earlier failures and their disposition. Current-head acceptance: all 57 checks are clean, including protected CI and security. Greptile gives 5/5 on `7df5c9c44ea9e171be3c61b4430d3727b10f3337` with no actionable findings. Earlier scores do not substitute for this result. No Ponytail verdict is present under the authorized missing-review policy. - No claim of a full local repository suite, deployed app/database sleep, expanded Cloud wake coverage or measured cost savings is made. ## Risks - Supported wake producers are in-process application services. This adds no cross-replica or independent SQL-writer transport. - A real unresolved export still needs recovery. Busy owners and pending Cloud stop evidence retain a 30-second scoped check. No timeout authorizes copy-back. - Result projection and resource release still use their existing recovery. Some existing cleanup failures remain best effort; this PR does not infer cleanup success from a workspace barrier. - Cloud alarm coverage and host-sleep safety remain unchanged. No deployed suspension or cost savings have been measured. - Parent #17717 is merged. Current base is master at `4c56702e06c698cc10d4a86514669aba0f4d9a9b`. Fresh current-head CI and review must pass before this slice merges. ## Model Used OpenAI Codex, GPT-6 family, with repository tools and code execution. The exact serving model ID and context-window size are not exposed in this session. No subagent was used for this slice. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: Codex --- ...2026-10-24-recurring-server-sleep-stack.md | 67 ++++ ...tive-workspace-finalization-sleep-audit.md | 267 +++++++++++++++ .../__tests__/issue-recovery-actions.test.ts | 2 +- ...native-workspace-finalization-work.test.ts | 317 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 22 +- server/src/index.ts | 12 +- .../services/delivery-work-notifications.ts | 1 + server/src/services/heartbeat.ts | 9 + server/src/services/heartbeat/recovery.ts | 17 +- server/src/services/instance-settings.ts | 2 + server/src/services/issues.ts | 2 + .../native-finalization-notifications.ts | 2 + .../native-finalization-reconciler.ts | 28 +- .../native-run-coordinator-store.ts | 2 + ...native-workspace-finalization-ownership.ts | 47 ++- .../native-workspace-finalization-work.ts | 104 ++++++ .../native-workspace-finalizer.test.ts | 30 +- .../native-workspace-finalizer.ts | 35 +- .../native-workspace-owner-recovery.ts | 2 + .../native-workspace-platform-proof.test.ts | 26 +- .../native-workspace-platform-proof.ts | 24 +- .../paperclip-control-plane-port.ts | 2 + ...ve-workspace-finalization-notifications.ts | 24 ++ server/src/services/workspace-operations.ts | 3 + .../src/services/workspace-recovery.test.ts | 4 +- server/src/services/workspace-recovery.ts | 2 + 26 files changed, 987 insertions(+), 66 deletions(-) create mode 100644 doc/plans/2026-10-26-native-workspace-finalization-sleep-audit.md create mode 100644 server/src/__tests__/native-workspace-finalization-work.test.ts create mode 100644 server/src/services/native-runtime/native-workspace-finalization-work.ts create mode 100644 server/src/services/native-workspace-finalization-notifications.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 4a7392f5a9e..85017a8f68e 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -3164,3 +3164,70 @@ are finished. Audit first-result publication in `paperclip-control-plane-port.ts workspace operation writes, durable workspace-owner release, retry updates and cleanup completion before disabling their existing recovery paths. No migration of these next duties is claimed by slice 51. + +## 2026-10-26: verified slice 51 and workspace-finalization migration + +[#17717](https://github.com/paperclipai/paperclip/pull/17717) is verified at +`d0036fce3e039e91ab2bf2602e3e6a87a4476491`: all 56 remote check/status entries +passed or intentionally skipped, Greptile 5/5 on that head, and no unresolved +review threads. It remains draft/open and unmerged. The older local full-suite +run was interrupted with exit 130 after current-head CI passed; do not count it +as a local full-suite pass. The separate local resource-limit failure remains +recorded. The explicitly approved 11 semaphore sets were removed once. + +Slice 52 is `codex/sleep-52-native-workspace-finalization` on that parent. +[Its audit](2026-10-26-native-workspace-finalization-sleep-audit.md) records the +copy-back obligation, exact owner/proof outcomes, commit/release producers, +shared-scheduler integration, and 12 passing real-database lifecycle cases. +The periodic orphan pass skips only this physical-copyback duty. Result projection +and resource settlement retain existing recovery and are the next separate slices. +Final full verification, publication and review are pending for slice 52. Inspected +master is `a1f77821b9cd5c9c846fdf7a0fa5cc620ad2b6bb`; stack-base propagation is +still outstanding. Older gates (#17490, #17572, #17576, #17580) and the rest of the +requested recurring-server inventory remain active. No merge is authorized. + +Slice 52 publication: [#17734](https://github.com/paperclipai/paperclip/pull/17734) +is draft on #17717. Workspace types/build passed. The initial broad command had +582 passes and 88 board-action failures from incorrectly shared external test +fixtures; it is not a pass. Separate databases then pass all 203 cases in four +complete follow-up suites. A reproduced stale-success-barrier regression is +fixed: result reconciliation prefers the existing successful scoped receipt even +when a stale controller later writes failure, without repeating copy-back. +The worker lifecycle now passes all 13 cases. Full-command disposition and +repaired-head remote gates remain pending. No merge; all older gates and later +recurring duties remain in scope. + +Slice 52 review repair: initial Greptile 3/5 identified the already-reproduced +stale barrier and missing heartbeat execution-suppression checks. Restore/worktree +regressions both failed red through public heartbeat registration. The worker now +consults the owning suppression policy before discovery/execution, and worktree +activation publishes a reconstruction signal. All 15 lifecycle cases pass, +including real settings activation after quiet. Latest validation/review are still +pending; the scope remains active with no merge. + + +Slice 52 startup repair: Greptile reached 5/5 at `020ccbc66d` with both threads +resolved, but CI server shard 4/12 found a missing startup fixture method. Added +that method and enabled/disabled lifecycle coverage, then corrected the exact +periodic skip expectation. All 74 startup cases pass. The old-head default local +full run was deliberately interrupted (Vitest 130, wrapper 1) before editing; +it is not a pass. Remote gates and review on the repaired head remain pending. + +Follow-up audit: eight earlier native worker registrations gate task drain but +lack the new worker's restore/worktree suppression boundary: retained session +cleanup, status reassessment, governed response, response presentation, pending +effects, unsafe exports, automatic completion review and obsolete policy review. +Verify each duty's prior suppression contract and its settings activation wake +before declaring it safe. This is outstanding stack acceptance work, not a claim +that every listed duty has the same defect. All remaining duties and older gates +remain active; no merge is authorized. + + +Slice 52 producer audit correction: the hidden PRP coordinator has a second +accepted-result writer. Added its missing transaction-aware workspace hint after +two red real-producer regressions. The complete lifecycle suite passes all 18 +cases, covering first result plus outer commit/rollback for both writer paths. +The previous head's startup gate passed, with Greptile 5/5. A separate CI import +hook timeout (118 agent-skills cases skipped) passed all 118 locally and received +one justified job retry. New producer-fix acceptance remains pending. No merge; +all prior gate, suppression-audit and remaining-duty obligations remain open. diff --git a/doc/plans/2026-10-26-native-workspace-finalization-sleep-audit.md b/doc/plans/2026-10-26-native-workspace-finalization-sleep-audit.md new file mode 100644 index 00000000000..43a79778cf2 --- /dev/null +++ b/doc/plans/2026-10-26-native-workspace-finalization-sleep-audit.md @@ -0,0 +1,267 @@ +# Native workspace finalization sleep audit + +Slice 52. Parent: [#17717](https://github.com/paperclipai/paperclip/pull/17717) +at `d0036fce3e039e91ab2bf2602e3e6a87a4476491`. +Branch: `codex/sleep-52-native-workspace-finalization`. +Inspected remote master: `a1f77821b9cd5c9c846fdf7a0fa5cc620ad2b6bb`. +This is a stacked branch, not a claim that current master has been integrated. + +## The obligation + +An accepted native result can still owe workspace copy-back or validation. The +existing native result, finalization coordinator, workspace-operation receipts, +and physical-owner receipt are its durable evidence. This change adds no table. +An accepted result does not prove copy-back, result projection, or resource release +finished. This slice migrates copy-back/validation only. The remaining result +commit/projection and resource-settlement duties retain their current recovery. + +The normal orphan pass used to re-enter workspace finalization on each scan. +The enabled server now registers a scoped deadline worker and skips that physical +operation in its periodic orphan pass. Explicit and startup reconciliation keep +the existing direct API. The remaining reconciler still scans for its other duties; +this slice alone does not make the whole orphan pass quiet. + +## Boundary decisions + +- `native-workspace-finalizer.ts` owns workspace execution and its existing failure + classification. Both direct reconciliation and scheduled execution use that policy. +- The ownership service retains its dedicated advisory lock, physical callback join, + exact durable receipt, same-boot join evidence, and current-task checks. No elapsed + timeout grants permission to overwrite another owner's files. +- The platform-proof boundary now distinguishes verified evidence, pending evidence, + and unsupported configuration. Only verified signed evidence permits takeover. + The worker does not interpret a Railway response or implement signature policy. +- Acquisition refusal identifies an active local owner, an external/busy owner, + pending platform proof, operator-required proof, or a changed task/owner binding. + This prevents a universal retry for all refusals. +- Local owner completion is a hint to read the saved obligation. It also fires + after the original callback joins when clearing its receipt fails. Different DB + objects without that completion subscription retain a scoped busy-owner retry. +- Heartbeat supplies physical-operation tracking. Index only registers, gates, and + stops the worker. Shutdown stops the coordinator and joins the physical callback. + +## Trigger and deadline inventory + +| Obligation / transition | Producer or durable source | Wake and stop condition | +| --- | --- | --- | +| First accepted result | `PaperclipControlPlanePort.completeRun` and `NativeRunCoordinatorStore.completeRun` (including terminal-event reconstruction) | Intent in the result's transaction; outer commit wakes, rollback creates no work | +| Saved retry or finalization transition | Existing native-finalization notification fanout, coordinator and finalizer writers | Reconstruct `nextAttemptAt`; do not advance it on inspection | +| Finalization lease | Existing coordinator row | Wait for the saved expiry; execution re-reads it before copy-back | +| Workspace-operation progress/completion | Workspace recorder's `workspace_finalize` writes | Same-transaction scoped notification; any successful barrier disarms this worker | +| Retained workspace repair completes | `workspace-recovery.ts` writes a success barrier directly | Same-transaction notification removes an obsolete copy-back obligation | +| Local writer still owns copy-back | Ownership service's exact run/token and DB owner | Completion hint; no SQL/network timer while that local callback is active | +| Busy lock without a subscribed local owner | Physical ownership acquisition | Scoped 30-second check while this concrete export remains unresolved | +| Supported Cloud stop evidence remains pending | Authenticated platform proof request | Scoped 30-second retry; success still checks the exact task/run/result/owner before copying | +| Unsupported, malformed, or unverifiable owner | Existing board recovery action | Remain fenced and quiet; no fallback poll awaiting operator action | +| Operator confirms the exact owner stopped | `resumeNativeWorkspaceAfterOwnerStop` | Intent before clearing the receipt in its existing transaction; resume without another provider turn | +| Task changes during proof; later returns to eligible state | Existing issue update owner | Refuse stale copy-back; a scoped issue mutation notification permits a fresh check | +| Startup/resume | Existing result/coordinator/operation records | Keyset pages of 100; one physical operation per fair dispatch batch | + +Completed barriers, terminal-failure coordinators, rejected/unbound results, and +absent work do not create recurring deadlines. An unresolved export can still +require recovery and prevent host sleep; it is not silently discarded for quiet. +Startup reconstruction may inspect an operator-required record once. Retained +history with a completed barrier is excluded by SQL. + +Supported writers are the same-process application services using the existing +transaction-notification boundary. This does not add cross-replica notification +transport or support independent SQL writers. Environment/configuration changes +that enable Cloud verification require the existing restart/configuration lifecycle. +External alarm coverage and conservative host-sleep safety are unchanged. + +## Verification checkpoint + +The new platform contract was reproduced red before implementation. The five +platform-proof cases then passed. Complete affected ownership, retained recovery, +and board-action suites passed 206 cases at the first contract checkpoint. + +The worker lifecycle suite now passes 12 cases against a disposable PostgreSQL +server, using the real result producer, recorder, owner lock, and operator API: + +- Empty → real accepted result → successful barrier → retained history → restart. +- No SQL across multiple old poll periods when empty, completed, locally owned, + operator-required, or waiting on a changed binding. +- Real result writes inside an outer commit and rollback. +- Startup and first-result writes during drain, followed by admission reopening. +- Reconstruction and enforcement of a saved retry deadline. +- Local owner release and shutdown joining a deliberately held physical callback. +- Operator confirmation after quiet; pending Cloud evidence followed by verified proof. +- Task restoration through the normal issue service after a stale-binding refusal. +- Real validation failure, persisted retry, and successful validation after repair. + +Preserved test failures: the first two drain cases used a nonexistent test-harness +`coordinator.resume()` method; the test now reopens admission and advances the +existing scheduler. The first failed-validation assertion checked the next deadline +before the retry-write notification settled; it now waits for notification processing +and still asserts the exact saved deadline. Neither required weakening a product guard. + +Server types passed at the initial worker checkpoint. The final complete affected +suites, workspace types/build, full test command, PR checks and current-head review +remain to be recorded before acceptance. No deployed app/database suspension or +fleet cost reduction has been measured. + +## Remaining work outside this slice + +The next two owners are accepted-result commit/projection and finalization resource +settlement. Existing cleanup paths swallow some release errors; a workspace success +barrier must not be used to declare that cleanup settled. Retained workspaces, native +status wakes, process/session recovery, scheduling, maintenance, and the older stack +review failures remain in the main stack ledger. No stack PR is merged by this work. + +## Publication and follow-up verification + +Draft [#17734](https://github.com/paperclipai/paperclip/pull/17734) was published +at `ad2b4bc9d87bb9c8e7cd6cba4abc122702ea7c5c`, on #17717. Workspace-wide types +and build passed at that checkpoint. The final eight-suite command reported +582 passing cases and 88 failures in the board-action suite. That command +incorrectly reused one external database for three suites; the board-action +suite's instance-wide reconciliation saw their retained fixtures. Do not count +that combined command as a pass. The corrected follow-up gives each suite its +own database and passes all 203 cases across four complete suites: workspace +worker lifecycle, board actions, native result recovery, and physical workspace +finalization. The other seven complete suites passed in the initial broad run, +including all 408 heartbeat process-recovery and 34 recovery-module cases. + +Review also found a real migration regression: the workspace worker correctly +excludes any successful barrier, but the remaining result reconciler used only +the latest receipt. A later stale failure could then leave result projection +waiting forever after periodic copy-back was disabled. The added regression +failed with `workspace_finalizing` instead of `committed`. The reconciler now +prefers a successful company/task/run-scoped receipt, matching the physical +finalizer's existing success rule. The complete 13-case worker lifecycle suite +passes with this repair, including the normal result-reconciliation path and no +repeated physical copy-back. + +Repaired-head workspace validation, the default full test command, current-head +CI, and Greptile acceptance remain to be recorded. The PR is not accepted or merged. + +## Execution-suppression review repair + +Greptile's first review found the stale-barrier defect above and a second valid +boundary omission: registration checked idle drain but did not check the existing +heartbeat suppression policy for restore and disabled worktree instances. Both +public-heartbeat regression cases failed by creating a successful workspace receipt +while recovery was suppressed. Registration now asks the existing fresh suppression +policy before discovery and before each operation. Suppressed work creates no +recurring deadline. The existing worktree-activation settings transaction publishes +a workspace notification so later activation reconstructs the saved obligation. +The full lifecycle suite now passes all 15 cases, including disabled startup, +quiet waiting, and activation through the real settings service. The source of +suppression policy remains heartbeat; neither the worker nor index interprets flags. + +Workspace types and build passed again at `b11b1229084d9f2f5cb332b6b11beca40ece5552` +before this suppression repair. Validation for the new repair and fresh review +remain required; the initial 3/5 review is not acceptance. + + +## Startup fixture repair and full-command disposition + +At `020ccbc66d`, Greptile reached 5/5 and both review threads were resolved. +CI server shard 4/12 failed 24 startup cases because its heartbeat fixture omitted +`registerNativeWorkspaceFinalizationWork`. The fixture now exposes that contract +and covers enabled/disabled registration plus the close subscription. A full +local rerun then exposed the stale exact periodic-recovery expectation; it now +includes `skipNativeWorkspaceFinalization: true`. The complete startup suite +passes all 74 cases, without weakening the remaining-duty assertions. + +The default local `pnpm test:run` at `020ccbc66d` was interrupted with SIGINT +once this concrete CI repair was known, before changing source or tests. The +Vitest child exited 130 and the wrapper exited 1. It is not a passing full run. +The log is `/tmp/sleep52-full-test-run.log`; affected-suite passes and full remote +CI remain separate evidence. No additional semaphore cleanup was performed. +Latest-head remote gates and review must be refreshed after the fixture repair. + + +## Second accepted-result producer + +The follow-on result-commit audit found the PRP coordinator's independent +`NativeRunCoordinatorStore.completeRun` writer. Its first accepted-result write +had no workspace notification; the narrower control-plane producer tests did +not cover it. Two new real-database cases reproduced the defect: first result +and outer commit both left workspace recovery quiet with no successful receipt. +Rollback remained quiet as expected. Register the shared intent before this +writer inserts the accepted row. Terminal-event reconstruction calls the same +method, so it inherits the transaction boundary. No new table or caller wrapper. + +The full lifecycle suite now passes 18 cases, covering both supported result +writers and their outer commit/rollback paths. Source search found these two +production accepted-result writers; coordinator retry changes already use the +native-finalization notification fanout. + +At the prior head `911bd1f7d0`, Greptile was 5/5 with both threads resolved and +the startup CI repair passed. An independent agent-skills suite timed out in its +30-second module-import hook (118 tests skipped); its complete local rerun passed +all 118 tests in 18.66 seconds. One justified CI retry was requested. The first +request returned HTTP 500 without starting an attempt; resubmission after verifying +attempt 1 was complete started attempt 2. This is not acceptance of the new +producer fix, whose current-head checks and review still need verification. + +## Readiness follow-up + +Startup now tracks the workspace-finalization registration's `ready` promise +through the existing idle/shutdown work tracker instead of awaiting its first +physical copy-back before declaring HTTP readiness. The subscription remains +registered for shutdown and the active work remains joined; no physical ownership +or finalization policy changes. A held-callback regression failed before the +change and now proves readiness, registration of later recovery workers, and +shutdown waiting for the callback. All 77 startup cases and all 18 real-database +workspace-finalization lifecycle cases passed. Integrated typecheck, build, and +remote checks remain required. + +## Current parent integration and scope regression (2026-10-29) + +The current parent is #17717 at reviewed head +`281d579de0d08e907194b8b790a0597b07d16b7b`. Its normal merge into this +slice preserves 1,198 parent-only paths, 15 owned-only paths, 11,517 shared +paths, and 10 automatic shared merges. The only manual conflict is ordinary +native-finalization fanout. Workspace finalization joins the existing issue/run +scope guard there. Explicit company-wide settings and owner-release notifications +keep their existing contracts. The entire Runner/package tree is byte-exact to +the parent, including combined source profile 80 and both historical identities. + +The new real-database regression starts an idle workspace worker and submits +three routine company-only finalization hints. A controlled baseline moves only +workspace fanout outside the guard, retaining every other parent's guard. It +produces three candidate SELECTs and fails the zero-query assertion. Restoring +the production bytes passes the same assertion; a real accepted result then +creates one successful workspace settlement and returns to quiet. The original +18 lifecycle cases are not included in that focused one-case result; the full +file is included in broader validation below. Both phases use fresh dedicated +PostgreSQL 17 databases, and restore the production notification helper exactly. + +Fifteen added comments in eight files were removed before verification. Their +nontrivia TypeScript token streams are unchanged; no directive was removed. +No test assertion or timeout was weakened. No new table, migration, provider +launch, independent timer or permanent database listener is introduced. + +### Complete current-source acceptance + +Source `4059b24e3ea5c623be54ade18ec9b6ab78eb0289` passes **2,376 distinct +cases in 16 complete files**, with no skips in those files. The 14-file main +command passes 1,693 cases in 270.99 seconds. Separate process recovery passes +410 in 162.79 seconds; external-chat response passes 273 in 53.79 seconds. Server +production typechecking, build, module boundaries, migration order and source +identity checks pass. Thirteen fresh PostgreSQL 17 databases use the unchanged +migration 413 schema. Embedded-only calls use existing fresh-per-call recovery +fixtures; all temporary imports are restored byte-for-byte, and every selected +test's saved hash matches after completion. Assertions and timeouts are unchanged. +The source-scoped tests cover both accepted-result producers, outer commit and +rollback, exact physical ownership, suppression/reactivation, coordinator leases, +copy-back retries/refusals, drain/resume, held startup work and shutdown joins. + +Parent #17717 merged as `4c56702e06c698cc10d4a86514669aba0f4d9a9b`, after +60 clean current-head checks and Greptile 5/5. Its complete reviewed tree equals +the landed squash. Normal merge `5c46c52873158529736b8349e4d44658865108ee` +adopts that ancestry and keeps this child's entire tested tree unchanged. This +slice can now target master. Fresh current-head CI and review remain separate +publication gates. + +No Runner source changed in this slice. The parent package tree's current profile +80 validation remains applicable: 130 passing installation/compatibility/transport +cases, 22 installed ACPX/Grok contract cases and 26 native-provider Rust cases. +One pre-existing opt-in closed Pi snapshot case was skipped because its distribution +root was unset; this does not qualify a live Pi installation. Earlier failed +commands and review findings above remain in the record. The new controlled +unscoped-hint baseline is an intended failing regression, with the same assertion +green after the scoped change. No full local repository suite, deployed sleep, +live provider qualification or expanded Cloud wake coverage is claimed. diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 11f05b1ca7e..fc9c02d48df 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -2728,7 +2728,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { const recovery = recoveryService(db, { enqueueWakeup: vi.fn(async () => null) }); let firstAction: typeof issueRecoveryActions.$inferSelect | undefined; for (let tick = 0; tick < 3; tick += 1) { - expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId }, copyback)).toEqual({ acquired: false }); + expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId }, copyback)).toEqual({ acquired: false, reason: "owner_unverified" }); const detail = await request(app).get(`/api/issues/${sourceIssueId}`).expect(200); expect(detail.body.activeRecoveryAction).toMatchObject({ cause: "native_workspace_finalization_owner_unverified", status: "active", attemptCount: 1, diff --git a/server/src/__tests__/native-workspace-finalization-work.test.ts b/server/src/__tests__/native-workspace-finalization-work.test.ts new file mode 100644 index 00000000000..b2fd74e8dfc --- /dev/null +++ b/server/src/__tests__/native-workspace-finalization-work.test.ts @@ -0,0 +1,317 @@ +import { heartbeatService } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { and, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, + nativeRunResults, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { issueService } from "../services/issues.js"; +import { nativeWorkspaceFinalizationWork } from "../services/native-runtime/native-workspace-finalization-work.js"; +import { withNativeWorkspaceFinalizationOwnership } from "../services/native-runtime/native-workspace-finalization-ownership.js"; +import { resumeNativeWorkspaceAfterOwnerStop } from "../services/native-runtime/native-workspace-owner-recovery.js"; +import * as platformProof from "../services/native-runtime/native-workspace-platform-proof.js"; +import { reconcileNativeFinalizations } from "../services/native-runtime/native-finalization-reconciler.js"; +import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; +import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; +import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +describe("workspace finalization scheduled lifecycle", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let coordinator: ReturnType | undefined; + let worker: ReturnType["register"]> | undefined; + let root: string; + const priorLogs = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + const companyId = randomUUID(), agentId = randomUUID(); + let admitted = true; + const errors: unknown[] = []; + const settled = vi.fn(async () => {}); + beforeAll(async () => { + root = await mkdtemp(join(tmpdir(), "workspace-finalization-work-")); + process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = join(root, "logs"); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_WORKSPACE_FINALIZATION_WORK_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("workspace-finalization-work-"); + db = createDb(external ?? temporary!.connectionString); + await db.insert(companies).values({ id: companyId, name: "Workspace finalization", issuePrefix: `WF${randomUUID().slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Workspace worker", adapterType: "paperclip_runner" }); + }, 30_000); + beforeEach(() => { + admitted = true; errors.length = 0; settled.mockClear(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + worker?.stop(); await coordinator?.stop(); worker = undefined; coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); await rm(root, { recursive: true, force: true }); + if (priorLogs === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorLogs; + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = nativeWorkspaceFinalizationWork(db, { onWorkspaceSettled: settled }).register(coordinator, () => admitted); + await worker.ready; + } + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 15_000 }); + } + async function quiet() { + const previous = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_id, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); await delay(10); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previous; } + } + async function seed() { + const runId = randomUUID(), issueId = randomUUID(), contractId = randomUUID(), sourceInstanceId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Preserve completed work", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "workspace-worker-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "workspace-worker-v1", objective: "Finish", criteria: [{ id: "objective", requirement: "Complete the task" }] }, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, runtimeMode: "native", status: "running", + nativeIssueId: issueId, nativeSessionId: runId, runnerInstanceId: sourceInstanceId, + completionContractId: contractId, completionContractSha256: contractId, contextSnapshot: { issueId } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + + await db.insert(workspaceOperations).values({ companyId, issueId, heartbeatRunId: runId, + phase: "workspace_finalize", status: "failed", cwd: root }); + return { companyId, issueId, runId, agentId, sessionId: runId, completionContractId: contractId, + completionContractSha256: contractId, sourceInstanceId, controlPlaneSourceInstanceId: "workspace-worker-test" }; + } + async function complete(binding: Awaited>, writer: Db = db, + producer: "control_plane" | "coordinator" = "control_plane") { + const port = producer === "control_plane" ? new PaperclipControlPlanePort(writer, binding) + : new NativeRunCoordinatorStore(writer, { ...binding, normalizedSessionId: binding.sessionId, + runnerSourceInstanceId: binding.sourceInstanceId, + completionContractRevision: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.contractRevision, + completionContractCriterionIds: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.criteria.map(item => item.criterionId) }); + await port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, + callerResultId: `${binding.runId}:result` }); + } + async function successful(runId: string) { + return db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, runId), + eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded"))); + } + it.each(["restore", "worktree"])("honors %s suppression through the public heartbeat registration", async mode => { + const binding = await seed(); await complete(binding); + const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" } + : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "workspace-suppression-test" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + if (mode === "worktree") await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const heartbeat = heartbeatService(db, { runtimeEnv }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = heartbeat.registerNativeWorkspaceFinalizationWork(coordinator, () => admitted); + await worker.ready; await flush(); await quiet(); + expect(await successful(binding.runId)).toHaveLength(0); + if (mode === "worktree") { + try { + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await flush(); expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + } + }); + it.each(["control_plane", "coordinator"] as const)("is quiet with no work and retained history, then wakes on the first %s accepted result", async producer => { + await start(); await flush(); await quiet(); + const binding = await seed(); + await complete(binding, db, producer); await flush(); + expect(await successful(binding.runId)).toHaveLength(1); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, binding.runId))).toHaveLength(1); + expect(settled).not.toHaveBeenCalled(); + await quiet(); + worker!.stop(); await coordinator!.stop(); await start(); await flush(); await quiet(); + expect(await successful(binding.runId)).toHaveLength(1); + }); + it("ignores routine company-only finalization hints and still wakes on accepted work", async () => { + await start(); await flush(); await quiet(); + const previous = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_id, query) => { + if (/from\s+"native_run_finalizations"/i.test(query)) queries.push(query); + }; + try { + for (let index = 0; index < 3; index += 1) { + await db.transaction(async tx => { await notifyNativeFinalizationWork(tx, { companyId }); }); + await flush(); + } + expect(queries).toEqual([]); + } finally { db.$client.options.debug = previous; } + const binding = await seed(); + await complete(binding); await flush(); + expect(await successful(binding.runId)).toHaveLength(1); + await quiet(); + }); + + it("lets result reconciliation use a successful barrier despite a later stale failure", async () => { + const binding = await seed(); await complete(binding); + await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: binding.runId, issueId: binding.issueId, + phase: "workspace_finalize", status: "succeeded", cwd: root }); + await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: binding.runId, issueId: binding.issueId, + phase: "workspace_finalize", status: "failed", cwd: root, createdAt: new Date(Date.now() + 1000) }); + await start(); await flush(); await quiet(); + await reconcileNativeFinalizations(db, [binding.runId], { skipNativeWorkspaceFinalization: true }); + const [result] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, binding.runId)); + expect(result.phase).toBe("committed"); + expect(await successful(binding.runId)).toHaveLength(1); + expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, binding.runId))).toHaveLength(3); + await flush(); await quiet(); + }); + it.each([ + ["commit", "control_plane"], ["rollback", "control_plane"], + ["commit", "coordinator"], ["rollback", "coordinator"], + ] as const)("respects the outer transaction's %s for the real %s result producer", async (outcome, producer) => { + await start(); await flush(); + const binding = await seed(); + const transaction = db.transaction(async tx => { + await complete(binding, tx as unknown as Db, producer); + await vi.advanceTimersByTimeAsync(5000); + expect(await successful(binding.runId)).toHaveLength(0); + if (outcome === "rollback") throw new Error("rollback fixture"); + }); + if (outcome === "rollback") await expect(transaction).rejects.toThrow("rollback fixture"); else await transaction; + await flush(); + expect(await successful(binding.runId)).toHaveLength(outcome === "commit" ? 1 : 0); + await quiet(); + }); + it.each(["startup", "first result"])("retains work during %s drain and resumes without a poll", async phase => { + admitted = false; + const binding = await seed(); + if (phase === "startup") await complete(binding); + await start(); + if (phase === "first result") await complete(binding); + await vi.advanceTimersByTimeAsync(120_000); + expect(await successful(binding.runId)).toHaveLength(0); + await quiet(); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + }); + it("reconstructs a durable retry deadline and runs only when it is due", async () => { + const binding = await seed(); await complete(binding); + const due = new Date(Date.now() + 60_000); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", nextAttemptAt: due }) + .where(eq(nativeRunFinalizations.runId, binding.runId)); + await start(); + expect(coordinator!.nextWakeAt()).toBe(due.getTime()); + await vi.advanceTimersByTimeAsync(30_000); + expect(await successful(binding.runId)).toHaveLength(0); + worker!.stop(); await coordinator!.stop(); await start(); + expect(coordinator!.nextWakeAt()).toBe(due.getTime()); + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + }); + it("waits quietly for the actual local owner to release", async () => { + const binding = await seed(); await complete(binding); + await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: binding.runId }, async () => { + await start(); await flush(); await quiet(); + expect(await successful(binding.runId)).toHaveLength(0); + }); + await flush(); expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + }); + it("joins the actual copyback callback during shutdown", async () => { + const binding = await seed(); await complete(binding); + let enter!: () => void, release!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const stat = fs.stat; + vi.spyOn(fs, "stat").mockImplementation(async (...args) => { + if (args[0] === root) { enter(); await held; } + return stat(...args); + }); + const starting = start(); + try { + await entered; + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await delay(20); + expect(stopped).toBe(false); + expect(await successful(binding.runId)).toHaveLength(0); + release(); await starting; await stopping; + expect(stopped).toBe(true); expect(await successful(binding.runId)).toHaveLength(1); + await quiet(); + } finally { release(); await starting; } + }); + it("retains an unverifiable owner without recurring SQL or physical writes", async () => { + const binding = await seed(); await complete(binding); + const owner = { token: randomUUID(), hostname: "another-controller", pid: 1234, processStartedAt: new Date().toISOString() }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, binding.runId)); + await start(); await flush(); await quiet(); + expect(await successful(binding.runId)).toHaveLength(0); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, binding.runId)))[0].runnerProfileJson?.nativeWorkspaceFinalizationOwner).toEqual(owner); + + await db.transaction(async tx => { await notifyNativeFinalizationWork(tx, binding); }); + await flush(); await quiet(); + expect(await successful(binding.runId)).toHaveLength(0); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, binding.issueId)); + await resumeNativeWorkspaceAfterOwnerStop({ ...binding, db, actionId: action.id, ownerToken: owner.token, + controllerAndCopybackStopped: true, stopEvidence: "The platform confirms this exact deployment and all copyback processes stopped.", actorId: "operator" }); + await flush(); expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + }); + + it("retries a concrete Cloud stop verification and disarms after verified copyback", async () => { + const binding = await seed(); await complete(binding); + const deployment = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "old" }; + const owner = { token: randomUUID(), hostname: "another-controller", pid: 1234, processStartedAt: new Date().toISOString(), worker: deployment }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, binding.runId)); + const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValueOnce({ status: "pending" }) + .mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", + sub: "stack", runId: binding.runId, ownerToken: owner.token, worker: deployment, stopped: true, iat: 1, exp: 61 } }); + await start(); + expect(probe).toHaveBeenCalledOnce(); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(15_000); + expect(probe).toHaveBeenCalledOnce(); expect(await successful(binding.runId)).toHaveLength(0); + await vi.advanceTimersByTimeAsync(15_000); await flush(); + expect(probe).toHaveBeenCalledTimes(2); expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + }); + + it("stays quiet after a task binding changes and wakes when the task is restored", async () => { + const binding = await seed(); await complete(binding); + const deployment = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "old" }; + const owner = { token: randomUUID(), hostname: "another-controller", pid: 1234, processStartedAt: new Date().toISOString(), worker: deployment }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, binding.runId)); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, binding.issueId)); + const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { + v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: binding.runId, + ownerToken: owner.token, worker: deployment, stopped: true, iat: 1, exp: 61, + } }); + await start(); await flush(); await quiet(); + expect(probe).toHaveBeenCalledOnce(); expect(await successful(binding.runId)).toHaveLength(0); + await issueService(db).update(binding.issueId, { status: "in_progress" }); + await flush(); expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + }); + it("records a failed validation once, honors its saved retry, and settles after repair", async () => { + const binding = await seed(); + await db.update(workspaceOperations).set({ cwd: join(root, "absent") }).where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + await complete(binding); await start(); + const [failed] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, binding.runId)); + expect(failed.phase).toBe("retryable_failure"); expect(failed.nextAttemptAt).not.toBeNull(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator!.nextWakeAt()).toBe(failed.nextAttemptAt!.getTime()); + }); + expect(await successful(binding.runId)).toHaveLength(0); + await db.update(workspaceOperations).set({ cwd: root }).where(eq(workspaceOperations.heartbeatRunId, binding.runId)); + await vi.advanceTimersByTimeAsync(Math.max(0, failed.nextAttemptAt!.getTime() - Date.now())); await flush(); + expect(await successful(binding.runId)).toHaveLength(1); await quiet(); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 5b79c9d6dc5..3747c5275ed 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -118,6 +118,7 @@ const { registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerRetainedNativeSessionCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), + registerNativeWorkspaceFinalizationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), registerNativeStatusReassessmentWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeReviewResponseWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeResponsePresentationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), @@ -574,6 +575,17 @@ describe("startServer feedback export wiring", () => { } }); + it.each([false, true])("registers workspace finalization with its shutdown subscription only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerNativeWorkspaceFinalizationWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) { + expect(heartbeatServiceMock.registerNativeWorkspaceFinalizationWork.mock.calls[0][1]()).toBe(true); + const worker = heartbeatServiceMock.registerNativeWorkspaceFinalizationWork.mock.results[0].value; + expect(fakeServer.on).toHaveBeenCalledWith("close", worker.stop); + } + }); + it.each([false, true])("registers retained cleanup with its shutdown subscription only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -1046,7 +1058,7 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { @@ -1202,19 +1214,21 @@ describe("startServer feedback export wiring", () => { } }); - it("becomes ready while retained cleanup is held but joins it at shutdown", async () => { + it.each(["retained cleanup", "workspace finalization"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); let releaseCleanup!: () => void; const ready = new Promise(resolve => { releaseCleanup = resolve; }); const stop = vi.fn(); - heartbeatServiceMock.registerRetainedNativeSessionCleanupWork.mockReturnValueOnce({ ready, stop }); + const register = owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork + : heartbeatServiceMock.registerNativeWorkspaceFinalizationWork; + register.mockReturnValueOnce({ ready, stop }); const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(() => 1 as unknown as ReturnType); let started = false; const starting = startServer().then(server => { started = true; return server; }); let stopping: Promise | undefined; let stopped = false; try { - await vi.waitFor(() => expect(heartbeatServiceMock.registerRetainedNativeSessionCleanupWork).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(register).toHaveBeenCalledOnce()); await settleStartupHousekeeping(); expect(started).toBe(true); expect(getStartupRecoveryState().phase).toBe("ready"); diff --git a/server/src/index.ts b/server/src/index.ts index 864edbeb808..ef3038a3288 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1404,6 +1404,7 @@ async function startServerWithDatabaseTeardown( }); let unregisterTerminalWorkspaceProbe: (() => void) | undefined; let unsubscribeNativeResponsePresentation: (() => void) | undefined; + let unsubscribeNativeWorkspaceFinalization: (() => void) | undefined; let unsubscribeRetainedNativeSessionCleanup: (() => void) | undefined; let unsubscribeNativeWorkspaceSyncRetention: (() => void) | undefined; const fastResponses = app.locals.fastResponses as ReturnType; @@ -1826,6 +1827,14 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + if (config.heartbeatSchedulerEnabled) { + const worker = heartbeat.registerNativeWorkspaceFinalizationWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); + unsubscribeNativeWorkspaceFinalization = worker.stop; + server.on("close", worker.stop); + + trackHeartbeatSchedulerWork(worker.ready); + } if (config.heartbeatSchedulerEnabled) { const worker = heartbeat.registerRetainedNativeSessionCleanupWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); @@ -1986,7 +1995,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. runPeriodicRecovery(() => heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); @@ -2183,6 +2192,7 @@ async function startServerWithDatabaseTeardown( await skillSourceSync.stop(); unsubscribeChatCompletions(); unsubscribeNativeResponsePresentation?.(); + unsubscribeNativeWorkspaceFinalization?.(); unsubscribeRetainedNativeSessionCleanup?.(); await deliveryWork.stop(); await workspaceRecoveryWork.stop(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index eaf9579001d..9c8c9fe6e48 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 = { + nativeWorkspaceFinalization: "native-workspace-finalization", retainedNativeSessionCleanup: "retained-native-session-cleanup", nativeStatusReassessment: "native-status-reassessment", nativeReviewResponse: "native-review-response", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 05b54143118..a1333fdc22b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1024,6 +1024,7 @@ export function heartbeatService( drainRunningRunsForShutdown, registerScheduledRetryCancellationWork, registerRetainedNativeSessionCleanupWork, + registerNativeWorkspaceFinalizationWork, } = createHeartbeatRecovery(db, { pluginWorkerManager: options.pluginWorkerManager, enterShutdown: () => { shutdownInProgress = true; }, @@ -4093,6 +4094,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeWorkspaceFinalizationWork: ( + coordinator: Parameters[0], canRun: () => boolean, + ) => registerNativeWorkspaceFinalizationWork(coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerRetainedNativeSessionCleanupWork: ( coordinator: Parameters[0], canRun: () => boolean, ) => registerRetainedNativeSessionCleanupWork(coordinator, () => canRun() && readTaskDrain(new Date()) === null), diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 37628a4ca91..4e755f92582 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,3 +1,5 @@ +import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js"; +import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js"; import { retainedNativeSessionCleanupWork } from "../native-runtime/retained-native-session-cleanup-work.js"; import { reconcileRetainedNativeSessionCleanups } from "../native-runtime/retained-native-session-cleanup.js"; import { createScheduledRetryCancellationRecovery } from "./scheduled-retry-cancellation.js"; @@ -1930,6 +1932,17 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD return operation; } const retainedNativeCleanup = retainedNativeSessionCleanupWork(db, cleanupRetainedNativeSession); + const nativeWorkspaceFinalization = nativeWorkspaceFinalizationWork(db, { + environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, + onWorkspaceSettled: settleRecoveredNativeWorkspace, + attempt: input => { + const operation = attemptNativeWorkspaceFinalization(input); + const tracked = operation.then(() => undefined, () => undefined); + activeRunExecutionPromises.add(tracked); + void tracked.then(() => activeRunExecutionPromises.delete(tracked)); + return operation; + }, + }); function scheduleRetainedNativeSessionCleanup() { // The per-database sweep joins startup and periodic callers. One bounded @@ -1954,7 +1967,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1979,6 +1992,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipNativeResponsePresentation: opts?.skipNativeResponsePresentation, skipNativeReviewResponse: opts?.skipNativeReviewResponse, skipNativeStatusReassessment: opts?.skipNativeStatusReassessment, + skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, @@ -2592,6 +2606,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD return { registerRetainedNativeSessionCleanupWork: retainedNativeCleanup.register, + registerNativeWorkspaceFinalizationWork: nativeWorkspaceFinalization.register, registerScheduledRetryCancellationWork: scheduledRetryCancellation.register, sweepPendingCleanupLeases, markNativeOwnershipUnverified, diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 46f531d4712..6e034f42303 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js"; @@ -626,6 +627,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDirectoryCleanup(tx); await notifyInstructionCollection(tx); await notifyCancelledWorkspacePreservation(tx); + await notifyNativeWorkspaceFinalization(tx); } if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx); await notifyToolConnectionHealth(tx, "*"); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 8deb995df11..c492a313e6d 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; @@ -11502,6 +11503,7 @@ export function issueService(db: Db) { if (patch.status !== undefined || patch.statusVersion !== undefined || patch.executionState !== undefined || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || blockedByIssueIds !== undefined) { + await notifyNativeWorkspaceFinalization(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeStatusReassessment(tx, { companyId: receiptExisting.companyId, issueId: id }); } const updated = await tx diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index 2740baed7b4..29b5479b571 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; @@ -8,6 +9,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { if (scope.issueId || scope.runId) { + await notifyNativeWorkspaceFinalization(transaction, scope); await notifyRetainedNativeSessionCleanup(transaction, scope); await notifyNativeStatusReassessment(transaction, scope); await notifyNativeResponsePresentation(transaction, scope); diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index f8538648ff0..ad0d46bf05b 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -37,7 +37,7 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueService } from "../issues.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js"; -import { resumeNativeWorkspaceFinalization } from "./native-workspace-finalizer.js"; +import { resumeNativeWorkspaceFinalization, recordNativeWorkspaceFinalizationFailure } from "./native-workspace-finalizer.js"; import { dismissObsoleteNativePolicyReviews } from "./obsolete-policy-reviews.js"; import { cleanupNativeWorkspaceSync, @@ -387,6 +387,7 @@ type NativeReconciliationOptions = { skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; + skipNativeWorkspaceFinalization?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -619,13 +620,18 @@ async function reconcileNativeFinalizationsPass( const barrier = await db.select({ status: workspaceOperations.status }) .from(workspaceOperations) .where(and( + eq(workspaceOperations.companyId, row.companyId), + eq(workspaceOperations.issueId, row.issueId), eq(workspaceOperations.heartbeatRunId, row.runId), eq(workspaceOperations.phase, "workspace_finalize"), )) - .orderBy(desc(workspaceOperations.createdAt)) + + + .orderBy(desc(sql`${workspaceOperations.status} = 'succeeded'`), desc(workspaceOperations.createdAt)) .limit(1) .then((entries) => entries[0] ?? null); if (!barrier || barrier.status !== "succeeded") { + if (options.skipNativeWorkspaceFinalization) continue; const recoveryDecision = resolveNativeReconciliationStatus({ facts: { workspaceOperationPending: true }, priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, @@ -645,23 +651,7 @@ async function reconcileNativeFinalizationsPass( const workspaceFinalizeStatus = operation.status === "succeeded" ? "succeeded" : "failed"; if (workspaceFinalizeStatus === "failed") { - const unrecoverable = operation.stderrExcerpt?.includes( - "workspace_sync_out_unrecoverable", - ); - const pluginFailure = operation.stderrExcerpt?.trim(); - const pluginRepair = pluginFailure === "plugin_workspace_branch_mismatch" || pluginFailure === "plugin_workspace_owner_mismatch"; - const failure = await recordNativeFinalizationFailure({ - db, - runId: row.runId, - error: new Error( - pluginRepair ? `native_${pluginFailure}` : unrecoverable - ? "native_workspace_sync_out_unrecoverable" - : "native_workspace_sync_out_failed", - ), - projectRunStatus: true, - failureScope: "workspace", - permanent: unrecoverable, - }); + const failure = await recordNativeWorkspaceFinalizationFailure(db, row.runId, operation); results.push({ ...failure, reconciliationAction: "resume_workspace_operation" as const, 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 1f911a854b3..36077aa71a6 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 { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; @@ -391,6 +392,7 @@ export class NativeRunCoordinatorStore { return { disposition: "duplicate" as const, resultId: existing.id }; } + await notifyNativeWorkspaceFinalization(tx, this.#binding); const [inserted] = await tx .insert(nativeRunResults) .values({ diff --git a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts index 60db370812e..3226c359f3a 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization, releaseNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { beginRunPhase } from "../execution-preparation-timing.js"; import { logActivity } from "../activity-log.js"; @@ -29,6 +30,7 @@ export async function tryLockNativeWorkspaceFinalization(tx: Db, input: { compan } const activeOwnerTokens = new Set(); +const activeRunOwners = new Map(); // Positive same-boot evidence, retained only when durable cleanup could not be // confirmed. Absence from the active set alone never proves another token joined. const joinedOwnerTokens = new Set(); @@ -40,6 +42,13 @@ type Owner = { processStartedAt: string; worker?: WorkspaceWorker; }; +export type NativeWorkspaceFinalizationWaitReason = + | "busy" + | "owner_active" + | "platform_stop_pending" + | "owner_unverified" + | "binding_changed"; + export type NativeWorkspaceFinalizationOwnership = { token: string; /** Recheck both the physical lock connection and the durable owner before publishing. */ @@ -72,7 +81,7 @@ function readOwner(value: unknown): Owner | null { export async function withNativeWorkspaceFinalizationOwnership( input: { db: Db; companyId: string; runId: string; retainedRepairActionId?: string }, action: (ownership: NativeWorkspaceFinalizationOwnership) => Promise, -): Promise<{ acquired: false; reason?: "owner_unverified" } | { acquired: true; value: T }> { +): Promise<{ acquired: false; reason: NativeWorkspaceFinalizationWaitReason } | { acquired: true; value: T }> { const identity = await currentNativeControllerIdentity(); const owner: Owner = { token: randomUUID(), controllerBootId: identity.bootId, hostname: os.hostname(), pid: identity.pid, @@ -81,6 +90,7 @@ export async function withNativeWorkspaceFinalizationOwnership( const scope = and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), input.retainedRepairActionId ? inArray(heartbeatRuns.runtimeMode, ["native", "legacy"]) : eq(heartbeatRuns.runtimeMode, "native")); const owned = and(scope, sql`${heartbeatRuns.runnerProfileJson}->${OWNER_KEY}->>'token' = ${owner.token}`); + const runKey = nativeWorkspaceFinalizationLockKey(input.companyId, input.runId); let claimed = false; let actionPromise: Promise | undefined; const finishAcquisition = beginRunPhase("execution.workspace_ownership_wait"); @@ -88,7 +98,10 @@ export async function withNativeWorkspaceFinalizationOwnership( return await withDedicatedDbConnection(input.db, (dedicated) => dedicated.transaction(async (lock) => { const rows = await lock.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(${nativeWorkspaceFinalizationLockKey(input.companyId, input.runId)}, 0)) as acquired, (select setting::bigint from pg_settings where name = 'idle_in_transaction_session_timeout') as idle_timeout_ms`); - if (rows[0]?.acquired !== true) return { acquired: false } as const; + if (rows[0]?.acquired !== true) return { + acquired: false, + reason: !input.retainedRepairActionId && activeRunOwners.get(runKey)?.db === input.db ? "owner_active" : "busy", + } as const; return withWorkspaceLockKeepalive(lock as unknown as Db, Number(rows[0].idle_timeout_ms), async () => { // Reads and durable progress use the ordinary pool, not the lock transaction. const [run] = await input.db.select({ profile: heartbeatRuns.runnerProfileJson, @@ -130,6 +143,7 @@ export async function withNativeWorkspaceFinalizationOwnership( }) : null; }; let platformProof: WorkspaceWorkerStopProof | null = null; + let platformStopPending = false; if (rawPrior != null) { const requireStopVerification = async () => { if (input.retainedRepairActionId) return { acquired: false, reason: "owner_unverified" } as const; @@ -150,7 +164,7 @@ export async function withNativeWorkspaceFinalizationOwnership( sql`(${issueRecoveryActions.nextAction} is distinct from ${nextAction} or ${issueRecoveryActions.evidence}#>>'{workspaceOwnerRecovery,state}' is distinct from 'operator_required')`, )); } - return { acquired: false } as const; + return { acquired: false, reason: platformStopPending ? "platform_stop_pending" : "owner_unverified" } as const; } if (run.issueId) await recovery.upsertSourceScoped({ companyId: input.companyId, sourceIssueId: run.issueId, @@ -161,11 +175,20 @@ export async function withNativeWorkspaceFinalizationOwnership( nextAction, wakePolicy: null, maxAttempts: 1, supersedeOnIdentityChange: true, }); - return { acquired: false } as const; + return { acquired: false, reason: platformStopPending ? "platform_stop_pending" : "owner_unverified" } as const; }; // The harness checks the exact owned deployment, including orphaned children. // A live deployment, unsigned answer or older receipt stays fenced. - if (prior?.worker) platformProof = await requestWorkspaceWorkerStopProof({ runId: input.runId, ownerToken: prior.token, worker: prior.worker }); + if (prior?.controllerBootId === identity.bootId && prior.pid === identity.pid + && activeOwnerTokens.has(prior.token)) { + return { acquired: false, reason: input.retainedRepairActionId ? "owner_unverified" + : activeRunOwners.get(runKey)?.db === input.db ? "owner_active" : "busy" } as const; + } + if (prior?.worker) { + const check = await requestWorkspaceWorkerStopProof({ runId: input.runId, ownerToken: prior.token, worker: prior.worker }); + platformProof = check.status === "verified" ? check.proof : null; + platformStopPending = check.status === "pending"; + } if (!prior) return requireStopVerification(); if (!platformProof && prior.hostname !== owner.hostname) return requireStopVerification(); const knownJoinedHere = prior.controllerBootId === identity.bootId @@ -194,6 +217,7 @@ export async function withNativeWorkspaceFinalizationOwnership( // only our token even if the server committed but its response was lost. const claimOwner = async (db: Db) => { activeOwnerTokens.add(owner.token); + activeRunOwners.set(runKey, { token: owner.token, db: input.db }); claimed = true; const [claimedRun] = await db.update(heartbeatRuns).set({ runnerProfileJson: sql`jsonb_set(coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb), array[${OWNER_KEY}], ${JSON.stringify(owner)}::jsonb) || ${JSON.stringify(platformProof ? { nativeWorkspaceOwnerStopProof: platformProof } : {})}::jsonb`, @@ -212,7 +236,14 @@ export async function withNativeWorkspaceFinalizationOwnership( if (claimedRun) automaticResultId = result.id; return claimedRun; }) : await claimOwner(input.db); - if (!claimedRun) return { acquired: false } as const; + if (!claimedRun) { + + + activeOwnerTokens.delete(owner.token); + if (activeRunOwners.get(runKey)?.token === owner.token) activeRunOwners.delete(runKey); + claimed = false; + return { acquired: false, reason: "binding_changed" } as const; + } if (platformProof && run.issueId) await logActivity(input.db, { companyId: input.companyId, actorType: "system", actorId: "workspace-recovery", action: "issue.workspace_owner_stop_verified", entityType: "issue", entityId: run.issueId, @@ -263,9 +294,11 @@ export async function withNativeWorkspaceFinalizationOwnership( await actionPromise?.catch(() => undefined); if (claimed) { activeOwnerTokens.delete(owner.token); + if (activeRunOwners.get(runKey)?.token === owner.token) activeRunOwners.delete(runKey); joinedOwnerTokens.add(owner.token); try { await input.db.transaction(async tx => { + await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, runId: input.runId }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.runId }); await tx.update(heartbeatRuns).set({ runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`, @@ -275,6 +308,8 @@ export async function withNativeWorkspaceFinalizationOwnership( } catch { // A reconnect in this exact controller boot can use the positive join // receipt. A new boot still needs durable barrier or operator proof. + } finally { + releaseNativeWorkspaceFinalization(input.db, { companyId: input.companyId, runId: input.runId }); } } } diff --git a/server/src/services/native-runtime/native-workspace-finalization-work.ts b/server/src/services/native-runtime/native-workspace-finalization-work.ts new file mode 100644 index 00000000000..7b36e747fb8 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-finalization-work.ts @@ -0,0 +1,104 @@ +import { and, asc, eq, gt, ne, notExists } from "drizzle-orm"; +import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { onNativeWorkspaceFinalizationRelease, type NativeWorkspaceFinalizationScope } from "../native-workspace-finalization-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js"; +import { attemptNativeWorkspaceFinalization, recordNativeWorkspaceFinalizationFailure } from "./native-workspace-finalizer.js"; + +type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; +type AttemptInput = Parameters[0]; + + +const OWNERSHIP_RECHECK_MS = 30_000; + +export function nativeWorkspaceFinalizationWork(db: Db, options: { + environmentRuntime?: AttemptInput["environmentRuntime"]; + pluginWorkerManager?: AttemptInput["pluginWorkerManager"]; + attempt?: typeof attemptNativeWorkspaceFinalization; + onWorkspaceSettled(input: { runId: string; companyId: string; agentId: string; succeeded: boolean }): Promise; +}) { + function candidates(scope?: NativeWorkspaceFinalizationScope, after?: string) { + return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId))) + .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.runId, nativeRunFinalizations.runId), eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), + eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), eq(nativeRunResults.schemaStatus, "accepted"))) + .where(and(eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition(), + ne(nativeRunFinalizations.phase, "terminal_failure"), + notExists(db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, nativeRunFinalizations.companyId), + eq(workspaceOperations.heartbeatRunId, nativeRunFinalizations.runId), eq(workspaceOperations.issueId, nativeRunFinalizations.issueId), + eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded"), + ))), + scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, + scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined, + after ? gt(nativeRunFinalizations.runId, after) : undefined, + )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); + } + function entry(row: Awaited>[number]): Entry { + return { id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, + dueAt: Math.max(row.nextAttemptAt?.getTime() ?? 0, row.leaseOwner ? row.leaseExpiresAt?.getTime() ?? 0 : 0) }; + } + async function load(scope?: NativeWorkspaceFinalizationScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (!signal?.aborted && canRun()) { + const rows = await candidates(scope, after); + entries.push(...rows.map(entry)); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].id; + } + throw new Error("Native workspace finalization discovery stopped"); + } + async function check(current: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Native workspace finalization stopped"); + const scope = { companyId: current.companyId, runId: current.id }; + const [row] = await candidates(scope); + if (!row) return null; + const due = entry(row); + if (due.dueAt > Date.now()) return due; + if (signal?.aborted) throw new Error("Native workspace finalization stopped"); + const outcome = await (options.attempt ?? attemptNativeWorkspaceFinalization)({ + db, runId: current.id, environmentRuntime: options.environmentRuntime, pluginWorkerManager: options.pluginWorkerManager, + }); + if (outcome.kind === "retry") return { ...due, dueAt: outcome.retryAt.getTime() }; + if (outcome.kind === "waiting") { + return outcome.reason === "busy" || outcome.reason === "platform_stop_pending" + ? { ...due, dueAt: Date.now() + OWNERSHIP_RECHECK_MS } : null; + } + if (outcome.kind !== "completed" || outcome.operation.status === "succeeded") return null; + const failure = await recordNativeWorkspaceFinalizationFailure(db, current.id, outcome.operation); + if (failure.phase === "terminal_failure") { + await options.onWorkspaceSettled({ runId: row.id, companyId: row.companyId, agentId: row.agentId, succeeded: false }); + return null; + } + const [remaining] = await candidates(scope); + return remaining ? entry(remaining) : null; + } + function register(coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise = async () => true) { + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeWorkspaceFinalization, batchSize: 1, + decodeScope: (hint: string): NativeWorkspaceFinalizationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (item: Entry, scope?: NativeWorkspaceFinalizationScope) => !scope || item.companyId === scope.companyId + && (!scope.issueId || item.issueId === scope.issueId) && (!scope.runId || item.id === scope.runId), + load: async (scope, signal) => await canRecover() ? load(scope, signal, canRun) : [], + check: async (item, signal) => { + if (!canRun()) throw new Error("Native workspace finalization paused"); + if (!await canRecover()) return null; + return check(item, signal); + }, + }, canRun); + const stop = onNativeWorkspaceFinalizationRelease(db, scope => worker.invalidate(JSON.stringify(scope))); + return { ...worker, stop }; + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-workspace-finalizer.test.ts b/server/src/services/native-runtime/native-workspace-finalizer.test.ts index 327515304d7..59270a2bb65 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.test.ts +++ b/server/src/services/native-runtime/native-workspace-finalizer.test.ts @@ -213,7 +213,7 @@ describe("native workspace finalization recovery", () => { it("releases only the verified owner and resumes the saved result without changing its disposition", async () => { const seed = await seedAbandonedCopyback(); const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId)); - expect(await withNativeWorkspaceFinalizationOwnership(seed, async () => "unexpected")).toEqual({ acquired: false }); + expect(await withNativeWorkspaceFinalizationOwnership(seed, async () => "unexpected")).toEqual({ acquired: false, reason: "owner_unverified" }); const observed: Array> = []; const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeStatusReassessment, () => { observed.push(db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns) @@ -337,7 +337,7 @@ describe("native workspace finalization recovery", () => { const worker = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "d" }; const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId)); await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceFinalizationOwner: { ...seed.owner, worker } } }).where(eq(heartbeatRuns.id, seed.runId)); - const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: seed.owner.token, worker, stopped: true, iat: 1, exp: 61 }); + const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: seed.owner.token, worker, stopped: true, iat: 1, exp: 61 } }); try { await reconcileNativeFinalizations(db, [seed.runId]); } finally { probe.mockRestore(); } } @@ -477,7 +477,7 @@ describe("native workspace finalization recovery", () => { await entered; await db.execute(sql`select pg_terminate_backend(pid) from pg_locks where locktype = 'advisory' and database = (select oid from pg_database where datname = current_database()) and pid <> pg_backend_pid()`); const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "must not run"); - expect(second).toEqual({ acquired: false }); + expect(second).toEqual({ acquired: false, reason: "busy" }); const [held] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); expect(held.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeTruthy(); } finally { finish(); } @@ -504,7 +504,7 @@ describe("native workspace finalization recovery", () => { await db.update(heartbeatRuns).set({ runnerProfileJson: { ...beforeReconnect.runnerProfileJson, nativeWorkspaceFinalizationOwner: unknown } }) .where(eq(heartbeatRuns.id, seeded.runId)); expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async () => "must not run")) - .toEqual({ acquired: false }); + .toEqual({ acquired: false, reason: "owner_unverified" }); } await db.update(heartbeatRuns).set({ runnerProfileJson: beforeReconnect.runnerProfileJson }).where(eq(heartbeatRuns.id, seeded.runId)); expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "resumed")) @@ -529,7 +529,7 @@ describe("native workspace finalization recovery", () => { token: randomUUID(), hostname: os.hostname(), pid: parent.pid, processStartedAt: startedAt, } } }).where(eq(heartbeatRuns.id, seeded.runId)); const work = vi.fn(async () => "second physical writer"); - expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false }); + expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false, reason: "owner_unverified" }); expect(work).not.toHaveBeenCalled(); const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId)); expect(action).toMatchObject({ ownerType: "board", cause: "native_workspace_finalization_owner_unverified", wakePolicy: null }); @@ -564,7 +564,7 @@ describe("native workspace finalization recovery", () => { await db.update(issueRecoveryActions).set({ fingerprint: nativeSha256({ runId: seed.runId, owner }), evidence: { runId: seed.runId, owner, preserved: "recovery evidence" } }).where(eq(issueRecoveryActions.id, seed.actionId)); const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId)); const proof = { v: 1 as const, iss: "paperclip-cloud" as const, aud: "paperclip-workspace-worker-stop/v1" as const, sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true as const, iat: 1, exp: 61 }; - const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue(proof); + const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof }); try { const work = vi.fn(async () => { const [recovery] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, seed.actionId)); @@ -590,18 +590,18 @@ describe("native workspace finalization recovery", () => { const worker = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "d" }; const owner = { ...seed.owner, worker }; await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seed.runId)); - const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue(null); + const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "pending" }); const work = vi.fn(); try { - expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false }); + expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false, reason: "platform_stop_pending" }); const [recovery] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, seed.issueId), eq(issueRecoveryActions.fingerprint, nativeSha256({ runId: seed.runId, owner })))); expect(recovery.evidence?.workspaceOwnerRecovery).toEqual({ state: "operator_required" }); const replacement = { ...owner, token: randomUUID() }; probe.mockImplementation(async () => { await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: replacement } }).where(eq(heartbeatRuns.id, seed.runId)); - return { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 }; + return { status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } }; }); - expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false }); + expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false, reason: "binding_changed" }); expect(work).not.toHaveBeenCalled(); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId)); expect(run.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toEqual(replacement); @@ -635,11 +635,11 @@ describe("native workspace finalization recovery", () => { createdAt: new Date(run.createdAt.getTime() + (changed === "newer run" ? 60_000 : -60_000)), }); } - return { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 }; + return { status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } }; }); const copyback = vi.fn(async () => fs.writeFile(file, "old sandbox work")); try { - expect(await withNativeWorkspaceFinalizationOwnership(seed, copyback)).toEqual({ acquired: false }); + expect(await withNativeWorkspaceFinalizationOwnership(seed, copyback)).toEqual({ acquired: false, reason: "binding_changed" }); expect(copyback).not.toHaveBeenCalled(); expect(await fs.readFile(file, "utf8")).toBe("new task work"); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId)); @@ -653,7 +653,7 @@ describe("native workspace finalization recovery", () => { const worker = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "d" }; const owner = { ...seed.owner, worker }; await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seed.runId)); - const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 }); + const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } }); const write = vi.fn(); try { await expect(withNativeWorkspaceFinalizationOwnership(seed, async ownership => { @@ -677,7 +677,7 @@ describe("native workspace finalization recovery", () => { ...(changed === "closed" ? { status: "done" } : { executionRunId: null }), }).where(eq(issues.id, seed.issueId)); await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seed.runId)); - const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 }); + const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } }); try { await reconcileNativeFinalizations(db, [seed.runId]); expect(probe).toHaveBeenCalledOnce(); @@ -692,7 +692,7 @@ describe("native workspace finalization recovery", () => { const owner = { token: randomUUID(), hostname: "other-controller.invalid", pid: 42, processStartedAt: new Date().toISOString() }; await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seeded.runId)); const work = vi.fn(); - expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false }); + expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false, reason: "owner_unverified" }); expect(work).not.toHaveBeenCalled(); const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId)); expect(action).toMatchObject({ ownerType: "board", cause: "native_workspace_finalization_owner_unverified", wakePolicy: null, evidence: { runId: seeded.runId, owner } }); diff --git a/server/src/services/native-runtime/native-workspace-finalizer.ts b/server/src/services/native-runtime/native-workspace-finalizer.ts index 0607284b972..ef8f6d09d6c 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.ts +++ b/server/src/services/native-runtime/native-workspace-finalizer.ts @@ -1,3 +1,4 @@ +import { recordNativeFinalizationFailure } from "./native-run-finalizer.js"; import { inspectPluginWorkspaceForFinalization } from "./native-plugin-workspace-finalization.js"; import type { WorkspaceWorkers } from "../plugin-workspaces.js"; import { restoreNativeWorkspaceBestEffort } from "./native-workspace-best-effort.js"; @@ -49,7 +50,7 @@ function workspaceSyncFailure( * older runs retain the local workspace validation path. The caller owns the * durable retry and failure policy. */ -export async function resumeNativeWorkspaceFinalization(input: { +export async function attemptNativeWorkspaceFinalization(input: { db: Db; runId: string; environmentRuntime?: EnvironmentRuntimeService; @@ -95,8 +96,10 @@ export async function resumeNativeWorkspaceFinalization(input: { if (!admission || admission.resultId !== bound.resultId) { throw new Error("native_workspace_finalization_binding_missing"); } - if (admission.phase === "terminal_failure" - || (admission.nextAttemptAt && admission.nextAttemptAt > new Date())) return null; + if (admission.phase === "terminal_failure") return { kind: "inactive" } as const; + if (admission.nextAttemptAt && admission.nextAttemptAt > new Date()) { + return { kind: "retry", retryAt: admission.nextAttemptAt } as const; + } const successful = await input.db.select().from(workspaceOperations).where(and( eq(workspaceOperations.companyId, bound.companyId), eq(workspaceOperations.heartbeatRunId, input.runId), @@ -105,7 +108,7 @@ export async function resumeNativeWorkspaceFinalization(input: { eq(workspaceOperations.status, "succeeded"), )).orderBy(desc(workspaceOperations.createdAt)).limit(1).then((rows) => rows[0] ?? null); // A stale failure from a pre-fencing controller cannot invalidate exported work. - if (successful) return successful; + if (successful) return { kind: "completed", operation: successful } as const; const previous = await input.db.select().from(workspaceOperations).where(and( eq(workspaceOperations.companyId, bound.companyId), eq(workspaceOperations.heartbeatRunId, input.runId), @@ -140,7 +143,7 @@ export async function resumeNativeWorkspaceFinalization(input: { executionWorkspaceId: workspace?.id ?? null, issueId: bound.issueId, }); - return recorder.recordOperation({ + const operation = await recorder.recordOperation({ phase: "workspace_finalize", cwd, metadata: { @@ -301,6 +304,26 @@ export async function resumeNativeWorkspaceFinalization(input: { }; }, }); + return { kind: "completed", operation } as const; + }); + return owned.acquired ? owned.value : { kind: "waiting", reason: owned.reason } as const; +} + + + +export async function resumeNativeWorkspaceFinalization(input: Parameters[0]) { + const result = await attemptNativeWorkspaceFinalization(input); + return result.kind === "completed" ? result.operation : null; +} + +export function recordNativeWorkspaceFinalizationFailure(db: Db, runId: string, operation: { stderrExcerpt: string | null }) { + const unrecoverable = operation.stderrExcerpt?.includes("workspace_sync_out_unrecoverable"); + const pluginFailure = operation.stderrExcerpt?.trim(); + const pluginRepair = pluginFailure === "plugin_workspace_branch_mismatch" || pluginFailure === "plugin_workspace_owner_mismatch"; + return recordNativeFinalizationFailure({ + db, runId, + error: new Error(pluginRepair ? `native_${pluginFailure}` : unrecoverable + ? "native_workspace_sync_out_unrecoverable" : "native_workspace_sync_out_failed"), + projectRunStatus: true, failureScope: "workspace", permanent: unrecoverable, }); - return owned.acquired ? owned.value : null; } diff --git a/server/src/services/native-runtime/native-workspace-owner-recovery.ts b/server/src/services/native-runtime/native-workspace-owner-recovery.ts index ae122b5edf7..59dc5a43db0 100644 --- a/server/src/services/native-runtime/native-workspace-owner-recovery.ts +++ b/server/src/services/native-runtime/native-workspace-owner-recovery.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import os from "node:os"; @@ -66,6 +67,7 @@ export async function resumeNativeWorkspaceAfterOwnerStop(input: { const now = new Date(); const proof = { owner: prior, resultId: result.id, actorId: input.actorId, stopEvidence: confirmation.stopEvidence, confirmedAt: now.toISOString() }; + await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); await tx.update(heartbeatRuns).set({ runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - ${OWNER_KEY}`, updatedAt: now, diff --git a/server/src/services/native-runtime/native-workspace-platform-proof.test.ts b/server/src/services/native-runtime/native-workspace-platform-proof.test.ts index fb3297ca80b..f0c72b53a2c 100644 --- a/server/src/services/native-runtime/native-workspace-platform-proof.test.ts +++ b/server/src/services/native-runtime/native-workspace-platform-proof.test.ts @@ -22,10 +22,30 @@ describe("workspace platform stop proof", () => { it("fails closed on live, unknown, unreachable and unsigned platform responses", async () => { for (const response of [{ stopped: false }, { stopped: true }, { receipt: "forged" }]) { const fetchImpl = vi.fn().mockResolvedValue(Response.json(response)); - expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl })).toBeNull(); + expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl })).toEqual({ status: "pending" }); } - expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockRejectedValue(new Error("offline")) })).toBeNull(); - expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockResolvedValue(Response.json({ receipt: receipt() })) })).toMatchObject({ stopped: true }); + expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockRejectedValue(new Error("offline")) })).toEqual({ status: "pending" }); + expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockResolvedValue(Response.json({ receipt: receipt() })) })).toMatchObject({ status: "verified", proof: { stopped: true } }); + }); + it("does not schedule platform verification without a supported authenticated endpoint", async () => { + for (const unavailable of [ + {}, + { ...env, PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "" }, + { ...env, PAPERCLIP_CLOUD_API_ORIGIN: "http://cloud.example" }, + { ...env, PAPERCLIP_CLOUD_API_ORIGIN: "https://user:password@cloud.example" }, + { ...env, PAPERCLIP_CLOUD_API_ORIGIN: "not a URL" }, + ]) { + const fetchImpl = vi.fn(); + expect(await requestWorkspaceWorkerStopProof(binding, { env: unavailable, now, fetchImpl })) + .toEqual({ status: "unsupported" }); + expect(fetchImpl).not.toHaveBeenCalled(); + } + }); + it("keeps an authenticated but temporarily unavailable verification pending", async () => { + const fetchImpl = vi.fn().mockResolvedValue(new Response(null, { status: 503 })); + expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl })) + .toEqual({ status: "pending" }); + expect(fetchImpl).toHaveBeenCalledOnce(); }); it("records a complete platform identity, and does not guess legacy ownership", () => { expect(currentWorkspaceWorker({ RAILWAY_PROJECT_ID: "project", RAILWAY_ENVIRONMENT_ID: "environment", RAILWAY_SERVICE_ID: "service", RAILWAY_DEPLOYMENT_ID: "old-deployment" })).toEqual(worker); diff --git a/server/src/services/native-runtime/native-workspace-platform-proof.ts b/server/src/services/native-runtime/native-workspace-platform-proof.ts index 04b452d1d76..47f31174c43 100644 --- a/server/src/services/native-runtime/native-workspace-platform-proof.ts +++ b/server/src/services/native-runtime/native-workspace-platform-proof.ts @@ -43,23 +43,31 @@ export function verifyWorkspaceWorkerStopReceipt( return proof; } catch { return null; } } -/** Unsupported platforms, legacy receipts and unavailable evidence never authorize takeover. */ +export type WorkspaceWorkerStopCheck = + | { status: "verified"; proof: WorkspaceWorkerStopProof } + | { status: "pending" } + | { status: "unsupported" }; + + export async function requestWorkspaceWorkerStopProof(binding: WorkspaceStopBinding, options: { env?: NodeJS.ProcessEnv; fetchImpl?: typeof fetch; now?: Date; -} = {}): Promise { +} = {}): Promise { const env = options.env ?? process.env; const context = getCloudStackContext(env); const token = env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN?.trim(); - if (!context?.stackId || !context.cloudOrigin || !token || !validWorker(binding.worker)) return null; + if (!context?.stackId || !context.cloudOrigin || !token || !validWorker(binding.worker)) return { status: "unsupported" }; + let origin: URL; + try { origin = new URL(context.cloudOrigin); } catch { return { status: "unsupported" }; } + if (origin.protocol !== "https:" || origin.username || origin.password) return { status: "unsupported" }; try { - const origin = new URL(context.cloudOrigin); - if (origin.protocol !== "https:" || origin.username || origin.password) return null; const response = await (options.fetchImpl ?? fetch)(`${origin.origin}/v1/tenant/workspace-worker-stop`, { method: "POST", redirect: "error", headers: { authorization: `Bearer ${token}`, "x-paperclip-cloud-stack-id": context.stackId, "content-type": "application/json" }, body: JSON.stringify(binding), signal: AbortSignal.timeout(10_000), }); - if (!response.ok) return null; + if (!response.ok) return { status: "pending" }; const payload = await response.json() as { receipt?: unknown }; - return typeof payload.receipt === "string" ? verifyWorkspaceWorkerStopReceipt(payload.receipt, binding, env, options.now) : null; - } catch { return null; } + const proof = typeof payload?.receipt === "string" + ? verifyWorkspaceWorkerStopReceipt(payload.receipt, binding, env, options.now) : null; + return proof ? { status: "verified", proof } : { status: "pending" }; + } catch { return { status: "pending" }; } } diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index b8e371da2bf..9b58bea9cc8 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { createAgentIdentityRedactor } from "../agent-identity-redaction.js"; import { and, asc, eq, gt, or } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -408,6 +409,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { ignoredAttentionRequests: normalizationDiagnostics.ignoredAttentionRequests, }, } as unknown as Record; + await notifyNativeWorkspaceFinalization(tx, this.#binding); const inserted = await tx.insert(nativeRunResults).values({ companyId: this.#binding.companyId, issueId: this.#binding.issueId, diff --git a/server/src/services/native-workspace-finalization-notifications.ts b/server/src/services/native-workspace-finalization-notifications.ts new file mode 100644 index 00000000000..2f4b45d1bee --- /dev/null +++ b/server/src/services/native-workspace-finalization-notifications.ts @@ -0,0 +1,24 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeWorkspaceFinalizationScope = { companyId: string; runId?: string; issueId?: string }; + +export function notifyNativeWorkspaceFinalization(tx: object, scope?: NativeWorkspaceFinalizationScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeWorkspaceFinalization, + scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"); +} + + + +const listeners = new WeakMap void>>(); +export function onNativeWorkspaceFinalizationRelease(db: object, listener: (scope: NativeWorkspaceFinalizationScope) => void) { + let owned = listeners.get(db); + if (!owned) { owned = new Set(); listeners.set(db, owned); } + owned.add(listener); + return () => { + owned.delete(listener); + if (!owned.size && listeners.get(db) === owned) listeners.delete(db); + }; +} +export function releaseNativeWorkspaceFinalization(db: object, scope: NativeWorkspaceFinalizationScope) { + for (const listener of listeners.get(db) ?? []) listener(scope); +} diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index 60103c15fbd..e2e3b81aa52 100644 --- a/server/src/services/workspace-operations.ts +++ b/server/src/services/workspace-operations.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; @@ -480,6 +481,8 @@ export function workspaceOperationService(db: Db) { async function writeState(write: (database: Db) => Promise): Promise { if (recordInput.phase !== "workspace_finalize" || !input.heartbeatRunId) return write(db); return db.transaction(async tx => { + await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, + runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined }); return write(tx as unknown as Db); diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 1c3e976b091..74af69eda36 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -1079,11 +1079,11 @@ describe("automatic retained workspace recovery", () => { await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson, nativeWorkspaceFinalizationOwner: prior } }).where(eq(heartbeatRuns.id, f.runId)); // This is the verified platform boundary. Signature and identity validation // have independent coverage in native-workspace-platform-proof.test.ts. - const proof = vi.spyOn(workerProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ + const proof = vi.spyOn(workerProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "test-stack", stopped: true, iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 30, runId: f.runId, ownerToken: prior.token, worker, - }); + } }); try { await recoverRetainedWorkspace(f.input); expect(proof).toHaveBeenCalledWith({ runId: f.runId, ownerToken: prior.token, worker }); diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 20ca973c20b..e82cf5cc0e7 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -1,3 +1,4 @@ +import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { beginIdleTrackedWork } from "./task-admission.js"; @@ -247,6 +248,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ const next = { ...current!, revision: current!.revision + 1, phase: "reconcile_effects" as const, state: "complete" as const, nextAttemptAt: null }; const activity = await db.transaction(async tx => { + await notifyNativeWorkspaceFinalization(tx, input); await notifyNativeStatusReassessment(tx, input); await tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId, heartbeatRunId: input.runId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(),