perf(chat): stop polling failed run retries (#15871)

Stop one-second scans of failed chat-run retries. Wake the existing retry queue after transaction commits and at saved retry or stale-claim deadlines. Preserve claim fences, authorization, recovery, and idempotent agent admission.

Resolve conflicts with the current chat-service extraction by applying the retry signals and shared deadline predicate in the failed-run retry module.

Verified with pnpm typecheck, pnpm build, all 1,328 chat and coordinator regression tests, and current-head CI. Greptile reviewed the final head at 5/5 with no unresolved threads.

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

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -110,6 +110,7 @@ describe("createChatReconciliationCoordinator", () => {
     const service = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8");
     const actionMaintenance = service.slice(service.indexOf("async function processPendingChatMaintenance("),
       service.indexOf("async function processPendingDeliveries("));
+    expect(actionMaintenance).not.toContain("processFailedChatRunRetries(");
     expect(actionMaintenance).not.toContain("processPendingProviderEffects(");
     expect(actionMaintenance).not.toContain("processPendingGitHubWebhookIngress(");
   });

server/src/__tests__/chat-action-work.test.ts

@@ -19,6 +19,7 @@ function setup() {
   const coordinator = createDeliveryWorkCoordinator({ owner, canRun: enabled, canReconcile: enabled, onError });
   coordinators.push(coordinator);
   const service = {
+    processFailedChatRunRetries: vi.fn(async () => {}),
     processPendingProviderEffects: vi.fn(async () => {}),
     processPendingGitHubWebhookIngress: vi.fn(async () => {}),
     processPendingSlackSessionSyncs: vi.fn(async () => {}),
@@ -30,7 +31,7 @@ function setup() {
     nextChatActionAt: vi.fn(async (): Promise<number | null> => null),
     nextVerificationMessageAt: vi.fn(async (): Promise<number | null> => null),
   };
-  const runs = [service.processPendingProviderEffects, service.processPendingGitHubWebhookIngress, service.processPendingSlackSessionSyncs, service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts,
+  const runs = [service.processFailedChatRunRetries, service.processPendingProviderEffects, service.processPendingGitHubWebhookIngress, service.processPendingSlackSessionSyncs, service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts,
     service.processPendingReceiptReactions, service.processPendingSlackSessionStops, service.processPendingVerificationMessages];
   return { owner, service, runs, coordinator, enabled, onError, start: () => registerChatActionWork(coordinator, service, enabled) };
 }
@@ -40,7 +41,7 @@ describe("chat action scheduling", () => {
     await s.start().ready;
     await vi.advanceTimersByTimeAsync(3_600_000);
     for (const run of s.runs) expect(run).toHaveBeenCalledTimes(1);
-    expect(s.service.nextChatActionAt).toHaveBeenCalledTimes(7);
+    expect(s.service.nextChatActionAt).toHaveBeenCalledTimes(8);
     expect(s.service.nextVerificationMessageAt).toHaveBeenCalledTimes(1);
     expect(s.coordinator.nextWakeAt()).toBeNull();
     expect(vi.getTimerCount()).toBe(0);

server/src/__tests__/chat-channels.integration.test.ts

@@ -1,3 +1,4 @@
+import { idleWorkSnapshot } from "../services/task-admission.js";
 import * as activityLogModule from "../services/activity-log.js";
 import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates } from "@paperclipai/db";
 import type { PaperclipCloudConnector, SealedConnectorCredentials } from "../services/paperclip-cloud-connector.js";
@@ -33026,6 +33027,26 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
           await set("delivery_unknown");
           expect(await f.service.nextChatActionAt(kind)).toBeNull();
           expect(await due()).toBe(false);
+        } else if (kind === "failed_run_retry") {
+          expect(await due()).toBe(false);
+          await set("issued");
+          expect(await due()).toBe(true);
+          await set("issued", { retryAt: later.toISOString() });
+          expect(await f.service.nextChatActionAt(kind)).toBe(later.getTime());
+          expect(await due()).toBe(false);
+          await set("processing");
+          expect(await f.service.nextChatActionAt(kind)).toBe(at.getTime() + 60_000);
+          expect(await due()).toBe(false);
+          // Both the claim lease and retry time must expire before replay.
+          await set("processing", { retryAt: later.toISOString() });
+          expect(await f.service.nextChatActionAt(kind)).toBe(later.getTime());
+          await set("processing", { retryAt: at.toISOString() });
+          expect(await f.service.nextChatActionAt(kind)).toBe(at.getTime() + 60_000);
+          for (const status of ["failed", "cancelled", "processed"]) {
+            await set(status, { retryable: true, retryAt: at.toISOString() });
+            expect(await f.service.nextChatActionAt(kind)).toBeNull();
+            expect(await due()).toBe(false);
+          }
         } else if (kind === "slack_session_sync") {
           expect(await due()).toBe(true);
           await set("received", { retryAt: later.toISOString() });
@@ -59779,6 +59800,81 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     },
   );
 
+  it.each(["commit", "retry", "startup", "stale_claim", "rollback"] as const)(
+    "drains failed chat retries from durable work and becomes idle: %s",
+    async (mode) => {
+      const context = await failedChatRetryFixture("telegram");
+      const onError = vi.fn();
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+      const input = { companyId: context.fixture.companyId, issueId: context.issue.id,
+        agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user" };
+      const stage = () => db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, input));
+      context.wakeup.mockClear();
+      try {
+        let actionId: string;
+        if (mode === "startup" || mode === "stale_claim") {
+          ({ actionId } = await stage());
+          if (mode === "stale_claim") await db.update(chatActions).set({ status: "processing",
+            updatedAt: new Date(Date.now() - 120_000) }).where(eq(chatActions.id, actionId));
+          await registerChatActionWork(coordinator, context.service, () => true).ready;
+        } else {
+          await registerChatActionWork(coordinator, context.service, () => true).ready;
+          if (mode === "rollback") {
+            await expect(db.transaction(async tx => {
+              await context.service.prepareFailedChatRunRetry(tx, input);
+              throw new Error("rollback retry intent");
+            })).rejects.toThrow("rollback retry intent");
+            await vi.waitFor(() => {
+              expect(coordinator.nextWakeAt()).toBeNull();
+              expect(idleWorkSnapshot().active).toBe(0);
+            }, { timeout: 10_000 });
+            expect(context.wakeup).not.toHaveBeenCalled();
+          }
+          if (mode === "retry") {
+            let failedAt = 0;
+            context.wakeup.mockImplementationOnce(async () => {
+              failedAt = Date.now();
+              throw new Error("temporary admission outage");
+            }).mockImplementationOnce(async () => {
+              expect(Date.now() - failedAt).toBeGreaterThanOrEqual(250);
+              return { accepted: true };
+            });
+          }
+          ({ actionId } = await db.transaction(async tx => {
+            const staged = await tx.transaction(nested => context.service.prepareFailedChatRunRetry(nested, input));
+            // A nested savepoint must not wake the worker before the outer commit.
+            await new Promise(resolve => setTimeout(resolve, 25));
+            expect(context.wakeup).not.toHaveBeenCalled();
+            return staged;
+          }));
+        }
+        await vi.waitFor(async () => {
+          const [action] = await db.select().from(chatActions).where(eq(chatActions.id, actionId));
+          expect(action?.status).toBe("processed");
+          expect(action?.result.attemptCount).toBe(mode === "retry" ? 2 : 1);
+          expect(coordinator.nextWakeAt()).toBeNull();
+        }, { timeout: 10_000 });
+        expect(context.wakeup).toHaveBeenCalledTimes(mode === "retry" ? 2 : 1);
+        const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, actionId));
+        expect(receipts).toHaveLength(1);
+        expect(receipts[0]?.payload).toMatchObject({ retryOfRunId: context.runId,
+          wakeCommentIds: [context.action.payload.commentId] });
+        // A running worker has no scheduled deadline. Wait for its final scan
+        // and any commit-driven follow-up before measuring the quiet period.
+        await vi.waitFor(() => {
+          expect(coordinator.nextWakeAt()).toBeNull();
+          expect(idleWorkSnapshot().active).toBe(0);
+        }, { timeout: 10_000 });
+        const scans = vi.spyOn(db, "select");
+        try {
+          await new Promise(resolve => setTimeout(resolve, 1200));
+          expect(scans).not.toHaveBeenCalled();
+        } finally { scans.mockRestore(); }
+        expect(onError).not.toHaveBeenCalled();
+      } finally { await coordinator.stop(); await context.service.shutdown(); }
+    },
+  );
+
   it.each(["slack", "telegram"] as const)(
     "retries the exact failed %s chat input with a new idempotent admission",
     async (provider) => {

server/src/services/chat-action-work.ts

@@ -8,6 +8,7 @@ import { registerChatQueueWork } from "./chat-delivery-work.js";
 export const SLACK_COMMAND_POST_STALE_MS = 60_000;
 export const SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS = 5 * 60_000;
 export const SLACK_COMMAND_ADMISSION_STALE_MS = 60_000;
+export const FAILED_RUN_RETRY_STALE_MS = 60_000;
 export const PROVIDER_EFFECT_STALE_MS = 60_000;
 
 /** The dispatch predicate and scheduler use the same eligibility and deadlines.
@@ -15,6 +16,9 @@ export const PROVIDER_EFFECT_STALE_MS = 60_000;
 export function chatActionDeadline(kind: ScheduledChatAction) {
   const updated = sql`extract(epoch from ${chatActions.updatedAt}) * 1000`;
   const retry = sql`coalesce(extract(epoch from (${chatActions.result}->>'retryAt')::timestamptz) * 1000, 0)`;
+  if (kind === "failed_run_retry") return sql<number>`case
+    when ${chatActions.status} = 'issued' then ${retry}
+    when ${chatActions.status} = 'processing' then greatest(${updated} + ${FAILED_RUN_RETRY_STALE_MS}, ${retry}) end`;
   if (kind === "slack_board_message") return sql<number>`case
     when ${chatActions.status} = 'received' and ${chatEndpoints.status} not in ('paused', 'attention') then 0::numeric end`;
   if (kind === "slash_task_start") return sql<number>`case
@@ -36,14 +40,16 @@ export function chatActionDeadline(kind: ScheduledChatAction) {
 function pendingChatAction(kind: ScheduledChatAction) {
   // Filter history before the endpoint join and aggregate. Failed actions
   // only remain live when the saved result permits another attempt.
-  const status = kind === "slack_board_message"
-    ? eq(chatActions.status, "received")
-    : kind === "slash_task_start"
-      ? inArray(chatActions.status, ["queued", "provider_confirmed", "admitting", "resolving", "validating"])
-      : kind === "slack_session_sync"
-        ? inArray(chatActions.status, ["received", "processing"])
-        : or(inArray(chatActions.status, ["received", "processing"]),
-        and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`));
+  const status = kind === "failed_run_retry"
+    ? inArray(chatActions.status, ["issued", "processing"])
+    : kind === "slack_board_message"
+      ? eq(chatActions.status, "received")
+      : kind === "slash_task_start"
+        ? inArray(chatActions.status, ["queued", "provider_confirmed", "admitting", "resolving", "validating"])
+        : kind === "slack_session_sync"
+          ? inArray(chatActions.status, ["received", "processing"])
+          : or(inArray(chatActions.status, ["received", "processing"]),
+          and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`));
   return and(eq(chatActions.kind, kind), status);
 }
 
@@ -60,6 +66,7 @@ export async function nextChatActionAt(db: Db, kind: ScheduledChatAction): Promi
 }
 
 type ChatActionService = {
+  processFailedChatRunRetries(): Promise<unknown>;
   processPendingProviderEffects(): Promise<unknown>;
   processPendingGitHubWebhookIngress(): Promise<unknown>;
   processPendingSlackSessionSyncs(): Promise<unknown>;
@@ -78,6 +85,7 @@ export function registerChatActionWork(
   canRun: () => boolean,
 ) {
   const actions = [
+    { kind: "failed_run_retry", run: service.processFailedChatRunRetries },
     { kind: "provider_effect", run: service.processPendingProviderEffects },
     { kind: "github_webhook_ingress", run: service.processPendingGitHubWebhookIngress },
     { kind: "slack_session_sync", run: service.processPendingSlackSessionSyncs },

server/src/services/chat-channels.ts

@@ -5729,7 +5729,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     lifecycleRuntimeFence,
     isShuttingDown: () => shuttingDown,
     wakeup: (...args) => options.heartbeat.wakeup(...args),
-    staleProcessingMs: DELIVERY_PROCESSING_STALE_MS,
   });
   const { stageTeamsBoardFileIntent, deriveTeamsFileTransferAuthority } = createTeamsFileAuthority({ runtimeGeneration, credentialFingerprint, runtime, runtimeContexts, localRuntimeEpoch, runtimeVersions, lockCurrentPrincipalAuthorization, lifecycleRuntimeFence, assertFailedRetryReceipt, inboundWakeScope, failedChatRetrySource, isExplicitOperatorPublication, retryCommentIds });
   async function authorizeRetainedChatSourcePublication(
@@ -15429,7 +15428,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
   async function processPendingChatMaintenance(limit = 25) {
     return Promise.allSettled([
       processPendingTelegramMaintenance(limit),
-      processFailedChatRunRetries(limit),
     ]);
   }
 
@@ -15440,7 +15438,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     // so a crashed processing claim is quarantined before the delivery worker
     // could otherwise replay it.
     const actionRecovery = onlyDeliveryId || !maintenance ? null : Promise.allSettled([
-      processPendingChatMaintenance(limit), processPendingGitHubWebhookIngress(limit),
+      processPendingChatMaintenance(limit), processFailedChatRunRetries(limit),
+      processPendingGitHubWebhookIngress(limit),
       processPendingProviderEffects(limit), processPendingSlackBoardMessages(limit),
       processPendingSlackTaskStarts(limit), processPendingReceiptReactions(limit),
       processPendingSlackSessionStops(limit), slackRegistration.processPendingVerificationMessages(limit),

server/src/services/chat-channels/failed-run-retry.ts

@@ -1,5 +1,6 @@
 import { continuingTeamsFilePublication, type ContinuingTeamsFileTransfer } from "./microsoft-teams/file-transfer-source.js";
-import { notifyChatPublicationWork } from "../chat-work-notifications.js";
+import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-notifications.js";
+import { FAILED_RUN_RETRY_STALE_MS, dueChatAction } from "../chat-action-work.js";
 
 import {
   createHash,
@@ -13,7 +14,6 @@ import {
   isNotNull,
   isNull,
   like,
-  lte,
   ne,
   notInArray,
   or,
@@ -104,7 +104,6 @@ type ChatFailedRunRetryOptions = {
   lifecycleRuntimeFence: (delivery: DeliveryRow) => RetryRuntimeFence | null;
   isShuttingDown: () => boolean;
   wakeup: IssueAssignmentWakeupDeps["wakeup"];
-  staleProcessingMs: number;
 };
 
 export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOptions) {
@@ -115,7 +114,6 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
     lifecycleRuntimeFence,
     isShuttingDown,
     wakeup,
-    staleProcessingMs: DELIVERY_PROCESSING_STALE_MS,
   } = options;
 
 
@@ -1265,6 +1263,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
       .limit(1);
     if (busy || !input.initiatedByUserId) throw failedChatRetryDenied();
     const now = new Date();
+    await notifyChatActionWork(tx, "failed_run_retry");
     const [intent] = await tx
       .insert(chatActions)
       .values({
@@ -1508,36 +1507,42 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
       !["issued", "processing"].includes(candidate.status) ||
       (candidate.status === "processing" &&
         candidate.updatedAt.getTime() >
-          now.getTime() - DELIVERY_PROCESSING_STALE_MS) ||
+          now.getTime() - FAILED_RUN_RETRY_STALE_MS) ||
       (typeof candidate.result?.retryAt === "string" &&
         Date.parse(candidate.result.retryAt) > now.getTime())
     )
       return failedChatRetryReceipt(candidate);
-    const [claimed] = await db
-      .update(chatActions)
-      .set({ status: "processing", updatedAt: now })
-      .where(
-        and(
-          eq(chatActions.id, candidate.id),
-          eq(chatActions.status, candidate.status),
-          eq(chatActions.updatedAt, candidate.updatedAt),
-        ),
-      )
-      .returning();
-    if (!claimed) return failedChatRetryReceipt(candidate);
-    const attemptCount = Number(candidate.result?.attemptCount ?? 0) + 1;
-    const settle = async (status: string, result: Record<string, unknown>) => {
-      const [current] = await db
+    const [claimed] = await db.transaction(async (tx) => {
+      await notifyChatActionWork(tx, "failed_run_retry");
+      return tx
         .update(chatActions)
-        .set({ status, result, updatedAt: new Date() })
+        .set({ status: "processing", updatedAt: now })
         .where(
           and(
-            eq(chatActions.id, claimed.id),
-            eq(chatActions.status, "processing"),
-            eq(chatActions.updatedAt, now),
+            eq(chatActions.id, candidate.id),
+            eq(chatActions.status, candidate.status),
+            eq(chatActions.updatedAt, candidate.updatedAt),
           ),
         )
         .returning();
+    });
+    if (!claimed) return failedChatRetryReceipt(candidate);
+    const attemptCount = Number(candidate.result?.attemptCount ?? 0) + 1;
+    const settle = async (status: string, result: Record<string, unknown>) => {
+      const [current] = await db.transaction(async (tx) => {
+        await notifyChatActionWork(tx, "failed_run_retry");
+        return tx
+          .update(chatActions)
+          .set({ status, result, updatedAt: new Date() })
+          .where(
+            and(
+              eq(chatActions.id, claimed.id),
+              eq(chatActions.status, "processing"),
+              eq(chatActions.updatedAt, now),
+            ),
+          )
+          .returning();
+      });
       return failedChatRetryReceipt(current ?? claimed);
     };
     try {
@@ -1669,26 +1674,10 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
 
   async function processFailedChatRunRetries(limit = 20) {
     if (isShuttingDown()) return 0;
-    const now = new Date();
     const candidates = await db
       .select({ id: chatActions.id })
       .from(chatActions)
-      .where(
-        and(
-          eq(chatActions.kind, "failed_run_retry"),
-          or(
-            eq(chatActions.status, "issued"),
-            and(
-              eq(chatActions.status, "processing"),
-              lte(
-                chatActions.updatedAt,
-                new Date(now.getTime() - DELIVERY_PROCESSING_STALE_MS),
-              ),
-            ),
-          ),
-          sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
-        ),
-      )
+      .where(dueChatAction("failed_run_retry"))
       .orderBy(asc(chatActions.createdAt), asc(chatActions.id))
       .limit(Math.max(1, Math.min(50, limit)));
     for (const candidate of candidates)

server/src/services/chat-work-notifications.ts

@@ -13,6 +13,7 @@ export async function notifyChatPublicationWork(tx: object): Promise<void> {
 export const CHAT_ACTION_QUEUES = {
   provider_effect: DELIVERY_QUEUES.chatProviderEffects,
   github_webhook_ingress: DELIVERY_QUEUES.chatGitHubIngress,
+  failed_run_retry: DELIVERY_QUEUES.chatFailedRunRetries,
   slack_session_sync: DELIVERY_QUEUES.chatSessionSyncs,
   slack_board_message: DELIVERY_QUEUES.chatBoardMessages,
   slash_task_start: DELIVERY_QUEUES.chatTaskStarts,

server/src/services/delivery-work-notifications.ts

@@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = {
   chatSessionStops: "chat-session-stops",
   chatProviderEffects: "chat-provider-effects",
   chatGitHubIngress: "chat-github-ingress",
+  chatFailedRunRetries: "chat-failed-run-retries",
   chatSessionSyncs: "chat-session-syncs",
   chatVerification: "chat-verification",
   chatInbound: "chat-inbound",