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 9a98efbf885..c9c2984936a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2051,3 +2051,26 @@ 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. 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..424d715b216 --- /dev/null +++ b/doc/plans/2026-10-25-scheduled-retry-cancellation-sleep-audit.md @@ -0,0 +1,133 @@ +# 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 + +`0394_breezy_mongu.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 0390 through 0394 are the retained newest five. +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. This slice and its parent stack are not authorized for merge. + +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. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 494492fdb31..54d6fce08b4 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -519,6 +519,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0391_jittery_norrin_radd.sql", "0392_charming_edwin_jarvis.sql", "0393_fearless_supernaut.sql", + "0394_breezy_mongu.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0394_breezy_mongu.sql b/packages/db/src/migrations/0394_breezy_mongu.sql new file mode 100644 index 00000000000..06ff3cf33bb --- /dev/null +++ b/packages/db/src/migrations/0394_breezy_mongu.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/0389_snapshot.json b/packages/db/src/migrations/meta/0394_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0389_snapshot.json rename to packages/db/src/migrations/meta/0394_snapshot.json index dadecf22e7f..a1c0ba77f16 100644 --- a/packages/db/src/migrations/meta/0389_snapshot.json +++ b/packages/db/src/migrations/meta/0394_snapshot.json @@ -1,6 +1,6 @@ { - "id": "8bd7f279-1063-4ba6-83bb-b85cba334eb3", - "prevId": "f83c3343-8665-48bc-919b-bc9fe7513b87", + "id": "90ac76c4-b6d8-4c38-b4b8-d51699ca2ec1", + "prevId": "f6db457e-383c-4518-a0de-7dde2fa6816d", "version": "7", "dialect": "postgresql", "tables": { @@ -9616,6 +9616,82 @@ "concurrently": false, "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": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"status\" = 'processed' and \"chat_actions\".\"delivery_id\" is not null\n and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "chat_actions_github_ack_work_idx": { + "name": "chat_actions_github_ack_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"delivery_id\" is not null and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'\n and (\"chat_actions\".\"status\" = 'received' or (\"chat_actions\".\"status\" = 'failed' and \"chat_actions\".\"result\"->>'retryable' = 'true'))", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -12043,6 +12119,40 @@ "method": "btree", "with": {} }, + "chat_publications_comment_destination_idx": { + "name": "chat_publications_comment_destination_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "comment_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "conversation_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_publications\".\"comment_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_publications_idempotency_uq": { "name": "chat_publications_idempotency_uq", "columns": [ @@ -13055,6 +13165,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": [ @@ -30460,6 +30586,34 @@ "method": "btree", "with": {} }, + "fast_response_requests_delivery_idx": { + "name": "fast_response_requests_delivery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "delivery_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"fast_response_requests\".\"delivery_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "fast_response_requests_history_idx": { "name": "fast_response_requests_history_idx", "columns": [ @@ -33489,6 +33643,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": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 0bed8cc9dd3..6f295be6990 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2745,6 +2745,13 @@ "when": 1792968423108, "tag": "0393_fearless_supernaut", "breakpoints": true + }, + { + "idx": 394, + "version": "7", + "when": 1792972400940, + "tag": "0394_breezy_mongu", + "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..1129988c6c8 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, @@ -29,7 +37,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 +54,11 @@ 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) db = createDb(externalTestDatabaseUrl); + else { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-scheduled-retry-routes-"); + db = createDb(tempDb.connectionString); + } }, 20_000); afterEach(async () => { @@ -65,6 +77,7 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -213,6 +226,353 @@ 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("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; + const initial = (await states()).filter(row => row.status === "cancelled"); + expect(initial).toHaveLength(20); + 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(28); + }, { 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/index.ts b/server/src/index.ts index 13c6362003b..db158e77e90 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1770,6 +1770,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; @@ -1897,7 +1900,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. trackHeartbeatSchedulerWork(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..0ac888de772 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"; @@ -44,7 +45,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 +60,8 @@ 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) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-run-dispatch-postgres-adapter-"); + db = createDb(external ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -84,6 +86,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 6d2453bf888..600f9598c8d 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"; @@ -94,6 +95,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 867369aeba8..fc04f316134 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", executionReplacement: "execution-replacement", executionContinuation: "execution-continuation", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e6e41be53f3..c4ee17f0fa0 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -869,11 +869,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; @@ -891,14 +891,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, }); @@ -1006,6 +1007,7 @@ export function heartbeatService( reapOrphanedRuns, sweepOrphanedActiveLeases, drainRunningRunsForShutdown, + registerScheduledRetryCancellationWork, } = createHeartbeatRecovery(db, { pluginWorkerManager: options.pluginWorkerManager, enterShutdown: () => { shutdownInProgress = true; }, @@ -4060,6 +4062,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 42f6bea5150..3f825cb67e2 100644 --- a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts +++ b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; import type { Db } from "@paperclipai/db"; import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns } from "@paperclipai/db"; import { sql } from "drizzle-orm"; @@ -63,15 +64,16 @@ export function callbacks(db: Db): TestCallbacks { failedProcessRunCancellations: new Map; failed: boolean; error?: unknown }>(), } satisfies HeartbeatRunControlDependencies; } -export const runControlPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RUN_CONTROL_TEST_DATABASE_URL"); +export const runControlPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); 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 () => { - database = await startEmbeddedPostgresTestDatabase(name); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase(name); + db = createDb(external ?? database!.connectionString); ready(db); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index c8611cb6765..21264b632b7 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"; @@ -3440,6 +3441,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 8faa8c7e865..f773350b09b 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environments, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; @@ -166,7 +167,8 @@ describe("heartbeat recovery module callbacks", () => { }); }); -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_TEST_DATABASE_URL"); +const support = external ? { 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}`); @@ -174,10 +176,10 @@ describePostgres("heartbeat recovery module ownership", () => { let db: Db; let database: Awaited>; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-"); + db = createDb(external ?? database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); - afterAll(async () => { await database?.cleanup(); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); async function seedRun() { const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), issueId = randomUUID(); @@ -394,7 +396,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(external ?? 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 73c586bdff3..6b70216769e 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 type { WorkspaceWorkers } from "../plugin-workspaces.js"; @@ -80,7 +81,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"; @@ -386,6 +386,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD scheduleInteractionContinuationInfrastructureRetryIfEligible, startNextQueuedRunForAgent, } = dependencies; + const scheduledRetryCancellation = createScheduledRetryCancellationRecovery(db, cancelRunInternal); async function enqueueProcessLossRetry( run: typeof heartbeatRuns.$inferSelect, @@ -1914,35 +1915,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 => { @@ -2551,6 +2528,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } return { + registerScheduledRetryCancellationWork: scheduledRetryCancellation.register, sweepPendingCleanupLeases, markNativeOwnershipUnverified, prepareHotRestartShutdown, diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 76e41c1e422..bd0acdf323a 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 { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { readTaskRetryIntent, reconcileTaskChatRetryReceipts } from "../task-retry.js"; @@ -1135,6 +1136,7 @@ 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, @@ -1219,6 +1221,7 @@ 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. @@ -1233,7 +1236,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; @@ -1608,6 +1611,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 14a11b03f6a..3eb7c28a2f5 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"; @@ -1012,11 +1013,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 }); @@ -1064,7 +1069,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, @@ -1160,7 +1165,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..b6d2440d680 --- /dev/null +++ b/server/src/services/heartbeat/scheduled-retry-cancellation.ts @@ -0,0 +1,76 @@ +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) { + const entries: Deadline[] = []; + let cursor: string | undefined; + while (admitted()) { + const rows = await candidates(scope, cursor); + for (const row of rows) if (readRunCancellation(row.result)?.expected) + entries.push({ id: row.id, companyId: row.companyId, dueAt: 0 }); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + throw new Error("Scheduled retry cancellation reconstruction stopped"); + } + 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() { + const entries = await load(undefined, () => true); + for (const entry of entries.slice(0, 20)) { + try { await check(entry, () => true); } + catch (error) { logger.warn({ err: error, runId: entry.id }, "failed to finish interrupted retry cancellation"); } + } + } + function register(coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise) { + 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) => await canRecover() + ? load(scope, () => canRun() && !signal?.aborted) : [], + 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/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index a076d761644..711b5d2aaf5 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -32,6 +32,7 @@ export function registerScopedDeadlineWork( return [...dirty].map(hint => { try { return options.decodeScope(hint); } catch { return undefined; } }); } function next() { + if (!canRun()) return dirty.size > 0 || entries.size > 0 ? Date.now() + 30_000 : null; let nextAt: number | null = null; const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; for (const hint of dirty.has("*") ? ["*"] : dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0));