fix(server): wake interrupted retry cancellation from committed intent

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<void>(resolve => { enter = resolve; });
+      const held = new Promise<void>(resolve => { release = resolve; });
+      const cancel = vi.fn(async (...args: Parameters<typeof service.cancelRun>) => {
+        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<void>(resolve => { enter = resolve; });
+      const held = new Promise<void>(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<void>(resolve => { enter = resolve; });
+      const held = new Promise<void>(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<void> | undefined;
+      try {
+        await entered;
+        let stopped = false;
+        stopping = coordinator.stop().then(() => { stopped = true; });
+        await new Promise<void>(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" });

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();

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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();
   });
 

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<CancelSuppressedRetryResult> {
+    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",

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));

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));

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",

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<typeof registerScheduledRetryCancellationWork>[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,

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<string, { settled: Promise<void>; 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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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 () => {

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)

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
   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])

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,

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),

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<T>(status: string, companyId: string | undefined, write: (tx: Db) => Promise<T>) {
+  async function writeRunStatus<T>(status: string, previous: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "status"> | null,
+    write: (tx: Db) => Promise<T>) {
+    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,

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<CancelRunOptions, "expectedStatus" | "errorCode" | "resultJson">,
+) => Promise<unknown>) {
+  function candidates(scope?: ScheduledRetryCancellationScope, cursor?: string, full = false) {
+    return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
+      result: full ? sql<Record<string, unknown>>`${heartbeatRuns.resultJson}`
+        : sql<Record<string, unknown>>`jsonb_build_object('cancellation', ${heartbeatRuns.resultJson}->'cancellation')`,
+      errorCode: sql<unknown>`${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<typeof createDeliveryWorkCoordinator>, canRun: () => boolean,
+    canRecover: () => Promise<boolean>) {
+    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 };
+}

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

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();

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) : "*");
+}

server/src/services/scoped-deadline-work.ts

@@ -32,6 +32,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     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));