From 27270e9bf8b7a294cf03f8b84f4857f1b1ff05c7 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 21:07:55 -0500 Subject: [PATCH] fix(server): wake interrupted retry cancellation from committed intent (#17667) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A parked scheduled retry has no executor to finish an interrupted Stop request. > - Its saved cancellation fence already records the work that remains. > - The orphan reaper currently scans for that work on every heartbeat, including idle periods. > - This PR registers the existing records with the shared delivery scheduler and wakes recovery after committed writes. > - Recovery keeps the existing cancellation guards and settlement behavior, then becomes quiet. ## Linked Issues or Issue Description Refs #17666, the merged preceding PR in the recurring-server-work migration stack. This PR targets master. Related cancellation work found in the dedup search: #12301 and #7340. This PR changes the trigger for interrupted parked-retry cancellation; it does not replace their cancellation or retry policies. **What happened?** The periodic orphan reaper queries scheduled retries with cancellation fences even when no cancellation needs recovery. Its first-20-row scan can also stop behind malformed evidence. **Expected behavior** Committed cancellation intent wakes bounded recovery. Failed work retains a retry. Startup reconstructs durable work. Empty or settled work performs no recurring SQL for this duty. **Steps to reproduce** 1. Start a server with heartbeat scheduling enabled and no interrupted retry cancellations. 2. Observe the orphan reaper repeatedly query for cancellation fences. 3. Cancel a scheduled retry and fail the final status write after its fence commits. Recovery must complete the existing Stop request without running a provider or dropping its metadata. **Paperclip version or commit** Current parent squash `7afee8f0622e273d10506b0febfb68f143d168b1`. Source acceptance `66107ae0a7ad85fd0476f91c4f8f98ae74a7a048`. **Deployment mode** Self-hosted server with the existing single-process transaction signal model. Local evidence uses PostgreSQL 17. ## What Changed - Add one domain-owned recovery worker over existing heartbeat run rows. No new table. It delegates to run control and preserves original cancellation, budget, and operator metadata. - Signal cancellation fences, lifecycle transitions, dispatch deferral/promotion/suppression, relevant settings changes, and deletion within their owning transactions. Wrap only the ordinary database fence in a short transaction; provider calls stay outside it. - Keep startup/manual recovery compatible and remove this duty from the periodic orphan sweep. Other reaper duties remain for separate PRs. - Extend the shared scoped worker to retain pending work with a memory-only admission retry while its owner is paused. Without pending work, it schedules nothing. Existing GitHub consumers retain their behavior. - Add a concurrent idempotent partial index and lifecycle/producer tests. Enable guarded disposable PostgreSQL URLs in affected test harnesses. - Reconstruct one hundred candidates per scope per scheduler pass. Resume from the saved scope cursor and retain fair twenty-item execution batches. Manual recovery starts cancellation before reading the full backlog. - Require the recovered browser reply to be saved before testing reload. Keep all original assertions and time limits. - Record the full producer audit, lifecycle limits, and failed attempts in `doc/plans/2026-10-25-scheduled-retry-cancellation-sleep-audit.md`. ## Verification Published head: `405e533394b26215e22d0452e9b0207d3f784a79`. Source acceptance: `a35fa43b98e2bd378d4a920d2775ce243654838b`. The final commit changes the audit only. - All 363 backend cases in ten complete owner and consumer files pass without skips. This includes actual Stop/retry producers, transaction settlement, interrupted-write recovery, pause/resume, shutdown, fair backlog processing and quiet after settlement. Seven separate fresh PostgreSQL 17 databases prevent fixture interference. - The two embedded-only backend fixtures temporarily use separate external databases on this IPC-exhausted host. Their fixture imports and native client teardown are restored byte for byte afterward. Assertions and timing limits are unchanged. CI must still pass with the original embedded fixtures. - Two paging regressions fail on the prior worker and pass on the fix. They cover cancellation before full reconstruction, a malformed first page, fair batches, a newly committed intent behind the cursor, and no later idle SQL. Per-candidate admission reads remain deliberate safety checks. - The repeated CI reload failure is reproduced with a temporary twelve-second fixture settlement delay. The trace retains the same five-second missing-reply assertion, even though a cleanup race masked it in the top-level error. Requiring exactly one saved reply from the resumed run passes under the identical delay. The fault is removed. All six browser journeys pass on the final source in 40.5 seconds, with zero retries and the original 120-second case and five-second reload limits. - Browser acceptance uses the canonical CLI/server/static-UI path and a fresh external database. Only fixture database-URI setup is temporarily adapted and then restored. This is not an unmodified embedded CI-environment reproduction. The six cases include the controlled case; repeated runs are not added to the distinct count. - Actual database build, server typecheck/build, module boundaries and exact-current migration order pass. The database source tree is unchanged from the verified 0405-to0406 upgrade: all 404 prior receipts and 300 companies stay intact, exactly one receipt is added, replay adds none, and the index is valid on upgraded and fresh databases. The five snapshots remain 0402 through 0406. - All earlier failures remain recorded in the audit. Prior head `48b3fcb83f305616df02a4337a42ec0867bb4b80` had exact-head Greptile 5/5 but failed the repeated browser assertion. It was not retried blindly. Its completed Ponytail review's backlog concern and old migration paragraph are addressed here. Old scores do not satisfy new-head gates. - Current-head CI passes. Exact-head Greptile and Ponytail both give 5/5, with no remaining actionable findings. These checks do not prove deployed instance or database suspension. ## Risks - Only supported in-process writers produce local transaction signals. Independent SQL writers or other replicas need a separate ingress contract. Startup reconstructs saved obligations. - A real failure retains bounded retries. Unknown commit outcomes retain the shared intent hold. Neither is discarded to report quiet. - Per-owner drain uses a memory-only admission recheck. The shared helper change affects existing consumers, which have focused regression coverage. - The partial index is concurrent and idempotent; it adds index storage. Each scoped load reads at most one hundred rows. The shared helper retains discovered pending IDs between passes, so a large unresolved backlog still uses memory. - Cloud alarms, idle-safety exemptions, and other reaper duties remain separate. Merge requires passing current-head CI and review gates. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, shell tools, and local PostgreSQL tests. The runtime does not expose a more precise model identifier or context-window size for this turn. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 33 ++ ...cheduled-retry-cancellation-sleep-audit.md | 347 +++++++++++++++ packages/db/src/client.ts | 1 + .../src/migrations/0406_zippy_war_machine.sql | 1 + ...{0401_snapshot.json => 0406_snapshot.json} | 277 +++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/heartbeat_runs.ts | 2 + .../issue-scheduled-retry-routes.test.ts | 399 +++++++++++++++++- .../server-startup-feedback-export.test.ts | 14 + server/src/index.ts | 5 +- .../run-dispatch/adapters/postgres.test.ts | 11 +- .../modules/run-dispatch/adapters/postgres.ts | 4 + server/src/services/agent-deletion.ts | 2 + server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + server/src/services/heartbeat.ts | 20 +- .../__tests__/run-control-fixtures.ts | 2 +- server/src/services/heartbeat/queue.ts | 2 + .../src/services/heartbeat/recovery.test.ts | 12 +- server/src/services/heartbeat/recovery.ts | 32 +- .../services/heartbeat/run-control.test.ts | 45 ++ server/src/services/heartbeat/run-control.ts | 18 +- .../src/services/heartbeat/run-lifecycle.ts | 11 +- .../heartbeat/scheduled-retry-cancellation.ts | 86 ++++ server/src/services/instance-settings.ts | 2 + .../native-cancellation-request.ts | 2 + ...eduled-retry-cancellation-notifications.ts | 7 + tests/e2e/legacy-failure-continuation.spec.ts | 8 + 28 files changed, 1297 insertions(+), 56 deletions(-) create mode 100644 doc/plans/2026-10-25-scheduled-retry-cancellation-sleep-audit.md create mode 100644 packages/db/src/migrations/0406_zippy_war_machine.sql rename packages/db/src/migrations/meta/{0401_snapshot.json => 0406_snapshot.json} (99%) create mode 100644 server/src/services/heartbeat/scheduled-retry-cancellation.ts create mode 100644 server/src/services/scheduled-retry-cancellation-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 c8313d5acb9..fb1924e10a2 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2502,3 +2502,36 @@ before worker settlement. The assertion now waits for settled idle tracking and the actual future deadline; all 37 affected lifecycle/publication/startup cases pass. The full failure remains in the audit; latest-head remote gates are required. No full-run success or stack completion is inferred from that focused repair. + + +## Slice 38 accepted; slice 39 retry cancellation + +[#17666](https://github.com/paperclipai/paperclip/pull/17666) at +`d8e1bd97f530fc7a5119549a35b871b7b3d9f737` has all 53 remote checks +passing, fresh Greptile 5/5, and its review finding resolved. The PR remains +draft and unmerged. Its earlier local failure and focused repair remain recorded. + +Slice 39 separates interrupted scheduled-retry cancellation from the periodic +orphan reaper. Existing run rows and transaction notifications drive bounded +recovery through the existing cancellation owner. No table or external alarm is +added. The producer audit includes dispatch deferral, budget fences, correlated +Stop, lifecycle transitions, settings, and deletion. See +[the owner audit](2026-10-25-scheduled-retry-cancellation-sleep-audit.md). +Thirty route tests and all 34 recovery tests pass. Shared-helper consumers pass +20 tests. Final combined owner acceptance and remote gates remain required. +Earlier #17490/#17572 CI failures, #17576/#17580 review findings, master +reconciliation, and every other original inventory duty remain outstanding. + +Final combined owner acceptance passed all 164 tests in five suites, with no +skips, in 41.14 seconds using separate dispatch and recovery databases. Together +with the 30 retry-route tests this covers 194 tests. Remote PR gates remain pending. + + +## Startup CI fixture repair + +Initial head bd490ee50b failed server shard 4/12: eleven tests in the startup +suite used a heartbeat test double without the new registration method. The +real worker and owner acceptance passed, but startup wiring coverage was missing +from the local selection. The double now exposes the method, and new assertions +verify scheduler enablement and periodic-reaper exclusion. All 48 startup tests +pass in 5.34 seconds. This is a test-only repair; fresh head CI/review is required. diff --git a/doc/plans/2026-10-25-scheduled-retry-cancellation-sleep-audit.md b/doc/plans/2026-10-25-scheduled-retry-cancellation-sleep-audit.md new file mode 100644 index 00000000000..3b9e4e39677 --- /dev/null +++ b/doc/plans/2026-10-25-scheduled-retry-cancellation-sleep-audit.md @@ -0,0 +1,347 @@ +# Scheduled retry cancellation sleep audit + +Slice 39, based on `d8e1bd97f530fc7a5119549a35b871b7b3d9f737` from +[#17666](https://github.com/paperclipai/paperclip/pull/17666). +Inspected remote master: `5a28f7acf8a8c86536c4f5c5369bfeb49a46c41a`. +This slice owns only interrupted cancellation of parked scheduled retries. +The other orphan-run recovery duties remain separate work. + +## Obligation and owner + +A scheduled retry has no active executor to finish a Stop request if the final +status write fails after the cancellation fence commits. The old orphan reaper +queried for these rows on every heartbeat. It inspected only the first 20 rows, +so malformed intents could also hide valid work behind that prefix. + +The source of truth remains `heartbeat_runs`: status `scheduled_retry`, an object +at `result_json.startupCancellation`, and a cancellation accepted by the existing +`readRunCancellation` parser with `expected: true`. No table is added. A correlated +Stop reservation without valid cancellation evidence is not an instruction to +cancel. The worker neither launches a provider nor replaces run-control policy. + +`heartbeat/scheduled-retry-cancellation.ts` owns candidate discovery and delegates +to the existing cancellation owner. Cancellation keeps its expected-status guard, +issue/wakeup settlement, budget markers, and original result metadata. Inventory +queries select only cancellation evidence. A selected candidate is read with its +full result before delegation. Every candidate is reread before execution. + +Startup and explicit manual reaping retain the compatibility sweep. The periodic +reaper skips this duty after its dedicated worker is registered. Registration +retains the heartbeat scheduler toggle, warm-standby and idle-drain behavior. + +## Producers and invalidation + +| Owner | Relevant write and signal | +| --- | --- | +| Run control | Ordinary cancellation fence, guarded handoff fence, and budget cancellation fanout signal before their writes. The ordinary fence gains a short transaction; no provider call is enclosed. Existing budget transaction ownership is reused. | +| Native cancellation request | Correlated Stop reservation signals its exact run. The worker still requires valid expected cancellation evidence. | +| Run lifecycle | Transitions from or to scheduled retry signal the exact run in the existing transaction. | +| Queue cancellation | Direct stale scheduled-retry cancellation signals its exact run. | +| Dispatch adapter | Promotion, gate suppression, and queued-to-scheduled cleanup deferral signal the exact run. Deferral can preserve an already-written cancellation intent. | +| Instance settings | Worktree execution enablement changes signal reconstruction. The worker reads this policy fresh; ordinary callers retain the existing cache. | +| Agent/company deletion | Company-scoped invalidation precedes deletion of run records. | + +New retry insertion paths in heartbeat retries, recovery service, native provider +capacity retry, and native safe replacement do not copy cancellation intents. +They create no work for this owner until a supported cancellation producer writes +its fence. Retained completed history and ordinary pending retries are quiet. + +The supported writer model is the existing single process using the instrumented +DB client. Independent SQL writers and other replicas do not produce these local +signals. This PR does not add cross-process ingress or claim those writers are +covered. Startup reconstruction recovers saved work after restart. + +## Shared scheduling and lifecycle + +The existing transaction signal boundary tracks intent before writes, including +outer commit/rollback and ambiguous outcome reconciliation. Signals are hints, +not a replacement for durable rows. No caller-managed counters are added. + +The shared scoped worker reconstructs candidates in keyset pages of 100 and +processes company-fair batches of 20. It retries concrete failures after 30 seconds, +with exponential backoff capped at five minutes. Settled work removes its deadline. +Malformed evidence is not actionable, and pagination continues past it. + +A small shared-helper change handles a per-owner admission gate that is closed +while the coordinator is otherwise running. Pending scopes and deadlines remain +in memory and get a 30-second admission recheck, without SQL. This avoids a tight +loop over an already-due deadline. Existing GitHub helper consumers were tested. +With no pending entries or dirty scopes there is no admission timer either. + +Worktree suppression returns no executable work. Its settings writer reactivates +the worker on enablement. Fresh reads propagate failures into recovery instead of +silently treating an unavailable setting as permanent disablement. Task drain uses +the synchronous admission gate so a temporary pause cannot discard durable work. +Shutdown joins active reads and does not start a cancellation after admission closes. + +## Migration + +`0406_zippy_war_machine.sql` creates a partial ID index for scheduled retries with an +object cancellation fence. It uses concurrent, idempotent creation and is listed +in the client's top-level migration set. It changes no rows or API contracts. +The migration was generated through `pnpm db:generate`. Journal entries and SQL +history remain intact; snapshots 0402 through 0406 are the retained newest five. +The initial unlanded draft used `0398_breezy_mongu.sql`; the later integration +sections retain its history and explain append-only regeneration. +Migration application, explicit replay, and migration validation passed. + +A rolled-back synthetic fixture with 10,000 completed rows and one candidate +used the new index: one row, two shared buffer hits, 0.013 ms execution. This is +local query-plan evidence, not a production performance or fleet savings claim. + +## Verification and failure record + +Local PostgreSQL 17 databases are dedicated test databases. The route, run-control, +dispatch, and recovery suites accept guarded external test URLs; embedded fallback +is unchanged. Never point these destructive fixture suites at a shared service DB. + +- Retry route suite: 30 passed, including 16 worker lifecycle cases. Covers quiet + history, real Stop with failed terminal write, lost response after commit, + dispatch cleanup deferral, metadata preservation, outer rollback/commit, + restart, drain/standby, actual task-drain release and expiry, worktree enablement, + concurrent intent, foreground race, shutdown, and fair paginated draining. +- Recovery module suite: all 34 passed on a fresh database. +- Existing shared-helper GitHub consumers: 20 passed. +- Workspace typecheck/build passed before the final test-only additions. Server + typecheck passed again with the final tests. Migration validation/replay passed. +- Initial owner acceptance was 138 passed and 26 skipped, not 164 passes. External + recovery support enabled those 26. Its first run had 33 passes and one fixture + failure: a second client still referenced the embedded connection. Updating it + to use the same external URL produced 34 passes. +- Initial worker selection had eight passes and one incorrect standby assertion. + Standby correctly retained the unresolved transaction hold. The corrected test + proves no passive SQL and eventual recovery without asserting away that hold. +- Removing run-control notifications made the committed-fence lost-response test + fail: the retry remained scheduled. Restoring the producer signal fixes it. +- A metadata regression test failed when recovery omitted the old full result + argument. Passing the candidate's full saved result preserves budget/operator + metadata and fixes the regression. +- A subsequent combined owner run had 95 passes and 69 failures in dispatch + fixture cleanup. Its shared database contained browser sessions from run-control; + the foreign key blocked issue deletion. Dispatch now has its own database for + acceptance. This was test-command isolation, not a production source repair. + The corrected combined acceptance result is recorded below when complete. + +## Limits and remaining work + +This proves the changed worker can become quiet and reactivate. It does not prove +host or database suspension, extend the Cloud alarm contract, or relax idle-safety +blockers. The scheduler cannot wake stopped compute by itself. Other run reaper +queries, scheduled retry dispatch, task recovery, and the original inventory remain +unfinished. At initial authoring, this slice and its parent stack were not authorized for merge. The later stack campaign authorizes an ordinary protected merge only after exact-current-head Greptile 5/5, clean CI, and resolution of required review findings. + +Final combined owner acceptance passed all 164 tests in five suites, with no +skips, in 41.14 seconds using separate dispatch and recovery databases. Together +with the 30 retry-route tests this covers 194 tests. Remote PR gates remain pending. + + +## Startup CI fixture repair + +Initial head bd490ee50b failed server shard 4/12: eleven tests in the startup +suite used a heartbeat test double without the new registration method. The +real worker and owner acceptance passed, but startup wiring coverage was missing +from the local selection. The double now exposes the method, and new assertions +verify scheduler enablement and periodic-reaper exclusion. All 48 startup tests +pass in 5.34 seconds. This is a test-only repair; fresh head CI/review is required. + +## Master integration and Stop attribution follow-up (2026-10-27) + +Integrated master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5`. A real PostgreSQL +regression held the foreground Stop after each cancellation fence and allowed an +independent scheduled-retry recovery owner to win terminal settlement. All three +user/board cases reproduced missing operator attribution. Both existing fence +transactions now persist `cancelledByActorType` and `cancelledByUserId`, so the +recovery winner retains the fields used to suppress automatic restart. No new +queue, table, or timer is introduced. Arbitrary result metadata is not copied +into the fence. + +All 49 run-control tests pass, including those three races. The integrated +recovery, dispatch, and scheduled-retry route suites pass on fresh disposable +PostgreSQL databases; their external-database fixtures apply migrations before +opening the test client. Production server typechecking passes after resolving +the overlapping upstream recovery fixture imports and connection variable. +The first targeted test command used an invalid project filter and ran no tests; +the corrected command established the failing baseline before the fix. These are +local behavior checks, not a full-repository or deployed-sleep claim. + + +## Current stack acceptance (2026-10-28) + +The current source acceptance ran at `e3a2472fe375f53158fd591d0eaf96ae8130474b`. +All ten complete test files passed: **362 cases, no skips**, in 72.12 seconds. +Counts are taken from the verbose passing case records, not from a name filter: + +| Complete file | Passing cases | +| --- | ---: | +| Scheduled-retry HTTP routes and worker | 30 | +| Heartbeat recovery | 45 | +| Heartbeat run control | 51 | +| PostgreSQL dispatch adapter | 69 | +| Heartbeat queue | 32 | +| Heartbeat run lifecycle | 31 | +| Native cancellation parser and admission | 32 | +| Native cancellation integration | 16 | +| Cancellation provenance | 5 | +| Server startup and feedback export | 51 | + +Routes, recovery, run control, dispatch, and queue each used a different fresh +disposable PostgreSQL database. They were migrated before starting the tests. +Lifecycle and native integration retained their real embedded PostgreSQL fixtures. +The earlier dispatch foreign-key cleanup failure remains recorded above; sharing +that database with run-control fixtures was not repeated. No tests were disabled +and no production behavior was changed to accommodate the fixtures. + +Database build, server typecheck, server build, module boundaries, migration order, +and all five fresh database migrations passed. The only draft migration owned +by this slice was regenerated as `0404_cynical_phil_sheldon.sql` after the verified +parent history. Its original partial-index definition is unchanged. It uses +concurrent, idempotent creation through the existing top-level migration path. +All prior SQL and journal entries, and all retained parent snapshots, are preserved; +the five newest snapshots are 0400 through 0404. No new table is added. + +A populated parent database upgraded from 0403 to 0404. All 402 existing migration +receipts remained identical; exactly one receipt was appended. Replaying appended +none. The scheduled-retry cancellation index is valid and ready on both that +upgraded database and a freshly migrated database. + +The latest normal parent integration is +`a1ae3be691e00cae4556c16f84a271ea6425864c`, carrying #17666 at +`836acef0370d42323fdfdbd6d62a805b2469f1d4`. The complete server, package, and UI +source trees and every owned source hash still match the tested revision. The +complete owned binary patch is identical against the corresponding parent tree. +The only carried changes are parent audits and unrelated onboarding smoke tests +and their feature map, with exact upstream bytes preserved. The parent startup +inspection correction is superseded by #17666's removal of that coordinator; +this slice does not restore the obsolete inspection. + +These results establish this worker's behavior and integration, not deployed host +or database suspension. Final publication still needs the actual landed parent +ancestry, then fresh current-head review and CI. Existing reviews on earlier heads +do not count for the eventual merge. + + +## Landed stale-review parent ancestry + +Parent #17664 landed as `8bc051245f77a47bee0f8ba45896b8bfa5922e0a`. Its entire +tree equals the already-adopted reviewed parent `0886a0229b1c4f68d2f582da86aa1f248e9124f3`. +A normal merge records the squash ancestry. Duplicate ancestry produced five +conflicts in the stack inventory, migration allowlist/journal, app wiring, and +startup inspection. The existing descendant content preserves the landed tree +and the later #17666 coordinator removal. After resolving those overlaps, the +**entire result tree** equals the previously validated local branch, not just +selected source hashes. The 0403 migration is therefore now in the actual master +ancestry before this slice publishes 0404. No production or test code changed. + +The required #17666 parent still must merge before this slice. Publishing the +stacked branch allows its independent CI and review to run while that parent +finishes; it does not authorize merging into the unmerged parent. + +## Current master integration after webhook recovery merged + +Parent #17666 merged as `7afee8f0622e273d10506b0febfb68f143d168b1`. +Master also gained the Coordinator workspace and prepared Runner changes. These +landed migrations 0404 and 0405 while this PR's prior 0404 was still unmerged. +Only this slice's unlanded migration artifacts were deferred. A normal merge +preserved every original runtime and schema edit, all unrelated upstream blobs, +and the exact parent migration tree. The repeated conflict in the shared stack +inventory retained the existing slice39 historical section; no runtime conflict +was resolved by choosing a side. The actual parent squash ancestry is recorded, +and its complete tree matches the tested normal parent composition. + +The existing partial index was then generated as +`0406_zippy_war_machine.sql`. Its concurrent, idempotent SQL bytes equal the +original owned migration. Every landed SQL file, prior journal entry, and retained +snapshot stays unchanged. The final snapshot set is 0402 through 0406. Four new +comment lines were removed to follow contributor rules; both files retain the +same complete non-comment TypeScript token streams. + +At `66107ae0a7ad85fd0476f91c4f8f98ae74a7a048`, all **362 cases in ten complete +files** pass, with no skipped cases, in 61.86 seconds. Seven separate fresh +PostgreSQL 17 databases prevent destructive fixture interference. Five fixtures +use their existing external URL paths. The two embedded-only lifecycle and native +cancellation fixtures temporarily used separate pre-migrated external databases +because this host cannot start more embedded clusters. Only their fixture imports +and native client teardown changed during that local run. All assertions and +original timing limits stayed intact. Both test files were restored byte for +byte afterward. This is a qualified external-database acceptance run, not an +unmodified embedded or CI-environment reproduction. Published CI keeps the +original embedded fixtures and must pass on the new head. + +Database build, server types/build, module boundaries, migration order, and all +seven fresh 0406 migration runs pass. A populated 0405 database retains all 404 +prior migration receipts and all 300 companies, adds exactly one receipt, and +adds none on replay. The partial index is valid and ready on upgraded and fresh +databases. Earlier failures, including the old head's unexplained decision-expiry +CI assertion and later successful single retry, remain in the evidence ledger. +The initial draft metadata equality probe and a short-SHA validation-command usage +error were rejected and corrected before acceptance; neither was a product pass. + +The previously green `9388f8c020d6f73d681cb1d183193f79754b324c` review and CI +are historical. The new published head requires fresh CI and exact-head Greptile +5/5. These results do not establish deployed host or database suspension. + +## Bounded reconstruction and durable browser recovery acceptance + +Current-head CI at `48b3fcb83f305616df02a4337a42ec0867bb4b80` failed the same +legacy `task_retry` reload assertion previously seen on the parent: an expected +reply was absent within the unchanged five-second reload check. The other five +legacy recovery journeys passed. The earlier parent retry had passed without +explaining the failure. This repeated failure was not retried blindly. + +The browser test checked a visible reply, task status Done, and one new run, but +none establishes that the canonical reply was saved. The fixture marks the task +Done before emitting its reply and returning its final result. Completion later +materializes the agent comment. A controlled temporary twelve-second delay +between the fixture's existing visible reply and final result reproduced the +same five-second missing-reply assertion. A fixture-directory cleanup race then +masked that assertion in the top-level failure; the retained Playwright trace +records the original reload assertion, and the cleanup failure remains recorded. +CI did not capture a database snapshot at reload, so this reproduction identifies +a sufficient precondition race rather than claiming direct observation of the +exact CI interleaving or production data loss. + +The test now requires exactly one saved comment from the resumed run and agent +before reload. It retains the original task, readiness, and reload limits and all +previous assertions. The same controlled delay then passes. Both the fixture +delay and temporary database-URI setup were removed/restored afterward. The +permanent fixture file is unchanged. All six journeys also pass without the +injected delay. No application polling or production timeout changed. + +Ponytail's current-head review found that reconstruction read the entire backlog +before any cancellation. Two real-PostgreSQL regressions failed on the old worker: +manual recovery read a second page before its first cancellation, and scoped +startup had already processed twenty items beyond the first malformed page. +The worker now loads at most one hundred raw candidates per scope per scheduler +pass. It retains a scope cursor through the existing helper's incomplete-load +contract. The shared scheduler continues only while that real reconstruction +obligation remains. Manual recovery streams pages and begins cancellation before +reading the full backlog, retaining its twenty-item execution limit and the +ability to pass malformed records. No independent timer or table was added. + +The regressions pass. The scoped case also commits a new valid intent behind the +active cursor, preserves fair twenty-item batches across two companies, settles +all twenty-nine valid obligations, and performs no SQL across the later former +poll periods. The worker keeps per-candidate admission reads: a settings change +or shutdown can occur during asynchronous batch work. The existing disabled/ +re-enabled and in-flight authority-read shutdown cases still pass. The optional +suggestion to cache that authority for a whole batch was not applied. The original +Migration paragraph now names the current 0406 artifacts, while retaining the +initial draft filename as history. + +Source `a35fa43b98e2bd378d4a920d2775ce243654838b` passes all **363 backend cases +in ten complete files**, with no skips, in 64.00 seconds. Seven separate fresh +PostgreSQL 17 databases migrated through 0406 isolate the fixtures. The temporary +external setup for the two embedded-only fixtures follows the qualification above; +both files were restored byte for byte. Actual database build, server typecheck/ +build, module boundaries, and exact-current migration order pass. The entire +database source tree is identical to the previously verified populated-upgrade +revision, so its 0405-to0406 upgrade and replay evidence remains applicable. + +The final six-browser-journey run on this source passes in 40.5 seconds, with +zero retries, the original 120-second case limit, and the original five-second +reload assertion. It uses the canonical CLI/server/static-UI path and a fresh +external PostgreSQL database. Only the fixture database URI was temporarily +adapted; this is not an unmodified embedded CI-environment reproduction. The +controlled red and green are one of those six journeys, not additional distinct +cases. The two focused paging cases are included in the 363, not added again. +Fresh review and CI on the new published head still must pass before merge. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 2b0f0663078..06becd2eba7 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -555,6 +555,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0401_yummy_ultimates.sql", "0402_ancient_malice.sql", "0403_natural_cerise.sql", + "0406_zippy_war_machine.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0406_zippy_war_machine.sql b/packages/db/src/migrations/0406_zippy_war_machine.sql new file mode 100644 index 00000000000..06ff3cf33bb --- /dev/null +++ b/packages/db/src/migrations/0406_zippy_war_machine.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_scheduled_retry_cancellation_idx" ON "heartbeat_runs" USING btree ("id") WHERE "heartbeat_runs"."status" = 'scheduled_retry' and jsonb_typeof("heartbeat_runs"."result_json"->'startupCancellation') = 'object'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0401_snapshot.json b/packages/db/src/migrations/meta/0406_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0401_snapshot.json rename to packages/db/src/migrations/meta/0406_snapshot.json index 95b8cabb6fc..866cacb35c5 100644 --- a/packages/db/src/migrations/meta/0401_snapshot.json +++ b/packages/db/src/migrations/meta/0406_snapshot.json @@ -1,6 +1,6 @@ { - "id": "c4c8a780-68b0-4b79-8c3b-5847f506ed7c", - "prevId": "843ab3c7-1378-42f3-b213-401a8d303e3f", + "id": "c34fcd91-0d57-4241-b8b4-399bc9f6dcb0", + "prevId": "c87ad360-f306-459b-b814-fde37327c6da", "version": "7", "dialect": "postgresql", "tables": { @@ -9689,6 +9689,50 @@ "method": "btree", "with": {} }, + "chat_actions_github_stale_head_idx": { + "name": "chat_actions_github_stale_head_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"payload\"->'event'->>'deliveryId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_review_check'\n and \"chat_actions\".\"status\" = 'cancelled' and \"chat_actions\".\"result\"->>'code' = 'stale_head'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "chat_actions_github_check_work_idx": { + "name": "chat_actions_github_check_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_review_check'\n and (\"chat_actions\".\"status\" = 'received' or (\"chat_actions\".\"status\" = 'failed' and \"chat_actions\".\"result\"->>'retryable' = 'true'))", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_actions_github_terminal_work_idx": { "name": "chat_actions_github_terminal_work_idx", "columns": [ @@ -13193,6 +13237,22 @@ } }, "indexes": { + "chat_github_reviews_projection_work_idx": { + "name": "chat_github_reviews_projection_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_github_reviews\".\"state\" in ('queued', 'running')\n and \"chat_github_reviews\".\"assessment\" is null\n and \"chat_github_reviews\".\"delivery_id\" ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_github_reviews_delivery_uq": { "name": "chat_github_reviews_delivery_uq", "columns": [ @@ -33683,6 +33743,22 @@ "method": "btree", "with": {} }, + "heartbeat_runs_scheduled_retry_cancellation_idx": { + "name": "heartbeat_runs_scheduled_retry_cancellation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"status\" = 'scheduled_retry' and jsonb_typeof(\"heartbeat_runs\".\"result_json\"->'startupCancellation') = 'object'", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -39051,6 +39127,140 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.issue_thread_interaction_idempotency_keys": { + "name": "issue_thread_interaction_idempotency_keys", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "issue_id": { + "name": "issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "interaction_id": { + "name": "interaction_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "idempotency_key": { + "name": "idempotency_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "issue_thread_interaction_idempotency_keys_scope_key_uq": { + "name": "issue_thread_interaction_idempotency_keys_scope_key_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "idempotency_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "issue_thread_interaction_idempotency_keys_interaction_idx": { + "name": "issue_thread_interaction_idempotency_keys_interaction_idx", + "columns": [ + { + "expression": "interaction_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "issue_thread_interaction_idempotency_keys_company_id_companies_id_fk": { + "name": "issue_thread_interaction_idempotency_keys_company_id_companies_id_fk", + "tableFrom": "issue_thread_interaction_idempotency_keys", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk": { + "name": "issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk", + "tableFrom": "issue_thread_interaction_idempotency_keys", + "tableTo": "issues", + "columnsFrom": [ + "issue_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk": { + "name": "issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk", + "tableFrom": "issue_thread_interaction_idempotency_keys", + "tableTo": "issue_thread_interactions", + "columnsFrom": [ + "interaction_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.issue_thread_interactions": { "name": "issue_thread_interactions", "schema": "", @@ -59903,6 +60113,69 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.user_workspace_preferences": { + "name": "user_workspace_preferences", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "coordinator_workspace_enabled": { + "name": "coordinator_workspace_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "user_workspace_preferences_user_uq": { + "name": "user_workspace_preferences_user_uq", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.work_assessments": { "name": "work_assessments", "schema": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index ac49afbe042..b032c1891e4 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2829,6 +2829,13 @@ "when": 1793231130478, "tag": "0405_harsh_avengers", "breakpoints": true + }, + { + "idx": 406, + "version": "7", + "when": 1793235984352, + "tag": "0406_zippy_war_machine", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 903b2a5249e..287ca78b7fb 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -120,6 +120,8 @@ export const heartbeatRuns = pgTable( sql`(${table.scopeKind} = 'company' AND ${table.issueId} IS NULL) OR ${table.scopeKind} = 'issue'`, ), + scheduledRetryCancellationIdx: index("heartbeat_runs_scheduled_retry_cancellation_idx").on(table.id) + .where(sql`${table.status} = 'scheduled_retry' and jsonb_typeof(${table.resultJson}->'startupCancellation') = 'object'`), costAccountingPendingIdx: index("heartbeat_runs_cost_accounting_pending_idx").on(table.updatedAt, table.id).where(sql`${table.costAccountingPending} = true`), costDataQualityIdx: index("heartbeat_runs_cost_data_quality_idx") .on(table.companyId, sql`coalesce(${table.finishedAt}, ${table.createdAt}) desc`, table.id) diff --git a/server/src/__tests__/issue-scheduled-retry-routes.test.ts b/server/src/__tests__/issue-scheduled-retry-routes.test.ts index 65adabfc09c..81168a07195 100644 --- a/server/src/__tests__/issue-scheduled-retry-routes.test.ts +++ b/server/src/__tests__/issue-scheduled-retry-routes.test.ts @@ -1,10 +1,18 @@ +import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createScheduledRetryCancellationRecovery } from "../services/heartbeat/scheduled-retry-cancellation.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyScheduledRetryCancellation } from "../services/scheduled-retry-cancellation-notifications.js"; +import { requestedRunCancellation } from "../services/run-cancellation.js"; +import { idleWorkSnapshot, startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; import { randomUUID } from "node:crypto"; import { spawn } from "node:child_process"; import { once } from "node:events"; import express from "express"; import request from "supertest"; import { and, eq, inArray, sql } from "drizzle-orm"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { activityLog, agents, @@ -12,6 +20,7 @@ import { companies, companyMemberships, createDb, + applyPendingMigrations, environmentLeases, heartbeatRunEvents, heartbeatRuns, @@ -29,7 +38,8 @@ import { issueRoutes } from "../routes/issues.js"; import { runningProcesses } from "../adapters/index.js"; import { heartbeatService } from "../services/heartbeat.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RETRY_CANCELLATION_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -45,8 +55,14 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-scheduled-retry-routes-"); - db = createDb(tempDb.connectionString); + if (externalTestDatabaseUrl) { + await applyPendingMigrations(externalTestDatabaseUrl); + db = createDb(externalTestDatabaseUrl); + } + else { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-scheduled-retry-routes-"); + db = createDb(tempDb.connectionString); + } }, 20_000); afterEach(async () => { @@ -65,6 +81,7 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -213,6 +230,380 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => { return { companyId, agentId, issueId, sourceRunId, retryRunId, scheduledRetryAt }; } + describe("scheduled retry cancellation worker", () => { + afterEach(() => { vi.useRealTimers(); vi.restoreAllMocks(); }); + + async function runRow(id: string) { + return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]); + } + function intent(reason = "Interrupted operator stop") { + return { cancellation: requestedRunCancellation({}, reason), startupCancellation: { requestedAt: new Date().toISOString() } }; + } + async function writeIntent(f: { companyId: string; retryRunId: string }, owner = db) { + await owner.transaction(async tx => { + await notifyScheduledRetryCancellation(tx, { companyId: f.companyId, runId: f.retryRunId }); + await tx.update(heartbeatRuns).set({ resultJson: intent() }).where(eq(heartbeatRuns.id, f.retryRunId)); + }); + } + async function waitCancelled(id: string) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(100); + expect((await runRow(id))?.status).toBe("cancelled"); + }, { timeout: 5000 }); + } + function clock() { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); } + + it("stays quiet with retries lacking cancellation and retained completed history", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + const historical = await seedIssueWithRetry({ agentStatus: "paused" }); + await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: intent() }).where(eq(heartbeatRuns.id, historical.retryRunId)); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(coordinator, () => true).ready; + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(180_000); + expect(reads).not.toHaveBeenCalled(); + reads.mockRestore(); + expect((await runRow(f.retryRunId)).status).toBe("scheduled_retry"); + } finally { await coordinator.stop(); } + }); + + it("recovers a real cancellation's failed final write from its committed signal and retry", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + const unrelated = await seedIssueWithRetry({ agentStatus: "paused" }); + const service = heartbeatService(db, { runtimeEnv: {} }); + const errors: unknown[] = []; + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await service.registerScheduledRetryCancellationWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + await db.execute(sql`ALTER TABLE heartbeat_runs ADD CONSTRAINT reject_worker_cancellation CHECK (status <> 'cancelled') NOT VALID`); + try { + await expect(service.cancelRun(f.retryRunId, "Human cancelled retry", { expectedStatus: "scheduled_retry" })).rejects.toThrow(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }); + expect((await runRow(f.retryRunId)).resultJson?.cancellation).toMatchObject({ reason: "Human cancelled retry" }); + } finally { await db.execute(sql`ALTER TABLE heartbeat_runs DROP CONSTRAINT reject_worker_cancellation`); } + await vi.advanceTimersByTimeAsync(31_000); + await waitCancelled(f.retryRunId); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect((await runRow(f.retryRunId)).error).toBe("Human cancelled retry"); + expect((await runRow(unrelated.retryRunId)).status).toBe("scheduled_retry"); + expect(await db.select().from(issues).where(eq(issues.id, f.issueId))) + .toEqual([expect.objectContaining({ executionRunId: null })]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.runId, f.retryRunId))) + .toEqual([expect.objectContaining({ status: "cancelled" })]); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId)); + await vi.advanceTimersByTimeAsync(120_000); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId))).toEqual(events); + expect(coordinator.nextWakeAt()).toBeNull(); + } finally { await coordinator.stop(); } + }); + + it("recovers when the caller loses the committed cancellation-fence response", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + const service = heartbeatService(db, { runtimeEnv: {} }); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + await service.registerScheduledRetryCancellationWork(coordinator, () => true).ready; + let injected = false; + const transaction = db.transaction.bind(db); + const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => { + const result = await transaction(callback, options); + if (!injected && result && typeof result === "object" && "id" in result && result.id === f.retryRunId) { + injected = true; + throw new Error("Lost cancellation-fence response"); + } + return result; + }) as typeof db.transaction); + try { + await expect(service.cancelRun(f.retryRunId, "Stop before settlement", { expectedStatus: "scheduled_retry" })) + .rejects.toThrow("Lost cancellation-fence response"); + expect(injected).toBe(true); + transactions.mockRestore(); + await waitCancelled(f.retryRunId); + expect((await runRow(f.retryRunId)).error).toBe("Stop before settlement"); + } finally { transactions.mockRestore(); await coordinator.stop(); } + }); + + it("reactivates when dispatch parks a queued retry with an existing cancellation", async () => { + const f = await seedIssueWithRetry({ retryStatus: "queued" }); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", resultJson: { + conversationContinuation: "continue_conversation_v1", workspaceRestoreFailure: "restore_failed", + } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", resultJson: { + ...intent(), conversationContinuation: "continue_conversation_v1", + } }).where(eq(heartbeatRuns.id, f.retryRunId)); + await db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.sourceRunId, status: "pending_cleanup", cleanupStatus: "failed", + }); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ companyId: f.companyId, + runId: f.retryRunId, expectedStatus: "queued", now: new Date(), + })).toMatchObject({ outcome: "deferred" }); + await waitCancelled(f.retryRunId); + expect((await runRow(f.retryRunId)).error).toBe("Interrupted operator stop"); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + } finally { await coordinator.stop(); } + }); + + it("preserves existing cancellation metadata through recovery", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + const budgetEnforcement = { companyId: f.companyId, scopeType: "agent", scopeId: f.agentId }; + const saved = intent(); + await db.update(heartbeatRuns).set({ resultJson: { ...saved, + startupCancellation: { ...saved.startupCancellation, budgetEnforcement }, + cancelledByUserId: "original-operator", recoveryEvidence: "retain-me", + } }).where(eq(heartbeatRuns.id, f.retryRunId)); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(coordinator, () => true).ready; + await waitCancelled(f.retryRunId); + expect((await runRow(f.retryRunId)).resultJson).toMatchObject({ + cancellation: saved.cancellation, startupCancellation: { budgetEnforcement }, + cancelledByUserId: "original-operator", recoveryEvidence: "retain-me", + }); + } finally { await coordinator.stop(); } + }); + + it("waits for the outer transaction and ignores a rolled-back intent", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(coordinator, () => true).ready; + await expect(db.transaction(async tx => { + await writeIntent(f, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(1000); + expect((await runRow(f.retryRunId)).resultJson).toBeNull(); + throw new Error("outer rollback"); + })).rejects.toThrow("outer rollback"); + await vi.advanceTimersByTimeAsync(1000); + expect((await runRow(f.retryRunId)).status).toBe("scheduled_retry"); + await db.transaction(async tx => { + await writeIntent(f, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(1000); + expect((await runRow(f.retryRunId)).resultJson).toBeNull(); + }); + await waitCancelled(f.retryRunId); + } finally { await coordinator.stop(); } + }); + + it.each(["restart", "drain", "standby"] as const)("retains a pending cancellation through %s", async mode => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + if (mode === "restart") await writeIntent(f); + let canRun = mode === "restart"; + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby", onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(coordinator, () => canRun).ready; + if (mode !== "restart") { + await writeIntent(f); + await vi.advanceTimersByTimeAsync(30_000); + if (mode !== "standby") await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect((await runRow(f.retryRunId)).status).toBe("scheduled_retry"); + canRun = true; + await vi.advanceTimersByTimeAsync(30_000); + } + await waitCancelled(f.retryRunId); + } finally { await coordinator.stop(); } + }); + + it.each(["manual", "expiry"] as const)("resumes a real task drain after %s release without passive SQL", async mode => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(coordinator, () => true).ready; + startTaskDrain(mode === "expiry" ? { ttlMs: 60_000 } : {}); + await writeIntent(f); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(30_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect((await runRow(f.retryRunId)).status).toBe("scheduled_retry"); + if (mode === "manual") stopTaskDrain(); + await vi.advanceTimersByTimeAsync(31_000); + await waitCancelled(f.retryRunId); + } finally { stopTaskDrain(); await coordinator.stop(); } + }); + + it("observes a real worktree activation without polling disabled settings", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + await writeIntent(f); + const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "sleep39-activation" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const service = heartbeatService(db, { runtimeEnv }); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await service.registerScheduledRetryCancellationWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await service.resolveSchedulingSuppression(); + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await waitCancelled(f.retryRunId); + } finally { await coordinator.stop(); await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + }); + + it("retains a second committed intent while the first cancellation is running", async () => { + const first = await seedIssueWithRetry({ agentStatus: "paused" }); + const second = await seedIssueWithRetry({ agentStatus: "paused" }); + await writeIntent(first); + const service = heartbeatService(db, { runtimeEnv: {} }); + let enter!: () => void, release!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const cancel = vi.fn(async (...args: Parameters) => { + if (args[0] === first.retryRunId) { enter(); await held; } + return service.cancelRun(...args); + }); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const worker = createScheduledRetryCancellationRecovery(db, cancel).register(coordinator, () => true, async () => true); + try { + await entered; + await writeIntent(second); + release(); await worker.ready; + await waitCancelled(first.retryRunId); await waitCancelled(second.retryRunId); + expect(cancel.mock.calls.filter(args => args[0] === first.retryRunId)).toHaveLength(1); + expect(cancel.mock.calls.filter(args => args[0] === second.retryRunId)).toHaveLength(1); + } finally { release(); await coordinator.stop(); } + }); + + it("does not repeat settlement when a foreground cancellation wins", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + await writeIntent(f); + const service = heartbeatService(db, { runtimeEnv: {} }); + let enter!: () => void, release!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const worker = createScheduledRetryCancellationRecovery(db, async (...args) => { + enter(); await held; return service.cancelRun(...args); + }).register(coordinator, () => true, async () => true); + try { + await entered; + await service.cancelRun(f.retryRunId, "Foreground stop", { expectedStatus: "scheduled_retry" }); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId)); + release(); await worker.ready; + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId))).toEqual(events); + expect((await runRow(f.retryRunId)).error).toBe("Foreground stop"); + } finally { release(); await coordinator.stop(); } + }); + + it("joins an in-flight authority read at shutdown without starting cancellation", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + await writeIntent(f); + const service = heartbeatService(db, { runtimeEnv: {} }); + let enter!: () => void, release!: () => void, checks = 0; + const entered = new Promise(resolve => { enter = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const cancel = vi.fn(service.cancelRun); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const worker = createScheduledRetryCancellationRecovery(db, cancel).register(coordinator, () => true, async () => { + if (++checks === 2) { enter(); await held; } + return true; + }); + let stopping: Promise | undefined; + try { + await entered; + let stopped = false; + stopping = coordinator.stop().then(() => { stopped = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(stopped).toBe(false); + release(); await worker.ready; await stopping; + expect(cancel).not.toHaveBeenCalled(); + expect((await runRow(f.retryRunId)).status).toBe("scheduled_retry"); + } finally { release(); await (stopping ?? coordinator.stop()); } + const restarted = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(restarted, () => true).ready; + expect((await runRow(f.retryRunId)).status).toBe("cancelled"); + } finally { await restarted.stop(); } + }); + + it("manual recovery starts cancellation without loading the full backlog", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + const work = await db.insert(heartbeatRuns).values(Array.from({ length: 120 }, () => ({ + companyId: f.companyId, agentId: f.agentId, status: "scheduled_retry", resultJson: intent(), + }))).returning({ id: heartbeatRuns.id }); + const service = heartbeatService(db, { runtimeEnv: {} }); + const reads = vi.spyOn(db, "select"); + let readsBeforeFirstCancellation: number | undefined; + try { + await createScheduledRetryCancellationRecovery(db, async (...args) => { + readsBeforeFirstCancellation ??= reads.mock.calls.filter(([fields]) => fields && + "result" in fields && "errorCode" in fields).length; + return service.cancelRun(...args); + }).sweep(); + expect(readsBeforeFirstCancellation).toBe(2); + const states = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, work.map(row => row.id))); + expect(states.filter(row => row.status === "cancelled")).toHaveLength(20); + expect(states.filter(row => row.status === "scheduled_retry")).toHaveLength(100); + } finally { reads.mockRestore(); } + }); + + it("pages past malformed intents and drains fair batches without revisiting completed history", async () => { + const a = await seedIssueWithRetry({ agentStatus: "paused" }); + const b = await seedIssueWithRetry({ agentStatus: "paused" }); + await db.insert(heartbeatRuns).values(Array.from({ length: 103 }, (_, i) => ({ + id: `00000000-0000-4000-8000-${String(i).padStart(12, "0")}`, companyId: a.companyId, agentId: a.agentId, + status: "scheduled_retry", resultJson: { startupCancellation: {}, cancellation: { expected: true } }, + }))); + const work = await db.insert(heartbeatRuns).values([...[a, b].flatMap((f, index) => + Array.from({ length: index === 0 ? 25 : 3 }, () => ({ companyId: f.companyId, agentId: f.agentId, + status: "scheduled_retry", resultJson: intent() }))), + ]).returning({ id: heartbeatRuns.id }); + const states = () => db.select({ companyId: heartbeatRuns.companyId, status: heartbeatRuns.status }).from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, work.map(row => row.id))); + clock(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await heartbeatService(db, { runtimeEnv: {} }).registerScheduledRetryCancellationWork(coordinator, () => true).ready; + expect((await states()).filter(row => row.status === "cancelled")).toHaveLength(0); + const earlierId = "00000000-0000-4000-8000-000000000000"; + await writeIntent({ companyId: a.companyId, retryRunId: earlierId }); + work.push({ id: earlierId }); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(async () => (await states()).filter(row => row.status === "cancelled").length).toBe(20); + const initial = (await states()).filter(row => row.status === "cancelled"); + expect(new Set(initial.map(row => row.companyId)).size).toBe(2); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await states()).filter(row => row.status === "cancelled")).toHaveLength(29); + }, { timeout: 10_000 }); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + await vi.advanceTimersByTimeAsync(1000); + await expect.poll(() => coordinator.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(180_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + } finally { await coordinator.stop(); } + }); + }); + it("finishes a parked retry cancellation after its owner loses the final database write", async () => { const f = await seedIssueWithRetry({ agentStatus: "paused" }); const unrelated = await seedIssueWithRetry({ agentStatus: "paused" }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 566a45df079..d4abf20fe82 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -114,6 +114,7 @@ const { })), reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })), reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), + registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })), resumeQueuedRuns: vi.fn(async () => undefined), recoverPendingSessionGoalActions: vi.fn(async () => ({ scanned: 0, enqueued: 0, alreadyQueued: 0, invalid: 0 })), @@ -549,6 +550,18 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers retry cancellation recovery only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerScheduledRetryCancellationWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) { + const [coordinator, admitted] = heartbeatServiceMock.registerScheduledRetryCancellationWork.mock.calls[0]; + const app = await createAppMock.mock.results.at(-1)!.value; + expect(coordinator).toBe(app.locals.deliveryWork); + expect(admitted()).toBe(true); + } + }); + it.each([false, true])("registers approval queue seeding only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -928,6 +941,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 }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index c7f7406b7a1..4b30bfdad2d 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1817,6 +1817,9 @@ async function startServerWithDatabaseTeardown( await startupHeartbeatRecovery; } + if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + const runRetentionSweep = async () => { const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active")); let archived = 0; @@ -1944,7 +1947,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 }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/modules/run-dispatch/adapters/postgres.test.ts b/server/src/modules/run-dispatch/adapters/postgres.test.ts index c9a8dc11a8f..397d7cc001c 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.test.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { createServer } from "node:http"; import { eq, sql } from "drizzle-orm"; @@ -12,6 +13,7 @@ import { companies, costEvents, createDb, + applyPendingMigrations, documentRevisions, documents, environmentLeases, @@ -44,7 +46,8 @@ import { costService } from "../../../services/costs.js"; // feeds the mapped facts back through the real gate to prove the wiring // produces the same suppression a caller would have seen before this module // existed. -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RUN_DISPATCH_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -58,8 +61,9 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-run-dispatch-postgres-adapter-"); - db = createDb(tempDb.connectionString); + if (external) await applyPendingMigrations(external); + else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-run-dispatch-postgres-adapter-"); + db = createDb(external ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -84,6 +88,7 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 558b353534e..547356e34fd 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js"; import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; import { getExecutionBlocker } from "../../../services/execution-blocker.js"; import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js"; @@ -666,6 +667,7 @@ export function createPostgresRunDispatchAdapter( | { applied: true; run: HeartbeatRun; postCommitEffects: PostCommitEffect[] } | { applied: false } > { + await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: issueId ?? undefined }); const [row] = await tx .update(heartbeatRuns) @@ -708,6 +710,7 @@ export function createPostgresRunDispatchAdapter( tx: Db, input: CancelSuppressedRetryInput, ): Promise { + await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); const [row] = await tx .update(heartbeatRuns) @@ -1085,6 +1088,7 @@ export function createPostgresRunDispatchAdapter( const previousWait = parseObject(run.resultJson?.executionWait); const scheduledRetryAt = new Date(now.getTime() + 30_000); const executionWait = { issueId, runId: blocker.runId, cause: blocker.cause }; + await notifyScheduledRetryCancellation(tx, { companyId: run.companyId, runId: run.id }); const [waiting] = await tx.update(heartbeatRuns) .set({ status: "scheduled_retry", diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index d8d02413fd8..54ff9a8f3bc 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js"; import { and, eq, or, sql } from "drizzle-orm"; import { agents, agentApiKeys, agentRuntimeState, agentTaskSessions, agentWakeupRequests, activityLog, budgetReservations, heartbeatRunEvents, heartbeatRuns, issueExecutionDecisions, issues, issueComments, @@ -42,6 +43,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str ); await tx.delete(issueExecutionDecisions).where(eq(issueExecutionDecisions.actorAgentId, id)); await tx.delete(issueComments).where(eq(issueComments.authorAgentId, id)); + await notifyScheduledRetryCancellation(tx, { companyId: companyId }); await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, id)); await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, id)); await tx.delete(agentApiKeys).where(eq(agentApiKeys.agentId, id)); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 819f2745abb..8685c7599fb 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js"; import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { chatCompanyDeletion } from "./chat-channels/company-deletion.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; @@ -96,6 +97,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(agentTaskSessions).where(eq(agentTaskSessions.companyId, id)); await tx.delete(activityLog).where(eq(activityLog.companyId, id)); await tx.delete(runIdentityContexts).where(eq(runIdentityContexts.companyId, id)); + await notifyScheduledRetryCancellation(tx, { companyId: id }); await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, id)); await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, id)); await tx.delete(agentApiKeys).where(eq(agentApiKeys.companyId, id)); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 2c2174eb20b..0a41dcf3690 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 = { + scheduledRetryCancellation: "scheduled-retry-cancellation", executionFinalization: "execution-finalization", workspaceRecovery: "workspace-recovery", executionReplacement: "execution-replacement", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 42a263a41cd..8535d7946dd 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -876,11 +876,11 @@ export function heartbeatService( at: 0, }; const WORKTREE_RUN_EXECUTION_OVERRIDE_TTL_MS = 3_000; - const resolveWorktreeRunExecutionOverride = async () => { + const resolveWorktreeRunExecutionOverride = async (fresh = false) => { if (!inWorktreeRuntime) return { allowed: false, cutoff: null }; const now = Date.now(); if ( - now - cachedWorktreeRunExecutionOverride.at < + !fresh && now - cachedWorktreeRunExecutionOverride.at < WORKTREE_RUN_EXECUTION_OVERRIDE_TTL_MS ) { return cachedWorktreeRunExecutionOverride; @@ -898,14 +898,15 @@ export function heartbeatService( cutoff: cutoff && !Number.isNaN(cutoff.getTime()) ? cutoff : null, at: now, }; - } catch { + } catch (error) { + if (fresh) throw error; // Keep the prior (default-false) value so a settings read failure fails // closed to the safe suppressed state. } return cachedWorktreeRunExecutionOverride; }; - const getSchedulingSuppression = async () => { - const override = await resolveWorktreeRunExecutionOverride(); + const getSchedulingSuppression = async (fresh = false) => { + const override = await resolveWorktreeRunExecutionOverride(fresh); return resolveHeartbeatSchedulingSuppression(runtimeEnv, { allowWorktreeRunExecution: override.allowed, }); @@ -1013,6 +1014,7 @@ export function heartbeatService( reapOrphanedRuns, sweepOrphanedActiveLeases, drainRunningRunsForShutdown, + registerScheduledRetryCancellationWork, } = createHeartbeatRecovery(db, { pluginWorkerManager: options.pluginWorkerManager, enterShutdown: () => { shutdownInProgress = true; }, @@ -4082,6 +4084,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerScheduledRetryCancellationWork: ( + coordinator: Parameters[0], canRun: () => boolean, + ) => registerScheduledRetryCancellationWork(coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), prepareHotRestartShutdown, reconcileHotRestartAdoption, diff --git a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts index 06ed71c35d1..561e21a6e5e 100644 --- a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts +++ b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts @@ -71,7 +71,7 @@ export const runControlPostgresSupport = externalTestDatabaseUrl ? { supported: if (!runControlPostgresSupport.supported) console.warn(`Skipping run-control database tests: ${runControlPostgresSupport.reason}`); export function setupRunControlDatabase(name: string, ready: (db: Db) => void) { - let database: Awaited>; + let database: Awaited> | undefined; let db: Db; beforeAll(async () => { if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index bd1694c3a41..aee0ca9e90e 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatResponseAvailability } from "../chat-response-availability.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; @@ -3488,6 +3489,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const reason = issueCancelled ? "Cancelled because the issue was cancelled before the scheduled retry became due" : "Cancelled because the issue was reassigned before the scheduled retry became due"; + await notifyScheduledRetryCancellation(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); await notifyChatRunMilestones(tx, { companyId: scheduledRun.companyId, issueId: issue.id }); const cancelled = await tx .update(heartbeatRuns) diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index d8dd720a109..3ab4dabb5fb 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -180,7 +180,7 @@ describe("heartbeat recovery module callbacks", () => { }); const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_TEST_DATABASE_URL"); -const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const support = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; if (!support.supported) console.warn(`Skipping recovery module database tests: ${support.reason}`); @@ -188,11 +188,9 @@ describePostgres("heartbeat recovery module ownership", () => { let db: Db; let database: Awaited>; beforeAll(async () => { - if (externalUrl) { - await applyPendingMigrations(externalUrl); - database = { connectionString: externalUrl, cleanup: async () => {} }; - } else database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-"); - db = createDb(database.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-"); + db = createDb(externalUrl ?? database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); @@ -473,7 +471,7 @@ describePostgres("heartbeat recovery module ownership", () => { const old = (await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]; await db.update(heartbeatRuns).set({ status: "interrupted", contextSnapshot: { issueId: f.run.issueId }, resultJson: { workspaceRestored: true } }).where(eq(heartbeatRuns.id, f.run.id)); await finishLegacyShutdownWorkspaceSettlement(db, f.run); - const single = createDb(database.connectionString, { maxConnections: 1 }); + const single = createDb(externalUrl ?? database.connectionString, { maxConnections: 1 }); try { expect(await allowLegacyShutdownWorkspaceCleanup(single, old, new Date(Date.now() + 60_000))).toBe(true); } finally { await single.$client.end({ timeout: 1 }); } expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]) diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 0454014d532..ce3f3de2172 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,3 +1,4 @@ +import { createScheduledRetryCancellationRecovery } from "./scheduled-retry-cancellation.js"; import { pendingCleanupDeadlineSql } from "../environment-cleanup-state.js"; import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js"; @@ -82,7 +83,6 @@ import { } from "../hot-restart.js"; import { serverVersion } from "../../version.js"; import { adapterExecutionControls, waitForAdapterStop } from "../adapter-execution-control.js"; -import { readRunCancellation } from "../run-cancellation.js"; import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; import type { environmentService } from "../environments.js"; @@ -390,6 +390,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD scheduleInteractionContinuationInfrastructureRetryIfEligible, startNextQueuedRunForAgent, } = dependencies; + const scheduledRetryCancellation = createScheduledRetryCancellationRecovery(db, cancelRunInternal); async function enqueueProcessLossRetry( run: typeof heartbeatRuns.$inferSelect, @@ -1946,35 +1947,11 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean }) { await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); - // A parked retry has no executor to finish a cancellation after a restart - // or a failed status write. Consume its durable intent through normal run - // control; the guarded CAS and terminal settlement make competing sweeps - // safe, and dispatch cannot promote a fenced retry in the meantime. - const interruptedRetryCancellations = await db.select().from(heartbeatRuns) - .where(and( - eq(heartbeatRuns.status, "scheduled_retry"), - sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') = 'object'`, - )) - .orderBy(asc(heartbeatRuns.updatedAt), asc(heartbeatRuns.id)) - .limit(20); - for (const run of interruptedRetryCancellations) { - const cancellation = readRunCancellation(run.resultJson); - if (!cancellation?.expected) continue; - try { - await cancelRunInternal(run.id, cancellation.reason, { - expectedStatus: "scheduled_retry", - errorCode: readNonEmptyString(parseObject(run.resultJson?.startupCancellation).errorCode) ?? "cancelled", - resultJson: parseObject(run.resultJson), - }); - } catch (error) { - // Leave the durable intent for the next bounded periodic sweep. - logger.warn({ err: error, runId: run.id }, "failed to finish interrupted retry cancellation"); - } - } + if (!opts?.skipScheduledRetryCancellation) await scheduledRetryCancellation.sweep(); // Recovery never launches a provider or infers stopped ownership from // terminal status. Uncaptured local copies require durable stop evidence. await instructionCopies.recoverStopped().catch(error => { @@ -2600,6 +2577,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } return { + registerScheduledRetryCancellationWork: scheduledRetryCancellation.register, sweepPendingCleanupLeases, markNativeOwnershipUnverified, prepareHotRestartShutdown, diff --git a/server/src/services/heartbeat/run-control.test.ts b/server/src/services/heartbeat/run-control.test.ts index c73dba4d120..bbcd348b1ca 100644 --- a/server/src/services/heartbeat/run-control.test.ts +++ b/server/src/services/heartbeat/run-control.test.ts @@ -17,6 +17,7 @@ import { terminateLocalService } from "../local-service-supervisor.js"; import { publishLiveEvent } from "../live-events.js"; import { createHeartbeatRunState } from "./run-state.js"; import { createHeartbeatRunControl } from "./run-control.js"; +import { createScheduledRetryCancellationRecovery } from "./scheduled-retry-cancellation.js"; import { preserveCancelledNativeWorkspaces } from "../native-runtime/native-failed-workspace-recovery.js"; import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js"; import { caseHandoffService } from "../case-handoffs.js"; @@ -273,6 +274,50 @@ describe.skipIf(!runControlPostgresSupport.supported)("heartbeat run-control dat expect(deps.appendRunEvent).not.toHaveBeenCalled(); }); + it.each([ + { boundary: "task", actorType: "user" }, + { boundary: "run", actorType: "user" }, + { boundary: "run", actorType: "board" }, + ] as const)("preserves $actorType Stop attribution when recovery wins after the $boundary fence", async ({ boundary, actorType }) => { + const { company, agent } = await fixture(); + const issue = boundary === "task" ? (await db.insert(issues).values({ + companyId: company.id, title: "Keep operator Stop", status: "in_progress", assigneeAgentId: agent.id, + }).returning())[0] : undefined; + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company.id, agentId: agent.id, issueId: issue?.id, status: "scheduled_retry", + runtimeMode: "legacy", runtimeModeResolvedAt: new Date(), + scheduledRetryAt: new Date(), scheduledRetryReason: "transient_failure", + }).returning(); + const deps = callbacks(db); + let enter!: () => void, release!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + + + if (boundary === "task") { + const getAgent = deps.getAgent.getMockImplementation()!; + deps.getAgent.mockImplementationOnce(async id => { enter(); await held; return getAgent(id); }); + } else { + const settle = deps.setRunStatusFromLive.getMockImplementation()!; + deps.setRunStatusFromLive.mockImplementationOnce(async (...args) => { enter(); await held; return settle(...args); }); + } + const stopping = createHeartbeatRunControl(db, deps).cancelRunInternal(run.id, "Operator Stop", { + expectedStatus: "scheduled_retry", resultJson: { cancelledByActorType: actorType, cancelledByUserId: "operator" }, + }); + try { + await Promise.race([entered, stopping.then(() => { throw new Error("Foreground Stop did not reach its fence"); })]); + const recovery = createHeartbeatRunControl(db, callbacks(db)); + await createScheduledRetryCancellationRecovery(db, recovery.cancelRunInternal).sweep(); + expect(await deps.getRun(run.id)).toMatchObject({ status: "cancelled", resultJson: { + cancelledByActorType: actorType, cancelledByUserId: "operator", + cancellation: { source: "operator", initiator: { type: "user", id: "operator" } }, + } }); + } finally { release(); await stopping; } + expect(await deps.getRun(run.id)).toMatchObject({ status: "cancelled", resultJson: { + cancelledByActorType: actorType, cancelledByUserId: "operator", + } }); + }); + it("blocks promotion after guarded cancellation wins but before it settles", async () => { const { company, agent } = await fixture(); await db.update(companies).set({ defaultResponsibleUserId: "operator" }).where(eq(companies.id, company.id)); diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index a6d57b7d46a..8159d2e04cf 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { resumeDeferredChatWake } from "../durable-chat-wakeup.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; @@ -292,6 +293,7 @@ export type CancelRunOptions = { function runCancellationFenceSql(input: { cancellation: ReturnType; errorCode: string; + attribution?: Pick, "cancelledByActorType" | "cancelledByUserId">; reassignmentStopRequested?: boolean; preserveCancellation?: boolean; budgetEnforcement?: BudgetEnforcementScope; @@ -317,6 +319,7 @@ function runCancellationFenceSql(input: { coalesce(nullif(${heartbeatRuns.resultJson}->'startupCancellation'->'budgetEnforcement', 'null'::jsonb), ${intent}::jsonb))`; } return sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || + ${JSON.stringify(input.attribution ?? {})}::jsonb || ${JSON.stringify(input.reassignmentStopRequested ? { reassignmentStopRequested: true } : {})}::jsonb || jsonb_build_object('cancellation', ${input.preserveCancellation ? sql`coalesce(${heartbeatRuns.resultJson}->'cancellation', ${cancellation}::jsonb)` @@ -1130,6 +1133,12 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont let run = await getRun(runId); if (!run) throw notFound("Heartbeat run not found"); if (options.expectedStatus && run.status !== options.expectedStatus) return null; + + + const attribution = { + cancelledByActorType: options.resultJson?.cancelledByActorType, + cancelledByUserId: options.resultJson?.cancelledByUserId, + }; const handoffIssueId = run.nativeIssueId ?? run.issueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null); const supersedesCaseHandoff = options.resultJson?.cancelledByActorType === "user" || options.errorCode === "issue_reassigned"; let cancellationClaimed = false; @@ -1175,9 +1184,10 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont startupCancellation: { requestedAt, beforeNativeSelection: currentRun.runtimeMode === "legacy" && !currentRun.runtimeModeResolvedAt && currentRun.executionStage === "preparing" && parseObject(currentRun.runnerProfileJson?.adapterDispatch).adapterType === "paperclip_runner" }, } : {}; + await notifyScheduledRetryCancellation(tx, { companyId: run!.companyId, runId }); const [fenced] = await tx.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ - executionCancellation: { state: "requested", requestedAt }, ...guardedCancellation, + ...attribution, executionCancellation: { state: "requested", requestedAt }, ...guardedCancellation, })}::jsonb`, }).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, run!.companyId), inArray(heartbeatRuns.status, cancellableStatuses))).returning(); @@ -1279,10 +1289,11 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont // additional durable fence before its existing cancellation path. if (!committedBudgetCancellation && (options.expectedStatus || options.budgetEnforcement || (!options.cancellationRequestId && (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control))))) { const fence = async (tx: Db) => { + await notifyScheduledRetryCancellation(tx, { companyId: run!.companyId, runId }); const [fenced] = await tx.update(heartbeatRuns).set({ // Record handoff intent before native cancellation can finalize and release // the run. Only its audited stop acknowledgement suppresses recovery. - resultJson: runCancellationFenceSql({ cancellation, errorCode, + resultJson: runCancellationFenceSql({ cancellation, errorCode, attribution, reassignmentStopRequested: options.errorCode === "issue_reassigned" && options.resultJson?.reassignmentStopConfirmed === true }), }).where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status, options.expectedStatus ? [options.expectedStatus] : pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES], @@ -1293,7 +1304,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont try { fenced = options.budgetEnforcement ? await withCurrentBudgetEnforcement(db, options.budgetEnforcement, fence) - : await fence(db); + : await db.transaction(tx => fence(tx as unknown as Db)); } catch (error) { stopOwnership?.release(); throw error; @@ -1668,6 +1679,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]), )).then((rows) => rows.map((row) => row.id)); if (!candidates.length) return []; + await notifyScheduledRetryCancellation(tx, { companyId: scope.companyId }); return tx.update(heartbeatRuns).set({ resultJson: runCancellationFenceSql({ cancellation: requestedRunCancellation({}, reason), errorCode: "cancelled", preserveCancellation: true, budgetEnforcement: scope, }) }).where(and(eq(heartbeatRuns.companyId, scope.companyId), inArray(heartbeatRuns.id, candidates), diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 06613c16844..d93a57f8fb3 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import os from "node:os"; @@ -1013,11 +1014,15 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl return recordHeartbeatRunRuntimeProgress(currentRun, update, issueId); } - async function writeRunStatus(status: string, companyId: string | undefined, write: (tx: Db) => Promise) { + async function writeRunStatus(status: string, previous: Pick | null, + write: (tx: Db) => Promise) { + const companyId = previous?.companyId; // Register on the outer transaction before the terminal write. An unknown // commit cannot strand newly-actionable accounting after the live event. return db.transaction(async transaction => { const tx = transaction as unknown as Db; + if (previous && (previous.status === "scheduled_retry" || status === "scheduled_retry")) + await notifyScheduledRetryCancellation(tx, { companyId: previous.companyId, runId: previous.id }); if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId); await notifyExecutionRunWork(tx); if (companyId) await notifyChatRunMilestones(tx, { companyId }); @@ -1065,7 +1070,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl }, }; } - const updated = await writeRunStatus(status, previousStatus?.companyId, writer => + const updated = await writeRunStatus(status, previousStatus, writer => previousStatus && previousStatus.runtimeMode === "legacy" && isHeartbeatRunTerminalStatus(status) ? terminalizeLegacyExecution({ db: writer, @@ -1161,7 +1166,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl }, }; } - const updated = await writeRunStatus(status, previousStatus?.companyId, writer => + const updated = await writeRunStatus(status, previousStatus, writer => previousStatus && previousStatus.runtimeMode === "legacy" && isHeartbeatRunTerminalStatus(status) ? terminalizeLegacyExecution({ db: writer, diff --git a/server/src/services/heartbeat/scheduled-retry-cancellation.ts b/server/src/services/heartbeat/scheduled-retry-cancellation.ts new file mode 100644 index 00000000000..85ebfce8f7c --- /dev/null +++ b/server/src/services/heartbeat/scheduled-retry-cancellation.ts @@ -0,0 +1,86 @@ +import { and, asc, eq, gt, sql } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { logger } from "../../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { readRunCancellation } from "../run-cancellation.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import type { ScheduledRetryCancellationScope } from "../scheduled-retry-cancellation-notifications.js"; +import type { CancelRunOptions } from "./run-control.js"; + +type Deadline = { id: string; companyId: string; dueAt: number }; + +export function createScheduledRetryCancellationRecovery(db: Db, cancelRun: ( + runId: string, reason: string, options: Pick, +) => Promise) { + function candidates(scope?: ScheduledRetryCancellationScope, cursor?: string, full = false) { + return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + result: full ? sql>`${heartbeatRuns.resultJson}` + : sql>`jsonb_build_object('cancellation', ${heartbeatRuns.resultJson}->'cancellation')`, + errorCode: sql`${heartbeatRuns.resultJson}->'startupCancellation'->'errorCode'`, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.status, "scheduled_retry"), + sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') = 'object'`, + scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, + cursor ? gt(heartbeatRuns.id, cursor) : undefined, + )).orderBy(asc(heartbeatRuns.id)).limit(100); + } + async function load(scope: ScheduledRetryCancellationScope | undefined, admitted: () => boolean, cursor?: string) { + if (!admitted()) throw new Error("Scheduled retry cancellation reconstruction stopped"); + const rows = await candidates(scope, cursor); + if (!admitted()) throw new Error("Scheduled retry cancellation reconstruction stopped"); + const entries: Deadline[] = rows.filter(row => readRunCancellation(row.result)?.expected) + .map(row => ({ id: row.id, companyId: row.companyId, dueAt: 0 })); + return { rows: entries, complete: rows.length < 100, cursor: rows.at(-1)?.id }; + } + async function check(entry: Deadline, admitted: () => boolean) { + const [run] = await candidates({ companyId: entry.companyId, runId: entry.id }, undefined, true); + const cancellation = readRunCancellation(run?.result); + if (!run || !cancellation?.expected) return null; + if (!admitted()) throw new Error("Scheduled retry cancellation stopped"); + await cancelRun(run.id, cancellation.reason, { expectedStatus: "scheduled_retry", + errorCode: typeof run.errorCode === "string" && run.errorCode.trim() ? run.errorCode.trim() : "cancelled", + resultJson: run.result }); + const [remaining] = await candidates({ companyId: entry.companyId, runId: entry.id }); + if (remaining && readRunCancellation(remaining.result)?.expected) + throw new Error("Scheduled retry cancellation remains pending"); + return null; + } + async function sweep() { + let cursor: string | undefined; + let checked = 0; + while (checked < 20) { + const page = await load(undefined, () => true, cursor); + for (const entry of page.rows) { + if (checked === 20) break; + checked++; + try { await check(entry, () => true); } + catch (error) { logger.warn({ err: error, runId: entry.id }, "failed to finish interrupted retry cancellation"); } + } + if (page.complete) break; + cursor = page.cursor; + } + } + function register(coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise) { + const cursors = new Map(); + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.scheduledRetryCancellation, + batchSize: 20, + decodeScope: (hint: string): ScheduledRetryCancellationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Deadline, scope?: ScheduledRetryCancellationScope) => !scope || + entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId), + load: async (scope, signal) => { + const key = scope ? `${scope.companyId}:${scope.runId ?? "*"}` : "*"; + if (!await canRecover()) { cursors.delete(key); return []; } + const page = await load(scope, () => canRun() && !signal?.aborted, cursors.get(key)); + if (page.complete) cursors.delete(key); else cursors.set(key, page.cursor!); + return { rows: page.rows, complete: page.complete }; + }, + check: async (entry, signal) => await canRecover() + ? check(entry, () => canRun() && !signal?.aborted) : null, + }, canRun); + } + return { sweep, register }; +} diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index c63ca707622..495d16cbe67 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js"; import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -615,6 +616,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); + if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx); await notifyToolConnectionHealth(tx, "*"); return tx diff --git a/server/src/services/native-runtime/native-cancellation-request.ts b/server/src/services/native-runtime/native-cancellation-request.ts index 4d588ad9b18..0ee79ed70b7 100644 --- a/server/src/services/native-runtime/native-cancellation-request.ts +++ b/server/src/services/native-runtime/native-cancellation-request.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { and, eq, sql } from "drizzle-orm"; import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { badRequest, conflict } from "../../errors.js"; @@ -111,6 +112,7 @@ export async function claimCancellationRequest(db: Db, runId: string, companyId: return run; } if (!pendingRetry && !["queued", "running", "scheduled_retry"].includes(run.status)) throw conflict("Correlated Stop run is already terminal"); + await notifyScheduledRetryCancellation(tx, { companyId, runId }); const [claimed] = await tx.update(heartbeatRuns).set({ resultJson: { ...result, startupCancellation: { requestedAt: new Date().toISOString(), beforeNativeSelection: false, cancellationRequestId: requestId, requestedBy: { type: "board", userId }, ...(run.status === "failed" ? { retryCancellation: true } : {}) }, } }).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).returning(); diff --git a/server/src/services/scheduled-retry-cancellation-notifications.ts b/server/src/services/scheduled-retry-cancellation-notifications.ts new file mode 100644 index 00000000000..8a8b4fe1a2f --- /dev/null +++ b/server/src/services/scheduled-retry-cancellation-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type ScheduledRetryCancellationScope = { companyId: string; runId?: string }; + +export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*"); +} diff --git a/tests/e2e/legacy-failure-continuation.spec.ts b/tests/e2e/legacy-failure-continuation.spec.ts index b04ca8e1e22..ed998017eae 100644 --- a/tests/e2e/legacy-failure-continuation.spec.ts +++ b/tests/e2e/legacy-failure-continuation.spec.ts @@ -138,6 +138,14 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag await expect(page.getByRole("button", { name: "Interrupt", exact: true })).toHaveCount(0); } if (action === "message") expect(prompts).toContain("Please continue the pending follow-up."); + const resumedRunId = runs.find(run => run.id !== sourceRunId)!.id; + await expect.poll(async () => { + const replies = await db.select({ body: issueComments.body }).from(issueComments).where(and( + eq(issueComments.companyId, company.id), eq(issueComments.issueId, issue.id), + eq(issueComments.authorAgentId, agent.id), eq(issueComments.createdByRunId, resumedRunId), + )); + return replies.filter(reply => reply.body.includes("Answered the pending follow-up once.")).length; + }, { timeout: 45_000 }).toBe(1); await page.reload(); await expect(page.getByText("Answered the pending follow-up once.", { exact: false })).toBeVisible(); } finally {