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 {