diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index e40716791f3..8177461f43b 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -444,8 +444,13 @@ Longer polling intervals alone do not complete an item. running sessions and pauses without rescheduling until an endpoint change. Tests exercise real producers with deferred fast paths disabled, retry and startup recovery, stale-send quarantine, and zero scans after settlement. - Remaining **1s** maintenance: provider runtimes, Telegram maintenance, failed - chat-run retries, run milestone/wake-notice projection, GitHub operation + Failed chat-run retries now also use commit wakes and durable deadlines from + existing action rows. Issued retries honor `retryAt`; processing claims wait + for both stale-claim expiry and any saved retry time. Startup recovers unfinished + requests. Claim and settlement writes signal in their own short transactions, + with agent wakeup outside those transactions. Empty and terminal queues disarm. + Remaining **1s** maintenance: provider runtimes, Telegram maintenance, + run milestone/wake-notice projection, GitHub operation recovery, and Teams protocol maintenance. GitHub pre-ingress recovery must still inspect provider history: a lost callback has no local queue row. This slice does **not** yet make the whole chat subsystem quiet. Independent diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 8f087241913..1c46075d2aa 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/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("); }); diff --git a/server/src/__tests__/chat-action-work.test.ts b/server/src/__tests__/chat-action-work.test.ts index b96541e79ee..294715f922f 100644 --- a/server/src/__tests__/chat-action-work.test.ts +++ b/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 => null), nextVerificationMessageAt: vi.fn(async (): Promise => 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); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index a550acedf8b..d8c86374466 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/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) => { diff --git a/server/src/services/chat-action-work.ts b/server/src/services/chat-action-work.ts index f794cc6a0c2..6d1847526ed 100644 --- a/server/src/services/chat-action-work.ts +++ b/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`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`case when ${chatActions.status} = 'received' and ${chatEndpoints.status} not in ('paused', 'attention') then 0::numeric end`; if (kind === "slash_task_start") return sql`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; processPendingProviderEffects(): Promise; processPendingGitHubWebhookIngress(): Promise; processPendingSlackSessionSyncs(): Promise; @@ -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 }, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 2bd74bbb9d6..7716efe06aa 100644 --- a/server/src/services/chat-channels.ts +++ b/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), diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index c0b8d43988f..ae041ef9251 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/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) => { - 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) => { + 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) diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index f76ac13c72e..16e2ac75135 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -13,6 +13,7 @@ export async function notifyChatPublicationWork(tx: object): Promise { 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, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index fdee30932f8..ab4ba5fe257 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/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",