diff --git a/extensions/telegram/src/bot-handlers.runtime.ts b/extensions/telegram/src/bot-handlers.runtime.ts index 5ce1f7ec3585..4895eab3d8d5 100644 --- a/extensions/telegram/src/bot-handlers.runtime.ts +++ b/extensions/telegram/src/bot-handlers.runtime.ts @@ -38,7 +38,7 @@ import { parseStrictPositiveInteger } from "openclaw/plugin-sdk/number-runtime"; import { DEFAULT_GROUP_HISTORY_LIMIT } from "openclaw/plugin-sdk/reply-history"; import { resolveAgentRoute } from "openclaw/plugin-sdk/routing"; import { resolveThreadSessionKeys } from "openclaw/plugin-sdk/routing"; -import { danger, logVerbose, warn } from "openclaw/plugin-sdk/runtime-env"; +import { danger, logVerbose, sleepWithAbort, warn } from "openclaw/plugin-sdk/runtime-env"; import { evaluateSupplementalContextVisibility } from "openclaw/plugin-sdk/security-runtime"; import { getSessionEntry, @@ -86,7 +86,9 @@ import type { TelegramAmbientTranscriptWatermark } from "./bot-message-context.t import { parseTelegramNativeCommandCallbackData } from "./bot-native-commands.js"; import type { RegisterTelegramHandlerParams } from "./bot-native-commands.js"; import { + createTelegramSpooledReplayParticipant, createTelegramSpooledReplayDeferredParticipant, + getTelegramSpooledReplayDeferredParticipant, isTelegramSpooledReplayUpdate, recordTelegramMessageProcessingResult, type TelegramMessageProcessingResult, @@ -159,7 +161,6 @@ import { claimTelegramMessageDispatchReplay, commitTelegramMessageDispatchReplay, createTelegramMessageDispatchReplayGuard, - forgetTelegramMessageDispatchReplay, releaseTelegramMessageDispatchReplay, } from "./message-dispatch-dedupe.js"; import { @@ -382,16 +383,14 @@ export const registerTelegramHandlers = ({ error, }); }; - const commitDispatchDedupeKeys = async (keys: readonly string[]) => { + const commitDispatchDedupeKeys = async ( + keys: readonly string[], + options: { requirePersistent?: boolean } = {}, + ) => { await commitTelegramMessageDispatchReplay({ guard: messageDispatchReplayGuard, keys, - }); - }; - const forgetDispatchDedupeKeys = async (keys: readonly string[]) => { - await forgetTelegramMessageDispatchReplay({ - guard: messageDispatchReplayGuard, - keys, + ...options, }); }; const buildFailedProcessingResult = (error: unknown): TelegramMessageProcessingResult => ({ @@ -642,6 +641,7 @@ export const registerTelegramHandlers = ({ ...spooledReplayOptions(spooledReplayParticipants), }, dispatchDedupeKeys: last.dispatchDedupeKeys, + spooledReplayParticipants, }); settleSpooledReplayParticipants(spooledReplayParticipants, result); return; @@ -652,6 +652,9 @@ export const registerTelegramHandlers = ({ .join("\n"); const combinedMedia = entries.flatMap((entry) => entry.allMedia); if (!combinedText.trim() && combinedMedia.length === 0) { + releaseDispatchDedupeKeys( + mergeDispatchDedupeKeys(...entries.map((entry) => entry.dispatchDedupeKeys)), + ); settleSpooledReplayParticipants(spooledReplayParticipants, { kind: "skipped" }); return; } @@ -691,6 +694,7 @@ export const registerTelegramHandlers = ({ dispatchDedupeKeys: mergeDispatchDedupeKeys( ...entries.map((entry) => entry.dispatchDedupeKeys), ), + spooledReplayParticipants, }); settleSpooledReplayParticipants(spooledReplayParticipants, result); } catch (err) { @@ -728,6 +732,9 @@ export const registerTelegramHandlers = ({ } }, onCancel: (items) => { + releaseDispatchDedupeKeys( + mergeDispatchDedupeKeys(...items.map((item) => item.dispatchDedupeKeys)), + ); settleSpooledReplayParticipants( items .map((item) => item.spooledReplayParticipant) @@ -737,9 +744,6 @@ export const registerTelegramHandlers = ({ ), { kind: "skipped" }, ); - releaseDispatchDedupeKeys( - mergeDispatchDedupeKeys(...items.map((item) => item.dispatchDedupeKeys)), - ); }, }); @@ -1106,6 +1110,7 @@ export const registerTelegramHandlers = ({ ...spooledReplayOptions(entry.spooledReplayParticipants), }, dispatchDedupeKeys: entry.dispatchDedupeKeys, + spooledReplayParticipants: entry.spooledReplayParticipants, }); settleSpooledReplayParticipants(entry.spooledReplayParticipants, result); } catch (err) { @@ -1166,6 +1171,7 @@ export const registerTelegramHandlers = ({ ...spooledReplayOptions(entry.spooledReplayParticipants), }, dispatchDedupeKeys: entry.dispatchDedupeKeys, + spooledReplayParticipants: entry.spooledReplayParticipants, }); settleSpooledReplayParticipants(entry.spooledReplayParticipants, result); } catch (err) { @@ -1461,14 +1467,100 @@ export const registerTelegramHandlers = ({ storeAllowFrom: string[]; options?: TelegramMessageContextOptions; dispatchDedupeKeys?: string[]; + spooledReplayParticipants?: readonly TelegramSpooledReplayDeferredParticipant[]; + spooledReplayAbortSignal?: AbortSignal; }): Promise => { let dispatchDedupeCommitted = false; - let dispatchDedupeRollbackAttempted = false; + let spooledReplayFinalResult: TelegramMessageProcessingResult | undefined; + let spooledReplayFinalization: Promise | undefined; + let spooledReplayAdoptionCommitInFlight = false; + let deferredProcessingCancellation: TelegramMessageProcessingResult | undefined; const spooledReplay = - params.options?.spooledReplay === true || isTelegramSpooledReplayUpdate(params.ctx.update); - const forgetCommittedDispatchDedupeKeys = async () => { - dispatchDedupeRollbackAttempted = true; - await forgetDispatchDedupeKeys(params.dispatchDedupeKeys ?? []); + params.options?.spooledReplay === true || + isTelegramSpooledReplayUpdate(params.ctx.update) || + Boolean(params.spooledReplayParticipants?.length); + const explicitParticipants = params.spooledReplayParticipants ?? []; + const frameParticipant = + spooledReplay && + explicitParticipants.length === 0 && + params.options?.isolateSpooledReplaySettlement !== true + ? (getTelegramSpooledReplayDeferredParticipant() ?? + createTelegramSpooledReplayDeferredParticipant( + `message:${params.msg.chat.id}:${params.msg.message_id}`, + ) ?? + undefined) + : undefined; + const processingParticipant = + explicitParticipants.length > 0 + ? createTelegramSpooledReplayParticipant( + `message-processing:${params.msg.chat.id}:${params.msg.message_id}`, + ) + : frameParticipant; + if (processingParticipant && explicitParticipants.length > 0) { + for (const participant of explicitParticipants) { + void participant.task.then((result) => { + if (spooledReplayAdoptionCommitInFlight && result.kind !== "completed") { + deferredProcessingCancellation ??= result; + return; + } + processingParticipant.settle(result); + }); + } + } + const spooledReplayParticipants = [ + ...new Set([ + ...explicitParticipants, + ...(frameParticipant ? [frameParticipant] : []), + ...(processingParticipant ? [processingParticipant] : []), + ]), + ]; + const finalizeSpooledReplayResult = async ( + result: TelegramMessageProcessingResult, + ): Promise => { + if (spooledReplayFinalResult) { + return spooledReplayFinalResult; + } + if (spooledReplayFinalization) { + return await spooledReplayFinalization; + } + const finalization = (async () => { + const finalized = result; + if (result.kind === "completed") { + // Do not cache or settle a durable-adoption failure. Deferred queue + // ownership retries this callback with the same spool participants. + spooledReplayAdoptionCommitInFlight = true; + try { + await commitDispatchDedupeKeys(params.dispatchDedupeKeys ?? [], { + requirePersistent: true, + }); + } catch (error) { + spooledReplayAdoptionCommitInFlight = false; + if (deferredProcessingCancellation) { + processingParticipant?.settle(deferredProcessingCancellation); + } + throw error; + } + spooledReplayAdoptionCommitInFlight = false; + deferredProcessingCancellation = undefined; + dispatchDedupeCommitted = true; + } else { + releaseDispatchDedupeKeys( + params.dispatchDedupeKeys ?? [], + result.kind === "failed-retryable" ? result.error : undefined, + ); + } + spooledReplayFinalResult = finalized; + settleSpooledReplayParticipants(spooledReplayParticipants, finalized); + return finalized; + })(); + spooledReplayFinalization = finalization; + try { + return await finalization; + } finally { + if (!spooledReplayFinalResult && spooledReplayFinalization === finalization) { + spooledReplayFinalization = undefined; + } + } }; try { const replyChainNodes = await buildReplyChainForMessage(params.msg); @@ -1577,20 +1669,29 @@ export const registerTelegramHandlers = ({ await commitDispatchDedupeKeys(params.dispatchDedupeKeys ?? []); dispatchDedupeCommitted = true; }, + spooledReplayAbortSignal: params.spooledReplayAbortSignal, + spooledReplayParticipant: processingParticipant, + finalizeSpooledReplayResult: async (result) => await finalizeSpooledReplayResult(result), + completeSpooledReplayAfterIrrevocableAdoption: async () => { + const completed = { kind: "completed" } satisfies TelegramMessageProcessingResult; + return await finalizeSpooledReplayResult(completed); + }, }, ); + if (spooledReplay) { + return await finalizeSpooledReplayResult(result); + } if (result.kind === "completed" && !dispatchDedupeCommitted) { await commitDispatchDedupeKeys(params.dispatchDedupeKeys ?? []); - } else if (result.kind === "failed-retryable" && dispatchDedupeCommitted && spooledReplay) { - await forgetCommittedDispatchDedupeKeys(); } else if (result.kind !== "completed" && !dispatchDedupeCommitted) { releaseDispatchDedupeKeys(params.dispatchDedupeKeys ?? []); } return result; } catch (err) { - if (dispatchDedupeCommitted && spooledReplay && !dispatchDedupeRollbackAttempted) { - await forgetCommittedDispatchDedupeKeys(); - } else if (!dispatchDedupeCommitted) { + if (spooledReplay) { + return await finalizeSpooledReplayResult(buildFailedProcessingResult(err)); + } + if (!dispatchDedupeCommitted) { releaseDispatchDedupeKeys(params.dispatchDedupeKeys ?? [], err); } throw err; @@ -1746,11 +1847,6 @@ export const registerTelegramHandlers = ({ const TELEGRAM_PLUGIN_CALLBACK_SUBMIT_RETRY_DELAYS_MS = [250, 1000, 2500] as const; const REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE = /reply session initialization conflicted for \S+/u; - const sleep = (ms: number): Promise => - new Promise((resolve) => { - setTimeout(resolve, ms); - }); - const resolvePluginCallbackSubmitText = (submitText: unknown): string | undefined => { if (typeof submitText !== "string") { return undefined; @@ -1771,6 +1867,17 @@ export const registerTelegramHandlers = ({ syntheticMessage: Parameters[0]["msg"]; storeAllowFrom: Parameters[0]["storeAllowFrom"]; }): Promise<"completed" | "skipped"> => { + const spooledReplayParticipant = isTelegramSpooledReplayUpdate(params.syntheticCtx.update) + ? (getTelegramSpooledReplayDeferredParticipant() ?? + createTelegramSpooledReplayDeferredParticipant( + `plugin-callback-submit:${params.callbackId}`, + ) ?? + undefined) + : undefined; + const settleFinalResult = (result: TelegramMessageProcessingResult) => { + spooledReplayParticipant?.settle(result); + return result.kind; + }; for (let attempt = 0; ; attempt++) { try { const result = await processMessageWithReplyChain({ @@ -1780,14 +1887,18 @@ export const registerTelegramHandlers = ({ storeAllowFrom: params.storeAllowFrom, options: { spooledReplay: true, + isolateSpooledReplaySettlement: true, forceWasMentioned: true, messageIdOverride: params.callbackId, }, + spooledReplayAbortSignal: spooledReplayParticipant?.abortSignal, }); if (result.kind === "completed") { + settleFinalResult(result); return "completed"; } if (result.kind === "skipped") { + settleFinalResult(result); return "skipped"; } const retryDelayMs = TELEGRAM_PLUGIN_CALLBACK_SUBMIT_RETRY_DELAYS_MS[attempt]; @@ -1797,17 +1908,18 @@ export const registerTelegramHandlers = ({ logVerbose( `telegram plugin callback submitText hit active reply session; retrying in ${retryDelayMs}ms`, ); - await sleep(retryDelayMs); + await sleepWithAbort(retryDelayMs, spooledReplayParticipant?.abortSignal); continue; } catch (err) { const retryDelayMs = TELEGRAM_PLUGIN_CALLBACK_SUBMIT_RETRY_DELAYS_MS[attempt]; if (!isReplySessionInitConflictError(err) || retryDelayMs === undefined) { + settleFinalResult(buildFailedProcessingResult(err)); throw err; } logVerbose( `telegram plugin callback submitText hit active reply session; retrying in ${retryDelayMs}ms`, ); - await sleep(retryDelayMs); + await sleepWithAbort(retryDelayMs, spooledReplayParticipant?.abortSignal); } } }; @@ -3249,6 +3361,9 @@ export const registerTelegramHandlers = ({ throw err.cause; } runtime.error?.(danger(`callback handler failed: ${String(err)}`)); + if (isTelegramSpooledReplayUpdate(ctx.update)) { + recordTelegramMessageProcessingResult({ kind: "failed-retryable", error: err }); + } } }); @@ -3581,11 +3696,12 @@ export const registerTelegramHandlers = ({ } catch (err) { releaseDispatchDedupeKeys(dispatchDedupeKeys, err); runtime.error?.(danger(`${event.errorMessage}: ${String(err)}`)); - if (err instanceof TelegramPairingStoreReadError) { + const spooledReplay = isTelegramSpooledReplayUpdate(event.ctx.update); + if (err instanceof TelegramPairingStoreReadError || spooledReplay) { recordTelegramMessageProcessingResult({ kind: "failed-retryable", error: err }); // Spooled replays are durably retried; live updates get one apology // because they are acked without replay. - if (isTelegramSpooledReplayUpdate(event.ctx.update)) { + if (spooledReplay) { return; } await withTelegramApiErrorLogging({ diff --git a/extensions/telegram/src/bot-message-context.types.ts b/extensions/telegram/src/bot-message-context.types.ts index e6e4e7b5a844..4865e7757d55 100644 --- a/extensions/telegram/src/bot-message-context.types.ts +++ b/extensions/telegram/src/bot-message-context.types.ts @@ -29,6 +29,8 @@ export type TelegramMessageContextOptions = { promptContextAmbientWatermark?: TelegramAmbientTranscriptWatermark; ambientTranscriptBody?: string; spooledReplay?: boolean; + /** Use an attempt-local participant so an outer retry loop owns final spool settlement. */ + isolateSpooledReplaySettlement?: boolean; }; export type TelegramPromptContextEntry = NonNullable< diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts index a2da555a97fc..5c78cfd9c5c8 100644 --- a/extensions/telegram/src/bot-message-dispatch.ts +++ b/extensions/telegram/src/bot-message-dispatch.ts @@ -248,6 +248,12 @@ type DispatchTelegramMessageParams = { suppressFailureFallback?: boolean; /** Fires after recovery-relevant session/run state is durably persisted. */ onTurnAdopted?: () => void | Promise; + /** Marks a queued follow-up whose adoption will happen at reply-lane admission. */ + onTurnDeferred?: () => void; + /** Releases a deferred turn that completed without ever owning the reply lane. */ + onTurnAbandoned?: () => void; + /** Cancels queued/model work when ingress ownership fails before adoption. */ + turnAbortSignal?: AbortSignal; }; type TelegramDispatchResult = { kind: "completed" } | { kind: "failed-retryable"; error: unknown }; @@ -788,6 +794,9 @@ export const dispatchTelegramMessage = async ({ retryDispatchErrors = false, suppressFailureFallback = false, onTurnAdopted, + onTurnDeferred, + onTurnAbandoned, + turnAbortSignal, }: DispatchTelegramMessageParams): Promise => { const dispatchStartedAt = Date.now(); const dispatchContext = resolveDispatchTelegramContext({ context }); @@ -872,7 +881,11 @@ export const dispatchTelegramMessage = async ({ let activeReplyFenceKey = replyFenceKey.activeKey; let replyFenceGeneration: number | undefined; const replyAbortController = new AbortController(); + const replyAbortSignal = turnAbortSignal + ? AbortSignal.any([replyAbortController.signal, turnAbortSignal]) + : replyAbortController.signal; let replyAbortControllerQueued = false; + let queuedTurnAdmitted = false; let dispatchWasSuperseded; const isDispatchSuperseded = () => replyFenceGeneration !== undefined && @@ -2615,26 +2628,39 @@ export const dispatchTelegramMessage = async ({ replyOptions: { skillFilter, disableBlockStreaming, - abortSignal: replyAbortController.signal, + abortSignal: replyAbortSignal, ...(onTurnAdopted ? { onTurnAdopted } : {}), sourceReplyDeliveryMode: isRoomEvent ? "message_tool_only" : undefined, queuedDeliveryCorrelations: isRoomEvent ? [{ begin: beginDeliveryCorrelation }] : undefined, - queuedFollowupLifecycle: isRoomEvent - ? { - onEnqueued: () => { - replyAbortControllerQueued = true; - }, - onComplete: () => { - replyAbortControllerQueued = false; - releaseTelegramReplyFenceAbortController( - activeReplyFenceKey, - replyAbortController, - ); - }, - } - : undefined, + queuedFollowupLifecycle: + isRoomEvent || onTurnAdopted || onTurnDeferred || onTurnAbandoned + ? { + onEnqueued: () => { + if (isRoomEvent) { + replyAbortControllerQueued = true; + } + onTurnDeferred?.(); + }, + onAdmitted: async () => { + await onTurnAdopted?.(); + queuedTurnAdmitted = true; + }, + onComplete: () => { + if (isRoomEvent) { + replyAbortControllerQueued = false; + releaseTelegramReplyFenceAbortController( + activeReplyFenceKey, + replyAbortController, + ); + } + if (!queuedTurnAdmitted) { + onTurnAbandoned?.(); + } + }, + } + : undefined, suppressTyping: isRoomEvent, onPartialReply: answerLane.stream || reasoningLane.stream diff --git a/extensions/telegram/src/bot-message.test.ts b/extensions/telegram/src/bot-message.test.ts index 2cb5e78841f3..e0432552d030 100644 --- a/extensions/telegram/src/bot-message.test.ts +++ b/extensions/telegram/src/bot-message.test.ts @@ -1,10 +1,17 @@ // Telegram tests cover bot message plugin behavior. import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import type { TelegramBotDeps } from "./bot-deps.js"; +import type { + TelegramMessageProcessingResult, + TelegramSpooledReplayDeferredParticipant, +} from "./bot-processing-outcome.js"; const buildTelegramMessageContext = vi.hoisted(() => vi.fn()); const dispatchTelegramMessage = vi.hoisted(() => vi.fn()); const telegramInboundInfo = vi.hoisted(() => vi.fn()); +const sleepWithAbort = vi.hoisted(() => + vi.fn<(delayMs: number, signal?: AbortSignal) => Promise>(async () => undefined), +); const upsertChannelPairingRequest = vi.hoisted(() => vi.fn(async () => ({ code: "PAIRCODE", created: true })), ); @@ -15,9 +22,11 @@ vi.mock("openclaw/plugin-sdk/runtime-env", () => ({ info: telegramInboundInfo, }), }), + computeBackoff: vi.fn((_policy: unknown, attempt: number) => attempt), danger: (message: string) => message, logVerbose: vi.fn(), shouldLogVerbose: () => false, + sleepWithAbort, })); vi.mock("./bot-message-context.js", () => ({ @@ -30,6 +39,7 @@ vi.mock("./bot-message-dispatch.js", () => ({ let createTelegramMessageProcessor: typeof import("./bot-message.js").createTelegramMessageProcessor; let formatTelegramInboundLogLine: typeof import("./bot-message.js").formatTelegramInboundLogLine; +let createTelegramSpooledReplayDeferredParticipant: typeof import("./bot-processing-outcome.js").createTelegramSpooledReplayDeferredParticipant; let runWithTelegramUpdateProcessingFrame: typeof import("./bot-processing-outcome.js").runWithTelegramUpdateProcessingFrame; let runWithTelegramSpooledReplayUpdate: typeof import("./bot-processing-outcome.js").runWithTelegramSpooledReplayUpdate; @@ -37,14 +47,18 @@ describe("telegram bot message processor", () => { beforeAll(async () => { ({ createTelegramMessageProcessor, formatTelegramInboundLogLine } = await import("./bot-message.js")); - ({ runWithTelegramUpdateProcessingFrame, runWithTelegramSpooledReplayUpdate } = - await import("./bot-processing-outcome.js")); + ({ + createTelegramSpooledReplayDeferredParticipant, + runWithTelegramUpdateProcessingFrame, + runWithTelegramSpooledReplayUpdate, + } = await import("./bot-processing-outcome.js")); }); beforeEach(() => { buildTelegramMessageContext.mockClear(); dispatchTelegramMessage.mockClear(); telegramInboundInfo.mockClear(); + sleepWithAbort.mockReset().mockResolvedValue(undefined); upsertChannelPairingRequest.mockClear(); }); @@ -289,12 +303,15 @@ describe("telegram bot message processor", () => { sendMessage, ); const update = { update_id: 123456 }; - // Spooled agent turns detach at adoption: processMessage returns once the - // deferred participant is registered; the real result is on deferred.task. + // Direct spooled turns return their adoption/pre-adoption outcome so the + // reply-chain owner can roll back dedupe before a retry. const replay = await runWithTelegramSpooledReplayUpdate(update, async () => processSampleMessage(processMessage, undefined, { update }), ); - expect(replay.value).toEqual({ kind: "completed" }); + expect(replay.value).toEqual({ + kind: "failed-retryable", + error: dispatchError, + }); expect(replay.deferredWork).toBeDefined(); await expect(replay.deferredWork!.task).resolves.toEqual({ kind: "failed-retryable", @@ -306,6 +323,359 @@ describe("telegram bot message processor", () => { ); }); + it("finalizes spooled adoption before settling the ingress participant", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const events: string[] = []; + const finalizeSpooledReplayResult = vi.fn( + async ( + result: TelegramMessageProcessingResult, + phase: "adopted" | "terminal", + ): Promise => { + events.push(`finalizer:${phase}`); + return result; + }, + ); + dispatchTelegramMessage.mockImplementationOnce(async ({ onTurnAdopted }) => { + await onTurnAdopted?.(); + return { kind: "completed" }; + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + const update = { update_id: 123458 }; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => { + const participant = createTelegramSpooledReplayDeferredParticipant("test:finalizer-order"); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + const settle = participant.settle; + participant.settle = (result) => { + events.push(`participant:${result.kind}`); + settle(result); + }; + return await processSampleMessage( + processMessage, + { finalizeSpooledReplayResult }, + { update }, + ); + }); + + expect(replay.value).toEqual({ kind: "completed" }); + await expect(replay.deferredWork?.task).resolves.toEqual({ kind: "completed" }); + expect(events).toEqual(["finalizer:adopted", "participant:completed"]); + expect(finalizeSpooledReplayResult).toHaveBeenCalledTimes(1); + }); + + it("keeps a spooled replay completed when dispatch fails after adoption", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const lateError = new Error("late dispatch failure"); + const finalizeSpooledReplayResult = vi.fn( + async (result: TelegramMessageProcessingResult): Promise => + result, + ); + dispatchTelegramMessage.mockImplementationOnce(async ({ onTurnAdopted }) => { + await onTurnAdopted?.(); + return { kind: "failed-retryable", error: lateError }; + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + const update = { update_id: 123459 }; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => + processSampleMessage(processMessage, { finalizeSpooledReplayResult }, { update }), + ); + + expect(replay.value).toEqual({ kind: "completed" }); + await expect(replay.deferredWork?.task).resolves.toEqual({ kind: "completed" }); + expect(finalizeSpooledReplayResult).toHaveBeenCalledTimes(1); + expect(finalizeSpooledReplayResult).toHaveBeenCalledWith({ kind: "completed" }, "adopted"); + }); + + it("retries durable replay protection after an active steer already committed", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const finalizerError = new Error("dedupe commit failed"); + const finalizeSpooledReplayResult = vi + .fn( + async ( + result: TelegramMessageProcessingResult, + _phase: "adopted" | "terminal", + ): Promise => result, + ) + .mockRejectedValueOnce(finalizerError); + const completeSpooledReplayAfterIrrevocableAdoption = vi.fn( + async () => await finalizeSpooledReplayResult({ kind: "completed" }, "adopted"), + ); + dispatchTelegramMessage.mockImplementationOnce(async ({ onTurnAdopted }) => { + await expect(onTurnAdopted?.()).rejects.toBe(finalizerError); + return { kind: "completed" }; + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + const update = { update_id: 1234591 }; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => + processSampleMessage( + processMessage, + { + finalizeSpooledReplayResult, + completeSpooledReplayAfterIrrevocableAdoption, + }, + { update }, + ), + ); + + expect(replay.value).toEqual({ kind: "completed" }); + await expect(replay.deferredWork?.task).resolves.toEqual({ kind: "completed" }); + expect(finalizeSpooledReplayResult).toHaveBeenCalledTimes(2); + expect(completeSpooledReplayAfterIrrevocableAdoption).toHaveBeenCalledWith(finalizerError); + }); + + it("retries active-steer durable replay protection through multiple transient failures", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const finalizerError = new Error("dedupe commit failed"); + const firstRetryError = new Error("first dedupe retry failed"); + const secondRetryError = new Error("second dedupe retry failed"); + const finalizeSpooledReplayResult = vi.fn(async () => { + throw finalizerError; + }); + const completeSpooledReplayAfterIrrevocableAdoption = vi + .fn<() => Promise>() + .mockRejectedValueOnce(firstRetryError) + .mockRejectedValueOnce(secondRetryError) + .mockResolvedValue({ kind: "completed" }); + dispatchTelegramMessage.mockImplementationOnce(async ({ onTurnAdopted }) => { + await expect(onTurnAdopted?.()).rejects.toBe(finalizerError); + return { kind: "completed" }; + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + const update = { update_id: 1234592 }; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => + processSampleMessage( + processMessage, + { + finalizeSpooledReplayResult, + completeSpooledReplayAfterIrrevocableAdoption, + }, + { update }, + ), + ); + + expect(replay.value).toEqual({ kind: "completed" }); + await expect(replay.deferredWork?.task).resolves.toEqual({ kind: "completed" }); + expect(completeSpooledReplayAfterIrrevocableAdoption).toHaveBeenCalledTimes(3); + expect(completeSpooledReplayAfterIrrevocableAdoption.mock.calls).toEqual([ + [finalizerError], + [firstRetryError], + [secondRetryError], + ]); + expect(sleepWithAbort).toHaveBeenCalledTimes(2); + }); + + it("stops active-steer durable replay retries when the outer spool owner is cancelled", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const finalizerError = new Error("dedupe commit failed"); + const retryError = new Error("dedupe retry failed"); + const cancellationError = new Error("outer spool timeout"); + const outerAbortController = new AbortController(); + const finalizeSpooledReplayResult = vi.fn(async () => { + throw finalizerError; + }); + const completeSpooledReplayAfterIrrevocableAdoption = vi + .fn<() => Promise>() + .mockRejectedValue(retryError); + dispatchTelegramMessage.mockImplementationOnce(async ({ onTurnAdopted }) => { + await expect(onTurnAdopted?.()).rejects.toBe(finalizerError); + return { kind: "completed" }; + }); + sleepWithAbort.mockImplementationOnce(async (_delayMs, signal) => { + outerAbortController.abort(cancellationError); + if (signal?.aborted) { + throw signal.reason; + } + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + + const processing = processSampleMessage( + processMessage, + { + finalizeSpooledReplayResult, + completeSpooledReplayAfterIrrevocableAdoption, + spooledReplayAbortSignal: outerAbortController.signal, + }, + {}, + { spooledReplay: true, isolateSpooledReplaySettlement: true }, + ); + + await expect(processing).resolves.toEqual({ + kind: "failed-retryable", + error: cancellationError, + }); + expect(completeSpooledReplayAfterIrrevocableAdoption).toHaveBeenCalledTimes(1); + expect(sleepWithAbort).toHaveBeenCalledTimes(1); + }); + + it("retries deferred adoption after finalization rejects without settling ingress", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const finalizerError = new Error("dedupe commit failed"); + const finalizeSpooledReplayResult = vi + .fn( + async (result: TelegramMessageProcessingResult): Promise => + result, + ) + .mockRejectedValueOnce(finalizerError); + let participantSettles = 0; + let settledAfterFirstAdmission = false; + let firstAdmissionError: unknown; + let secondAdmissionError: unknown; + let thirdAdmissionError: unknown; + dispatchTelegramMessage.mockImplementationOnce(async ({ onTurnAdopted, onTurnDeferred }) => { + onTurnDeferred?.(); + try { + await onTurnAdopted?.(); + } catch (error) { + firstAdmissionError = error; + } + settledAfterFirstAdmission = participantSettles > 0; + try { + await onTurnAdopted?.(); + } catch (error) { + secondAdmissionError = error; + } + try { + await onTurnAdopted?.(); + } catch (error) { + thirdAdmissionError = error; + } + return { kind: "completed" }; + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + const update = { update_id: 123460 }; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => { + const participant = createTelegramSpooledReplayDeferredParticipant( + "test:adoption-finalizer-retry", + ); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + const settle = participant.settle; + participant.settle = (result) => { + participantSettles += 1; + settle(result); + }; + return await processSampleMessage( + processMessage, + { finalizeSpooledReplayResult }, + { update }, + ); + }); + + expect(firstAdmissionError).toBe(finalizerError); + expect(settledAfterFirstAdmission).toBe(false); + expect(secondAdmissionError).toBeUndefined(); + expect(thirdAdmissionError).toBeUndefined(); + expect(finalizeSpooledReplayResult).toHaveBeenCalledTimes(2); + expect(participantSettles).toBe(1); + expect(replay.value).toEqual({ kind: "completed" }); + await expect(replay.deferredWork?.task).resolves.toEqual({ kind: "completed" }); + }); + + it("settles an abandoned deferred turn as skipped", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const finalizeSpooledReplayResult = vi.fn( + async (result: TelegramMessageProcessingResult): Promise => + result, + ); + dispatchTelegramMessage.mockImplementationOnce(async ({ onTurnAbandoned, onTurnDeferred }) => { + onTurnDeferred?.(); + onTurnAbandoned?.(); + return { kind: "completed" }; + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + const update = { update_id: 123461 }; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => + processSampleMessage(processMessage, { finalizeSpooledReplayResult }, { update }), + ); + + expect(replay.value).toEqual({ kind: "skipped" }); + await expect(replay.deferredWork?.task).resolves.toEqual({ kind: "skipped" }); + expect(finalizeSpooledReplayResult).toHaveBeenCalledTimes(1); + expect(finalizeSpooledReplayResult).toHaveBeenCalledWith({ kind: "skipped" }, "terminal"); + }); + + it("keeps isolated retry settlement separate from the outer spool participant", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const retryError = new Error("retry this attempt"); + dispatchTelegramMessage.mockResolvedValueOnce({ + kind: "failed-retryable", + error: retryError, + }); + const processMessage = createTelegramMessageProcessor(baseDeps); + const update = { update_id: 123462 }; + let outerSettles = 0; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => { + const participant = createTelegramSpooledReplayDeferredParticipant("test:outer-retry"); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + const settle = participant.settle; + participant.settle = (result) => { + outerSettles += 1; + settle(result); + }; + return await processSampleMessage( + processMessage, + undefined, + { update }, + { spooledReplay: true, isolateSpooledReplaySettlement: true }, + ); + }); + + expect(replay.value).toEqual({ kind: "failed-retryable", error: retryError }); + expect(outerSettles).toBe(0); + const outerParticipant = replay.deferredWork; + expect(outerParticipant).toBeDefined(); + outerParticipant?.settle({ kind: "skipped" }); + await expect(outerParticipant?.task).resolves.toEqual({ kind: "skipped" }); + }); + + it("aborts an isolated queued retry when its outer spool owner is cancelled", async () => { + buildTelegramMessageContext.mockResolvedValue(createMessageContext()); + const outerAbortController = new AbortController(); + let markDeferred: (() => void) | undefined; + const deferred = new Promise((resolve) => { + markDeferred = resolve; + }); + let queuedAbortSignal: AbortSignal | undefined; + dispatchTelegramMessage.mockImplementationOnce( + async ({ onTurnAbandoned, onTurnDeferred, turnAbortSignal }) => { + queuedAbortSignal = turnAbortSignal; + onTurnDeferred?.(); + markDeferred?.(); + if (!turnAbortSignal?.aborted) { + await new Promise((resolve) => { + turnAbortSignal?.addEventListener("abort", () => resolve(), { once: true }); + }); + } + onTurnAbandoned?.(); + return { kind: "completed" }; + }, + ); + const processMessage = createTelegramMessageProcessor(baseDeps); + + const processing = processSampleMessage( + processMessage, + { spooledReplayAbortSignal: outerAbortController.signal }, + {}, + { spooledReplay: true, isolateSpooledReplaySettlement: true }, + ); + await deferred; + outerAbortController.abort(new Error("outer spool timeout")); + + await expect(processing).resolves.toEqual({ kind: "skipped" }); + expect(queuedAbortSignal?.aborted).toBe(true); + }); + it("suppresses user-visible fallback for synthetic buffered spooled replay contexts", async () => { const sendMessage = vi.fn().mockResolvedValue(undefined); const { processMessage, runtimeError, dispatchError } = createDispatchFailureHarness( @@ -368,7 +738,10 @@ describe("telegram bot message processor", () => { const replay = await runWithTelegramSpooledReplayUpdate(update, async () => processSampleMessage(processMessage, undefined, { update }), ); - expect(replay.value).toEqual({ kind: "completed" }); + expect(replay.value).toEqual({ + kind: "failed-retryable", + error: dispatchError, + }); expect(replay.deferredWork).toBeDefined(); await expect(replay.deferredWork!.task).resolves.toEqual({ kind: "failed-retryable", diff --git a/extensions/telegram/src/bot-message.ts b/extensions/telegram/src/bot-message.ts index f22f9a112ad8..09aaeedc18c2 100644 --- a/extensions/telegram/src/bot-message.ts +++ b/extensions/telegram/src/bot-message.ts @@ -6,6 +6,7 @@ import { danger, logVerbose, shouldLogVerbose, + sleepWithAbort, } from "openclaw/plugin-sdk/runtime-env"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import type { TelegramBotDeps } from "./bot-deps.js"; @@ -18,16 +19,19 @@ import type { TelegramMessageContextOptions } from "./bot-message-context.types. import type { TelegramPromptContextEntry } from "./bot-message-context.types.js"; import { dispatchTelegramMessage } from "./bot-message-dispatch.js"; import { + createTelegramSpooledReplayParticipant, createTelegramSpooledReplayDeferredParticipant, getTelegramSpooledReplayDeferredParticipant, isTelegramSpooledReplayUpdate, recordTelegramMessageProcessingResult, type TelegramMessageProcessingResult, + type TelegramSpooledReplayDeferredParticipant, } from "./bot-processing-outcome.js"; import type { TelegramBotOptions } from "./bot.types.js"; import { buildTelegramThreadParams } from "./bot/helpers.js"; import type { TelegramContext, TelegramStreamMode } from "./bot/types.js"; import type { TelegramReplyChainEntry } from "./message-cache.js"; +import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./spooled-update-retry-policy.js"; const telegramInboundLog = createSubsystemLogger("gateway/channels/telegram").child("inbound"); @@ -57,6 +61,16 @@ type TelegramMessageProcessorDeps = Omit< export type TelegramMessageProcessorLifecycle = { onDispatchStart?: () => Promise | void; + /** One-way cancellation from an outer spool owner into an isolated retry attempt. */ + spooledReplayAbortSignal?: AbortSignal; + spooledReplayParticipant?: TelegramSpooledReplayDeferredParticipant; + finalizeSpooledReplayResult?: ( + result: TelegramMessageProcessingResult, + phase: "adopted" | "terminal", + ) => Promise; + completeSpooledReplayAfterIrrevocableAdoption?: ( + error: unknown, + ) => Promise | TelegramMessageProcessingResult; }; export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDeps) => { @@ -201,11 +215,16 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep mediaType: allMedia[0]?.contentType, }), ); - await lifecycle?.onDispatchStart?.(); const spooledReplay = options?.spooledReplay === true || isTelegramSpooledReplayUpdate(primaryCtx.update); + if (!spooledReplay) { + await lifecycle?.onDispatchStart?.(); + } const runDispatch = async (params: { onTurnAdopted?: () => void | Promise; + onTurnDeferred?: () => void; + onTurnAbandoned?: () => void; + turnAbortSignal?: AbortSignal; }): Promise => { try { const dispatchResult = await dispatchTelegramMessage({ @@ -222,6 +241,9 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep retryDispatchErrors: spooledReplay, suppressFailureFallback: spooledReplay, onTurnAdopted: params.onTurnAdopted, + onTurnDeferred: params.onTurnDeferred, + onTurnAbandoned: params.onTurnAbandoned, + turnAbortSignal: params.turnAbortSignal, }); if (dispatchResult?.kind === "failed-retryable") { const result: TelegramMessageProcessingResult = { @@ -264,40 +286,168 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep // persisted), not settle. The deferred participant hands ownership back to // the spool drain so the per-chat lane frees while the agent turn continues. if (spooledReplay) { - const existingParticipant = getTelegramSpooledReplayDeferredParticipant(); + const existingParticipant = + lifecycle?.spooledReplayParticipant ?? + (options?.isolateSpooledReplaySettlement + ? undefined + : getTelegramSpooledReplayDeferredParticipant()); const participant = existingParticipant ?? - createTelegramSpooledReplayDeferredParticipant( + (options?.isolateSpooledReplaySettlement + ? undefined + : createTelegramSpooledReplayDeferredParticipant( + `agent-turn:${context.chatId}:${context.ctxPayload.MessageSid ?? Date.now()}`, + )) ?? + createTelegramSpooledReplayParticipant( `agent-turn:${context.chatId}:${context.ctxPayload.MessageSid ?? Date.now()}`, ); - if (participant) { - let adopted = false; - const settleIfNeeded = (result: TelegramMessageProcessingResult) => { - if (adopted) { - return; - } - participant.settle(result); - }; - const run = async () => { - const result = await runDispatch({ - onTurnAdopted: async () => { - if (adopted) { - return; - } - adopted = true; - participant.settle({ kind: "completed" }); - }, - }); - settleIfNeeded(result); - return result; - }; - if (existingParticipant) { - return await run(); + let adopted = false; + let adoptionAttempted = false; + let adoptionFinalizationError: unknown; + let deferred = false; + let settledResult: TelegramMessageProcessingResult | undefined; + let settlement: Promise | undefined; + const settle = async ( + result: TelegramMessageProcessingResult, + phase: "adopted" | "terminal", + ): Promise => { + if (settledResult) { + return settledResult; } - void run(); - const detached: TelegramMessageProcessingResult = { kind: "completed" }; - return detached; - } + if (settlement) { + return await settlement; + } + settlement = (async () => { + let finalized: TelegramMessageProcessingResult; + try { + finalized = lifecycle?.finalizeSpooledReplayResult + ? await lifecycle.finalizeSpooledReplayResult(result, phase) + : result; + } catch (error) { + finalized = { kind: "failed-retryable", error }; + } + // A deferred queue item still owns the turn when its admission + // callback fails. Leave the spool participant pending so the queue + // can retry admission without creating a second ingress owner. + if (phase === "adopted" && finalized.kind !== "completed") { + return finalized; + } + if (phase === "adopted" && finalized.kind === "completed") { + adopted = true; + } + settledResult = finalized; + participant.settle(finalized); + return finalized; + })(); + try { + return await settlement; + } finally { + if (!settledResult) { + settlement = undefined; + } + } + }; + const run = async () => { + const turnAbortSignal = lifecycle?.spooledReplayAbortSignal + ? AbortSignal.any([participant.abortSignal, lifecycle.spooledReplayAbortSignal]) + : participant.abortSignal; + const result = await runDispatch({ + turnAbortSignal, + onTurnAdopted: async () => { + if (adopted) { + return; + } + adoptionAttempted = true; + const adoptedResult = await settle({ kind: "completed" }, "adopted"); + if (adoptedResult.kind !== "completed") { + adoptionFinalizationError = + adoptedResult.kind === "failed-retryable" + ? adoptedResult.error + : new Error("telegram spooled turn adoption was not completed"); + throw adoptedResult.kind === "failed-retryable" + ? adoptedResult.error + : new Error("telegram spooled turn adoption was not completed"); + } + }, + onTurnDeferred: () => { + deferred = true; + }, + onTurnAbandoned: () => { + if (!adopted) { + void settle({ kind: "skipped" }, "terminal"); + } + }, + }); + if (adopted) { + return { kind: "completed" } satisfies TelegramMessageProcessingResult; + } + if (settledResult) { + return settledResult; + } + if (adoptionAttempted && !deferred && result.kind === "completed") { + runtime.error?.( + danger( + `telegram spooled turn adoption finalization failed after active steer commit: ${String( + adoptionFinalizationError, + )}`, + ), + ); + let retryError = adoptionFinalizationError; + let retryAttempt = 0; + while (!turnAbortSignal.aborted) { + retryAttempt += 1; + try { + const completed = + (await lifecycle?.completeSpooledReplayAfterIrrevocableAdoption?.(retryError)) ?? + ({ kind: "completed" } satisfies TelegramMessageProcessingResult); + if (completed.kind === "completed") { + adopted = true; + settledResult = completed; + participant.settle(completed); + return completed; + } + retryError = + completed.kind === "failed-retryable" + ? completed.error + : new Error("telegram spooled turn adoption was not completed"); + } catch (error) { + retryError = error; + } + const delayMs = resolveSpooledUpdatePersistenceRetryDelayMs(retryAttempt); + runtime.error?.( + danger( + `telegram spooled turn durable replay protection retry ${retryAttempt} failed after active steer commit; retrying in ${delayMs}ms: ${String(retryError)}`, + ), + ); + try { + await sleepWithAbort(delayMs, turnAbortSignal); + } catch { + break; + } + } + if (turnAbortSignal.aborted && !participant.abortSignal.aborted) { + const abortResult: TelegramMessageProcessingResult = + turnAbortSignal.reason === "skipped" + ? { kind: "skipped" } + : { + kind: "failed-retryable", + error: + turnAbortSignal.reason ?? + new Error("telegram spooled replay owner cancelled"), + }; + participant.settle(abortResult); + } + return await participant.task; + } + if (deferred) { + return await participant.task; + } + return await settle(result, "terminal"); + }; + // The participant is the ingress ownership boundary. Direct and buffered + // callers both return when it is durably adopted or terminally rejected. + void run(); + return await participant.task; } return await runDispatch({}); diff --git a/extensions/telegram/src/bot-processing-outcome.ts b/extensions/telegram/src/bot-processing-outcome.ts index 9f33fd8597f9..cf91f901ed7b 100644 --- a/extensions/telegram/src/bot-processing-outcome.ts +++ b/extensions/telegram/src/bot-processing-outcome.ts @@ -16,6 +16,7 @@ type TelegramSpooledReplayFrame = { export type TelegramSpooledReplayDeferredParticipant = { key: string; + abortSignal: AbortSignal; task: Promise; settle: (result: TelegramMessageProcessingResult) => void; }; @@ -58,9 +59,10 @@ export function recordTelegramMessageProcessingResult( } } -function createTelegramSpooledReplayParticipant( +export function createTelegramSpooledReplayParticipant( key: string, ): TelegramSpooledReplayDeferredParticipant { + const abortController = new AbortController(); let settled = false; let resolveTask: (result: TelegramMessageProcessingResult) => void = () => {}; const task = new Promise((resolve) => { @@ -68,12 +70,16 @@ function createTelegramSpooledReplayParticipant( }); return { key, + abortSignal: abortController.signal, task, settle: (result) => { if (settled) { return; } settled = true; + if (result.kind !== "completed") { + abortController.abort(result.kind === "failed-retryable" ? result.error : result.kind); + } resolveTask(result); }, }; diff --git a/extensions/telegram/src/bot.create-telegram-bot.test.ts b/extensions/telegram/src/bot.create-telegram-bot.test.ts index 9c88ad7848bc..6566d990559a 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.test.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.test.ts @@ -58,6 +58,7 @@ type BuildModelsProviderDataMock = ReturnType< typeof vi.fn> >; const { resolveTelegramFetch } = await import("./fetch.js"); +const messageDispatchDedupe = await import("./message-dispatch-dedupe.js"); const { createTelegramBotCore: createTelegramBotBase, getTelegramSequentialKey, @@ -1068,6 +1069,418 @@ describe("createTelegramBot", () => { await expect(deferredWork.task).resolves.toEqual({ kind: "skipped" }); }); + it("keeps forced text-fragment flush settlement isolated from the triggering replay", async () => { + loadConfig.mockReturnValue({ + agents: { + defaults: { + envelopeTimezone: "utc", + }, + }, + channels: { + telegram: { dmPolicy: "open", allowFrom: ["*"] }, + }, + }); + + installPerKeySequentializer(); + const secondDispatchError = new Error("triggering replay failed before adoption"); + replySpy + .mockResolvedValueOnce({ text: "buffered replay completed" }) + .mockRejectedValueOnce(secondDispatchError); + + createTelegramBot({ token: "tok" }); + const messageHandler = getOnHandler("message") as ( + ctx: TelegramMiddlewareTestContext, + ) => Promise; + const dispatchSpooledMessage = async (params: { + updateId: number; + messageId: number; + text: string; + }) => { + const message = { + chat: { id: 7, type: "private" as const }, + text: params.text, + date: 1736380800 + params.updateId, + message_id: params.messageId, + from: { id: 42, first_name: "Ada" }, + }; + const update = { update_id: params.updateId, message }; + return await runWithTelegramSpooledReplayUpdate(update, async () => { + await runTelegramMiddlewareChain({ + ctx: { + update, + message, + me: { username: "openclaw_bot" }, + getFile: async () => ({}), + }, + finalHandler: messageHandler, + }); + }); + }; + + const bufferedReplay = await dispatchSpooledMessage({ + updateId: 213, + messageId: 213, + text: "A".repeat(4050), + }); + const bufferedParticipant = requireValue( + bufferedReplay.deferredWork, + "buffered replay participant", + ); + + const triggeringReplay = await dispatchSpooledMessage({ + updateId: 214, + messageId: 215, + text: "B", + }); + const triggeringParticipant = requireValue( + triggeringReplay.deferredWork, + "triggering replay participant", + ); + + expect(triggeringParticipant).not.toBe(bufferedParticipant); + await expect(bufferedParticipant.task).resolves.toEqual({ kind: "completed" }); + await expect(triggeringParticipant.task).resolves.toEqual({ + kind: "failed-retryable", + error: secondDispatchError, + }); + expect(replySpy).toHaveBeenCalledTimes(2); + }); + + it("retries deferred adoption after durable commit fails without settling buffered participants", async () => { + const DEBOUNCE_MS = 4321; + loadConfig.mockReturnValue({ + agents: { + defaults: { + envelopeTimezone: "utc", + }, + }, + messages: { + inbound: { + debounceMs: DEBOUNCE_MS, + }, + }, + channels: { + telegram: { dmPolicy: "open", allowFrom: ["*"] }, + }, + }); + + installPerKeySequentializer(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const commitError = new Error("durable dispatch commit failed"); + const commitSpy = vi + .spyOn(messageDispatchDedupe, "commitTelegramMessageDispatchReplay") + .mockRejectedValueOnce(commitError); + let queuedLifecycle: GetReplyOptions["queuedFollowupLifecycle"]; + replySpy.mockImplementationOnce(async (_ctx: MsgContext, opts?: GetReplyOptions) => { + queuedLifecycle = opts?.queuedFollowupLifecycle; + queuedLifecycle?.onEnqueued?.(); + return undefined; + }); + + try { + createTelegramBot({ token: "tok" }); + const messageHandler = getOnHandler("message") as ( + ctx: TelegramMiddlewareTestContext, + ) => Promise; + const dispatchSpooledMessage = async (updateId: number, text: string) => { + const update = { update_id: updateId }; + return await runWithTelegramSpooledReplayUpdate(update, async () => { + await runTelegramMiddlewareChain({ + ctx: { + update, + message: { + chat: { id: 7, type: "private" }, + text, + date: 1736380800 + updateId, + message_id: updateId, + from: { id: 42, first_name: "Ada" }, + }, + me: { username: "openclaw_bot" }, + getFile: async () => ({}), + }, + finalHandler: messageHandler, + }); + }); + }; + + const firstReplay = await dispatchSpooledMessage(221, "first buffered message"); + const secondReplay = await dispatchSpooledMessage(222, "second buffered message"); + const firstParticipant = requireValue( + firstReplay.deferredWork, + "first buffered replay participant", + ); + const secondParticipant = requireValue( + secondReplay.deferredWork, + "second buffered replay participant", + ); + let firstSettled = false; + let secondSettled = false; + void firstParticipant.task.then(() => { + firstSettled = true; + }); + void secondParticipant.task.then(() => { + secondSettled = true; + }); + + const debounceCallIndex = setTimeoutSpy.mock.calls.findLastIndex( + (call) => call[1] === DEBOUNCE_MS, + ); + expect(debounceCallIndex).toBeGreaterThanOrEqual(0); + clearTimeout( + setTimeoutSpy.mock.results[debounceCallIndex]?.value as ReturnType, + ); + const flush = setTimeoutSpy.mock.calls[debounceCallIndex]?.[0] as (() => void) | undefined; + flush?.(); + await vi.waitFor(() => { + expect(queuedLifecycle?.onAdmitted).toEqual(expect.any(Function)); + }); + + await expect(queuedLifecycle?.onAdmitted?.()).rejects.toBe(commitError); + await flushTelegramTestMicrotasks(); + expect(firstSettled).toBe(false); + expect(secondSettled).toBe(false); + + await queuedLifecycle?.onAdmitted?.(); + await expect(Promise.all([firstParticipant.task, secondParticipant.task])).resolves.toEqual([ + { kind: "completed" }, + { kind: "completed" }, + ]); + expect(commitSpy).toHaveBeenCalledTimes(2); + queuedLifecycle?.onComplete?.(); + } finally { + commitSpy.mockRestore(); + setTimeoutSpy.mockRestore(); + } + }); + + it("serializes timeout settlement behind an in-flight durable adoption commit", async () => { + const DEBOUNCE_MS = 4321; + loadConfig.mockReturnValue({ + agents: { + defaults: { + envelopeTimezone: "utc", + }, + }, + messages: { + inbound: { + debounceMs: DEBOUNCE_MS, + }, + }, + channels: { + telegram: { dmPolicy: "open", allowFrom: ["*"] }, + }, + }); + + installPerKeySequentializer(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + let markCommitStarted: (() => void) | undefined; + let releaseCommit: (() => void) | undefined; + const commitStarted = new Promise((resolve) => { + markCommitStarted = resolve; + }); + const commitGate = new Promise((resolve) => { + releaseCommit = resolve; + }); + const commitSpy = vi + .spyOn(messageDispatchDedupe, "commitTelegramMessageDispatchReplay") + .mockImplementationOnce(async () => { + markCommitStarted?.(); + await commitGate; + }); + const releaseSpy = vi.spyOn(messageDispatchDedupe, "releaseTelegramMessageDispatchReplay"); + let queuedLifecycle: GetReplyOptions["queuedFollowupLifecycle"]; + let queuedAbortSignal: AbortSignal | undefined; + let runQueuedTurn: (() => Promise) | undefined; + let modelTurnRan = false; + replySpy.mockImplementationOnce(async (_ctx: MsgContext, opts?: GetReplyOptions) => { + queuedLifecycle = opts?.queuedFollowupLifecycle; + queuedAbortSignal = opts?.abortSignal; + queuedLifecycle?.onEnqueued?.(); + runQueuedTurn = async () => { + await queuedLifecycle?.onAdmitted?.(); + if (queuedAbortSignal?.aborted) { + throw queuedAbortSignal.reason; + } + modelTurnRan = true; + queuedLifecycle?.onComplete?.(); + }; + return undefined; + }); + + try { + createTelegramBot({ token: "tok" }); + const messageHandler = getOnHandler("message") as ( + ctx: TelegramMiddlewareTestContext, + ) => Promise; + const dispatchSpooledMessage = async (updateId: number, text: string) => { + const update = { update_id: updateId }; + return await runWithTelegramSpooledReplayUpdate(update, async () => { + await runTelegramMiddlewareChain({ + ctx: { + update, + message: { + chat: { id: 7, type: "private" }, + text, + date: 1736380800 + updateId, + message_id: updateId, + from: { id: 42, first_name: "Ada" }, + }, + me: { username: "openclaw_bot" }, + getFile: async () => ({}), + }, + finalHandler: messageHandler, + }); + }); + }; + + const firstReplay = await dispatchSpooledMessage(225, "first buffered message"); + const secondReplay = await dispatchSpooledMessage(226, "second buffered message"); + const firstParticipant = requireValue( + firstReplay.deferredWork, + "first buffered replay participant", + ); + const secondParticipant = requireValue( + secondReplay.deferredWork, + "second buffered replay participant", + ); + + const debounceCallIndex = setTimeoutSpy.mock.calls.findLastIndex( + (call) => call[1] === DEBOUNCE_MS, + ); + expect(debounceCallIndex).toBeGreaterThanOrEqual(0); + clearTimeout( + setTimeoutSpy.mock.results[debounceCallIndex]?.value as ReturnType, + ); + const flush = setTimeoutSpy.mock.calls[debounceCallIndex]?.[0] as (() => void) | undefined; + flush?.(); + await vi.waitFor(() => { + expect(runQueuedTurn).toEqual(expect.any(Function)); + }); + + const queuedTurn = runQueuedTurn?.(); + await commitStarted; + const timeoutError = new Error("spooled replay timed out during durable adoption"); + firstParticipant.settle({ kind: "failed-retryable", error: timeoutError }); + await expect(firstParticipant.task).resolves.toEqual({ + kind: "failed-retryable", + error: timeoutError, + }); + await flushTelegramTestMicrotasks(); + expect(releaseSpy).not.toHaveBeenCalled(); + + releaseCommit?.(); + await queuedTurn; + expect(modelTurnRan).toBe(true); + expect(queuedAbortSignal?.aborted).toBe(false); + await expect(secondParticipant.task).resolves.toEqual({ kind: "completed" }); + expect(commitSpy).toHaveBeenCalledTimes(1); + expect(releaseSpy).not.toHaveBeenCalled(); + } finally { + releaseCommit?.(); + commitSpy.mockRestore(); + releaseSpy.mockRestore(); + setTimeoutSpy.mockRestore(); + } + }); + + it("blocks buffered adoption after an exposed replay participant times out", async () => { + const DEBOUNCE_MS = 4321; + loadConfig.mockReturnValue({ + agents: { + defaults: { + envelopeTimezone: "utc", + }, + }, + messages: { + inbound: { + debounceMs: DEBOUNCE_MS, + }, + }, + channels: { + telegram: { dmPolicy: "open", allowFrom: ["*"] }, + }, + }); + + installPerKeySequentializer(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const commitSpy = vi.spyOn(messageDispatchDedupe, "commitTelegramMessageDispatchReplay"); + let queuedLifecycle: GetReplyOptions["queuedFollowupLifecycle"]; + let queuedAbortSignal: AbortSignal | undefined; + replySpy.mockImplementationOnce(async (_ctx: MsgContext, opts?: GetReplyOptions) => { + queuedLifecycle = opts?.queuedFollowupLifecycle; + queuedAbortSignal = opts?.abortSignal; + queuedLifecycle?.onEnqueued?.(); + return undefined; + }); + + try { + createTelegramBot({ token: "tok" }); + const messageHandler = getOnHandler("message") as ( + ctx: TelegramMiddlewareTestContext, + ) => Promise; + const dispatchSpooledMessage = async (updateId: number, text: string) => { + const update = { update_id: updateId }; + return await runWithTelegramSpooledReplayUpdate(update, async () => { + await runTelegramMiddlewareChain({ + ctx: { + update, + message: { + chat: { id: 7, type: "private" }, + text, + date: 1736380800 + updateId, + message_id: updateId, + from: { id: 42, first_name: "Ada" }, + }, + me: { username: "openclaw_bot" }, + getFile: async () => ({}), + }, + finalHandler: messageHandler, + }); + }); + }; + + const firstReplay = await dispatchSpooledMessage(223, "first buffered message"); + const secondReplay = await dispatchSpooledMessage(224, "second buffered message"); + const firstParticipant = requireValue( + firstReplay.deferredWork, + "first buffered replay participant", + ); + const secondParticipant = requireValue( + secondReplay.deferredWork, + "second buffered replay participant", + ); + + const debounceCallIndex = setTimeoutSpy.mock.calls.findLastIndex( + (call) => call[1] === DEBOUNCE_MS, + ); + expect(debounceCallIndex).toBeGreaterThanOrEqual(0); + clearTimeout( + setTimeoutSpy.mock.results[debounceCallIndex]?.value as ReturnType, + ); + const flush = setTimeoutSpy.mock.calls[debounceCallIndex]?.[0] as (() => void) | undefined; + flush?.(); + await vi.waitFor(() => { + expect(queuedLifecycle?.onAdmitted).toEqual(expect.any(Function)); + }); + + const timeoutError = new Error("spooled replay timed out before admission"); + firstParticipant.settle({ kind: "failed-retryable", error: timeoutError }); + await vi.waitFor(() => { + expect(queuedAbortSignal?.aborted).toBe(true); + }); + await expect(secondParticipant.task).resolves.toEqual({ + kind: "failed-retryable", + error: timeoutError, + }); + await expect(queuedLifecycle?.onAdmitted?.()).rejects.toBe(timeoutError); + expect(commitSpy).not.toHaveBeenCalled(); + expect(replySpy).toHaveBeenCalledTimes(1); + } finally { + commitSpy.mockRestore(); + setTimeoutSpy.mockRestore(); + } + }); + it("lets stop cancel pending same-chat forwarded debounce", async () => { const DEBOUNCE_MS = 4321; loadConfig.mockReturnValue({ @@ -2497,6 +2910,63 @@ describe("createTelegramBot", () => { expect(replySpy).toHaveBeenCalledTimes(1); }); + it("retries a spooled message after dispatch fails before turn adoption", async () => { + loadConfig.mockReturnValue({ + channels: { telegram: { dmPolicy: "open", allowFrom: ["*"] } }, + }); + const dispatchError = new Error("failed before turn adoption"); + replySpy.mockRejectedValueOnce(dispatchError).mockResolvedValueOnce({ text: "recovered" }); + + createTelegramBot({ token: "tok" }); + const messageHandler = getOnHandler("message") as ( + ctx: Record, + ) => Promise; + const replayedCtx = () => { + const message = { + chat: { id: 123, type: "private" }, + from: { id: 456, username: "testuser" }, + text: "retry after pre-adoption failure", + date: 1736380800, + message_id: 44, + }; + const update = { update_id: 8488603, message }; + return { + update, + message, + me: { username: "openclaw_bot" }, + getFile: async () => ({ download: async () => new Uint8Array() }), + }; + }; + + const firstCtx = replayedCtx(); + const firstReplay = await runWithTelegramSpooledReplayUpdate(firstCtx.update, async () => { + await runTelegramMiddlewareChain({ + ctx: firstCtx, + finalHandler: messageHandler, + }); + }); + const firstDeferredWork = requireValue(firstReplay.deferredWork, "first replay deferred work"); + await expect(firstDeferredWork.task).resolves.toEqual({ + kind: "failed-retryable", + error: dispatchError, + }); + await flushTelegramTestMicrotasks(); + + const secondCtx = replayedCtx(); + const secondReplay = await runWithTelegramSpooledReplayUpdate(secondCtx.update, async () => { + await runTelegramMiddlewareChain({ + ctx: secondCtx, + finalHandler: messageHandler, + }); + }); + const secondDeferredWork = requireValue( + secondReplay.deferredWork, + "second replay deferred work", + ); + await expect(secondDeferredWork.task).resolves.toEqual({ kind: "completed" }); + expect(replySpy).toHaveBeenCalledTimes(2); + }); + it("persists update offsets after successful dispatch completion", async () => { // For this test we need sequentialize(...) to behave like a normal middleware and call next(). sequentializeSpy.mockImplementationOnce( diff --git a/extensions/telegram/src/bot.test.ts b/extensions/telegram/src/bot.test.ts index 0807257c6ba4..3c820746527b 100644 --- a/extensions/telegram/src/bot.test.ts +++ b/extensions/telegram/src/bot.test.ts @@ -49,6 +49,7 @@ const { wasSentByBot, } = await import("./bot.create-telegram-bot.test-harness.js"); const { recordOutboundMessageForPromptContext } = await import("./outbound-message-context.js"); +const { runWithTelegramSpooledReplayUpdate } = await import("./bot-processing-outcome.js"); let createTelegramBotBase: typeof import("./bot-core.js").createTelegramBotCore; let setTelegramBotRuntimeForTest: typeof import("./bot-core.js").setTelegramBotRuntimeForTest; @@ -4223,7 +4224,7 @@ describe("createTelegramBot", () => { expect(payload.SenderUsername).toBe("ada_bot"); }); - it("retries plugin-owned callback text when the previous reply session is still closing", async () => { + it("settles spooled plugin callback text after a reply-session conflict retry succeeds", async () => { onSpy.mockClear(); replySpy.mockClear(); editMessageReplyMarkupSpy.mockClear(); @@ -4256,22 +4257,30 @@ describe("createTelegramBot", () => { }, }); const callbackHandler = getTelegramCallbackHandlerForTests(); - - await callbackHandler({ - callbackQuery: { - id: "cbq-smart-reply-submit-retry", - data: "openclaw-smart-replies:v1:TWFrZSBBbGljZSBmdW5uaWVy", - from: { id: 9, first_name: "Ada", username: "ada_bot" }, - message: { - chat: { id: 9, type: "private" }, - date: 1736380800, - message_id: 11, - text: "Pick a direction", - }, + const callbackQuery = { + id: "cbq-smart-reply-submit-retry", + data: "openclaw-smart-replies:v1:TWFrZSBBbGljZSBmdW5uaWVy", + from: { id: 9, first_name: "Ada", username: "ada_bot" }, + message: { + chat: { id: 9, type: "private" }, + date: 1736380800, + message_id: 11, + text: "Pick a direction", }, + }; + const update = { update_id: 403, callback_query: callbackQuery }; + const callbackContext = { + update, + callbackQuery, me: { username: "openclaw_bot" }, getFile: async () => ({ download: async () => new Uint8Array() }), + }; + + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => { + await callbackHandler(callbackContext); }); + expect(replay.deferredWork).toBeDefined(); + await expect(replay.deferredWork?.task).resolves.toEqual({ kind: "completed" }); } finally { clearTelegramRuntime(); } diff --git a/extensions/telegram/src/message-dispatch-dedupe.test.ts b/extensions/telegram/src/message-dispatch-dedupe.test.ts index 894d987a1a87..338e0d0c0b52 100644 --- a/extensions/telegram/src/message-dispatch-dedupe.test.ts +++ b/extensions/telegram/src/message-dispatch-dedupe.test.ts @@ -43,6 +43,33 @@ function storedReplayKey(accountId: string, msg: Message): string { return buildTelegramMessageDispatchAccountReplayKey({ accountId, key }); } +function createTestReplayGuard( + params: { + commit?: TelegramMessageDispatchReplayGuard["commit"]; + forget?: TelegramMessageDispatchReplayGuard["forget"]; + release?: TelegramMessageDispatchReplayGuard["release"]; + } = {}, +): TelegramMessageDispatchReplayGuard { + return { + claim: async () => ({ kind: "claimed" }), + commit: params.commit ?? (async () => true), + forget: params.forget ?? (async () => true), + hasRecent: async () => false, + warmup: async () => 0, + clearMemory: () => {}, + memorySize: () => 0, + release: params.release ?? (() => {}), + }; +} + +function createDeferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + beforeEach(() => { previousStateDir = process.env.OPENCLAW_STATE_DIR; process.env.OPENCLAW_STATE_DIR = createStateDir(); @@ -116,6 +143,123 @@ describe("Telegram message dispatch replay guard", () => { ); }); + it("commits replay keys serially before starting the next write", async () => { + const events: string[] = []; + const firstGate = createDeferred(); + const secondGate = createDeferred(); + const secondStarted = createDeferred(); + const guard = createTestReplayGuard({ + commit: async (key) => { + events.push(`start:${key}`); + if (key === "first") { + await firstGate.promise; + } else if (key === "second") { + secondStarted.resolve(); + await secondGate.promise; + } + events.push(`finish:${key}`); + return true; + }, + }); + + const commit = commitTelegramMessageDispatchReplay({ + guard, + keys: ["first", "second", "third"], + }); + + expect(events).toEqual(["start:first"]); + firstGate.resolve(); + await secondStarted.promise; + expect(events).toEqual(["start:first", "finish:first", "start:second"]); + + secondGate.resolve(); + await commit; + expect(events).toEqual([ + "start:first", + "finish:first", + "start:second", + "finish:second", + "start:third", + "finish:third", + ]); + }); + + it("propagates per-key disk errors and stops the commit sequence", async () => { + const diskError = new Error("dedupe disk write failed"); + const commitCalls: string[] = []; + const guard = createTestReplayGuard({ + commit: async (key, options) => { + commitCalls.push(key); + if (key === "second") { + options?.onDiskError?.(diskError); + } + return true; + }, + }); + + await expect( + commitTelegramMessageDispatchReplay({ + guard, + keys: ["first", "second", "third"], + requirePersistent: true, + }), + ).rejects.toBe(diskError); + expect(commitCalls).toEqual(["first", "second"]); + }); + + it("keeps live dispatch commits fail-open on dedupe disk errors", async () => { + const diskError = new Error("dedupe disk write failed"); + const guard = createTestReplayGuard({ + commit: async (_key, options) => { + options?.onDiskError?.(diskError); + return true; + }, + }); + + await expect( + commitTelegramMessageDispatchReplay({ + guard, + keys: ["live-message"], + }), + ).resolves.toBeUndefined(); + }); + + it("rolls back partial multi-key commits after a later disk failure", async () => { + const diskError = new Error("second key was not persisted"); + const committed = new Set(); + const commitCalls: string[] = []; + const forgetCalls: string[] = []; + const releaseCalls: string[] = []; + const guard = createTestReplayGuard({ + commit: async (key, options) => { + commitCalls.push(key); + committed.add(key); + if (key === "second") { + options?.onDiskError?.(diskError); + } + return true; + }, + forget: async (key) => { + forgetCalls.push(key); + committed.delete(key); + return true; + }, + release: (key) => { + releaseCalls.push(key); + }, + }); + const keys = ["first", "second", "third"]; + + await expect( + commitTelegramMessageDispatchReplay({ guard, keys, requirePersistent: true }), + ).rejects.toBe(diskError); + + expect(commitCalls).toEqual(["first", "second"]); + expect(forgetCalls).toEqual(["first", "second"]); + expect(releaseCalls).toEqual(["third"]); + expect([...committed]).toEqual([]); + }); + it("uses one persisted namespace across Telegram accounts", async () => { const writer = createTelegramMessageDispatchReplayGuard(); const first = await claimTelegramMessageDispatchReplay({ diff --git a/extensions/telegram/src/message-dispatch-dedupe.ts b/extensions/telegram/src/message-dispatch-dedupe.ts index 3f341ea5ab78..8e2f9d6aa384 100644 --- a/extensions/telegram/src/message-dispatch-dedupe.ts +++ b/extensions/telegram/src/message-dispatch-dedupe.ts @@ -149,13 +149,76 @@ function normalizeReplayKeys(keys?: readonly string[]): string[] { export async function commitTelegramMessageDispatchReplay(params: { guard: TelegramMessageDispatchReplayGuard; keys?: readonly string[]; + /** Require every claim to reach SQLite before the caller acknowledges durable adoption. */ + requirePersistent?: boolean; }): Promise { const keys = normalizeReplayKeys(params.keys); - await Promise.all( - keys.map((key) => - params.guard.commit(key, { namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE }), - ), - ); + const committedKeys: string[] = []; + // Commit serially so a later failure has no still-running sibling write that + // can race rollback and recreate a key after it was forgotten. + for (const [index, key] of keys.entries()) { + let diskError: unknown; + let recorded = false; + try { + recorded = await params.guard.commit( + key, + params.requirePersistent === true + ? { + namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, + onDiskError: (error) => { + diskError = error; + }, + } + : { namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE }, + ); + if (params.requirePersistent === true && diskError !== undefined) { + throw diskError; + } + } catch (error) { + for (const pendingKey of keys.slice(index + 1)) { + params.guard.release(pendingKey, { + namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, + error, + }); + } + + const failures: TelegramMessageDispatchReplayForgetFailure[] = []; + for (const committedKey of committedKeys) { + try { + const forgotten = await params.guard.forget(committedKey, { + namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, + }); + if (!forgotten) { + failures.push({ key: committedKey }); + } + } catch (rollbackError) { + failures.push({ key: committedKey, error: rollbackError }); + } + } + + let failedKeyCleanupError: unknown; + try { + await params.guard.forget(key, { + namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, + onDiskError: (rollbackError) => { + failedKeyCleanupError = rollbackError; + }, + }); + } catch (rollbackError) { + failedKeyCleanupError = rollbackError; + } + if (failedKeyCleanupError !== undefined) { + failures.push({ key, error: failedKeyCleanupError }); + } + if (failures.length > 0) { + throw new TelegramMessageDispatchReplayForgetError(failures); + } + throw error; + } + if (recorded) { + committedKeys.push(key); + } + } } export async function forgetTelegramMessageDispatchReplay(params: { diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts index 74561fe310c0..12e806e216fe 100644 --- a/extensions/telegram/src/polling-session.test.ts +++ b/extensions/telegram/src/polling-session.test.ts @@ -1963,6 +1963,124 @@ describe("TelegramPollingSession", () => { }); }); + it("keeps a deferred claim owned while adopted completion retries", async () => { + const refreshHarness = installSpooledClaimRefreshHarness(); + await withTempSpool(async (tempDir) => { + let completeAttempts = 0; + let refreshAttempts = 0; + let failRefreshAfterAdoption = false; + let releaseCompletion: (() => void) | undefined; + let markCompletionRetryStarted: (() => void) | undefined; + const completionGate = new Promise((resolve) => { + releaseCompletion = resolve; + }); + const completionRetryStarted = new Promise((resolve) => { + markCompletionRetryStarted = resolve; + }); + setTelegramRuntime({ + state: { + resolveStateDir: () => tempDir, + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => { + const queue = createChannelIngressQueue({ ...options, channelId: "telegram" }); + return { + ...queue, + refreshClaim: async (...args: Parameters>) => { + refreshAttempts += 1; + if (failRefreshAfterAdoption) { + return false; + } + return (await queue.refreshClaim?.(...args)) ?? false; + }, + complete: async (...args: Parameters) => { + completeAttempts += 1; + if (completeAttempts === 1) { + throw new Error("transient completion write failure"); + } + if (completeAttempts === 2) { + markCompletionRetryStarted?.(); + await completionGate; + } + return await queue.complete(...args); + }, + }; + }, + }, + } as TelegramRuntime); + const abort = new AbortController(); + const log = vi.fn(); + const events: number[] = []; + const participants: TelegramSpooledReplayDeferredParticipant[] = []; + const replyFenceKey = "test-completion-retry:topic-10"; + const replyFenceAbortController = new AbortController(); + beginTelegramReplyFence({ + key: replyFenceKey, + laneKey: buildTelegramReplyFenceLaneKey({ + accountId: "default", + sequentialKey: "telegram:-100:topic:10", + }), + supersede: false, + abortController: replyFenceAbortController, + }); + await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "buffered topic 10 turn")]); + + const { runPromise, stopWorker } = startIsolatedIngressSession({ + abort, + spoolDir: tempDir, + log, + handleUpdate: async (update) => { + events.push(Number(update.update_id)); + const participant = createTelegramSpooledReplayDeferredParticipant( + `test-completion-retry:${update.update_id}`, + ); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + participants.push(participant); + }, + }); + + try { + await vi.waitFor(() => expect(participants).toHaveLength(1)); + participants[0]?.settle({ kind: "completed" }); + await completionRetryStarted; + + expect(events).toEqual([42]); + expect( + (await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).map( + (claim) => claim.updateId, + ), + ).toEqual([42]); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); + expectLogIncludes(log, "buffered completion retry 1 scheduled"); + + failRefreshAfterAdoption = true; + refreshHarness.triggerRefresh(); + await vi.waitFor(() => expect(refreshAttempts).toBe(1)); + expect(replyFenceAbortController.signal.aborted).toBe(false); + + releaseCompletion?.(); + await vi.waitFor(async () => + expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]), + ); + await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "telegram refetch")]); + await new Promise((resolve) => { + setTimeout(resolve, 20); + }); + expect(events).toEqual([42]); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); + } finally { + releaseCompletion?.(); + abort.abort(); + stopWorker(); + endTelegramReplyFence(replyFenceKey, replyFenceAbortController); + refreshHarness.restore(); + await runPromise; + } + }); + }); + it("releases buffered spooled claims for retry when deferred processing fails", async () => { await withTempSpool(async (tempDir) => { const abort = new AbortController(); diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts index c756b8e0fbb0..b1e826f8e4c7 100644 --- a/extensions/telegram/src/polling-session.ts +++ b/extensions/telegram/src/polling-session.ts @@ -38,7 +38,7 @@ import { } from "./spooled-update-retry-policy.js"; import { claimNextTelegramSpooledUpdate, - completeTelegramSpooledUpdate, + completeTelegramSpooledUpdateWithRetry, failTelegramSpooledUpdateClaim, isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess, listTelegramSpooledUpdateClaims, @@ -628,6 +628,7 @@ export class TelegramPollingSession { async #handleClaimedSpooledUpdate(params: { bot: TelegramBot; + onTurnAdopted: () => void; stopClaimRefresh: () => void; update: ClaimedTelegramSpooledUpdate; }): Promise { @@ -649,18 +650,26 @@ export class TelegramPollingSession { this.#registerDeferredSpooledUpdate({ deferredWork: replay.deferredWork, laneKey: this.#spooledUpdateLaneKey(params.update), + onTurnAdopted: params.onTurnAdopted, stopClaimRefresh: params.stopClaimRefresh, update: params.update, }); return true; } try { - params.stopClaimRefresh(); - await completeTelegramSpooledUpdate(params.update); + await completeTelegramSpooledUpdateWithRetry({ + update: params.update, + abortSignal: this.opts.abortSignal, + onRetry: ({ attempt, delayMs, error }) => { + this.opts.log( + `[telegram][diag] spooled update ${params.update.updateId} completion retry ${attempt} scheduled in ${formatDurationPrecise(delayMs)}: ${formatErrorMessage(error)}`, + ); + }, + }); return true; } catch (err) { this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} completed but processing marker cleanup failed: ${formatErrorMessage(err)}`, + `[telegram][diag] spooled update ${params.update.updateId} completed but could not tombstone its claimed spool row: ${formatErrorMessage(err)}`, ); return false; } @@ -669,6 +678,7 @@ export class TelegramPollingSession { #registerDeferredSpooledUpdate(params: { deferredWork: TelegramSpooledReplayDeferredParticipant; laneKey: string; + onTurnAdopted: () => void; stopClaimRefresh: () => void; update: ClaimedTelegramSpooledUpdate; }): void { @@ -682,6 +692,13 @@ export class TelegramPollingSession { deferredSpooledUpdateClaimsByKey.delete(claimKey); } let settled = false; + const releaseState = (): void => { + state.stopClaimRefresh(); + if (deferredSpooledUpdateClaimsByKey.get(claimKey) === state) { + deferredSpooledUpdateClaimsByKey.delete(claimKey); + } + this.#deferredSpooledUpdateClaimKeys.delete(claimKey); + }; const finish = async (result: TelegramMessageProcessingResult): Promise => { if (settled) { return; @@ -690,12 +707,13 @@ export class TelegramPollingSession { if (state.timer) { clearTimeout(state.timer); } - state.stopClaimRefresh(); - if (deferredSpooledUpdateClaimsByKey.get(claimKey) === state) { - deferredSpooledUpdateClaimsByKey.delete(claimKey); + if (result.kind === "completed") { + // Claim refresh must continue through tombstone retry, but durable + // adoption transfers cancellation ownership away from ingress. + params.onTurnAdopted(); } - this.#deferredSpooledUpdateClaimKeys.delete(claimKey); if (result.kind === "failed-retryable") { + releaseState(); if (state.timedOutMessage) { await this.#failTimedOutDeferredSpooledUpdate(state); return; @@ -707,11 +725,21 @@ export class TelegramPollingSession { return; } try { - await completeTelegramSpooledUpdate(params.update); + await completeTelegramSpooledUpdateWithRetry({ + update: params.update, + abortSignal: this.opts.abortSignal, + onRetry: ({ attempt, delayMs, error }) => { + this.opts.log( + `[telegram][diag] spooled update ${params.update.updateId} buffered completion retry ${attempt} scheduled in ${formatDurationPrecise(delayMs)}: ${formatErrorMessage(error)}`, + ); + }, + }); } catch (err) { this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} completed after buffered processing but processing marker cleanup failed: ${formatErrorMessage(err)}`, + `[telegram][diag] spooled update ${params.update.updateId} completed after buffered processing but could not tombstone its claimed spool row: ${formatErrorMessage(err)}`, ); + } finally { + releaseState(); } }; const state: DeferredSpooledUpdateClaimState = { @@ -969,10 +997,14 @@ export class TelegramPollingSession { blockedLaneKeys.add(laneKey); continue; } + let abortReplyWorkOnClaimRefreshFailure = true; const stopClaimRefresh = this.#startSpooledUpdateClaimRefresh( claimedUpdate, params.isDrainHealthy, () => { + if (!abortReplyWorkOnClaimRefreshFailure) { + return; + } const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({ accountId: this.opts.accountId, sequentialKey: laneKey, @@ -987,6 +1019,9 @@ export class TelegramPollingSession { ); const handler = this.#handleClaimedSpooledUpdate({ bot: params.bot, + onTurnAdopted: () => { + abortReplyWorkOnClaimRefreshFailure = false; + }, stopClaimRefresh, update: claimedUpdate, }); diff --git a/extensions/telegram/src/spooled-update-retry-policy.ts b/extensions/telegram/src/spooled-update-retry-policy.ts index 6730ebedb91e..1c29c7bb065a 100644 --- a/extensions/telegram/src/spooled-update-retry-policy.ts +++ b/extensions/telegram/src/spooled-update-retry-policy.ts @@ -4,6 +4,8 @@ import { formatErrorMessage, readErrorName, } from "openclaw/plugin-sdk/error-runtime"; +import type { BackoffPolicy } from "openclaw/plugin-sdk/runtime-env"; +import { computeBackoff } from "openclaw/plugin-sdk/runtime-env"; import { isTelegramMessageDispatchReplayForgetError } from "./message-dispatch-dedupe.js"; import type { TelegramSpooledUpdate } from "./telegram-ingress-spool.js"; @@ -11,6 +13,12 @@ export const TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS = 8; export const TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS = 24 * 60 * 60 * 1000; const TELEGRAM_SPOOLED_RETRY_BASE_MS = 1_000; const TELEGRAM_SPOOLED_RETRY_MAX_MS = 3 * 60_000; +const TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY: BackoffPolicy = { + initialMs: 250, + maxMs: 5_000, + factor: 2, + jitter: 0.2, +}; const MISSING_AGENT_HARNESS_ERROR_NAME = "MissingAgentHarnessError"; const MISSING_AGENT_HARNESS_MESSAGE_RE = /Requested agent harness "[^"]+" is not registered\./u; @@ -63,6 +71,10 @@ export function resolveSpooledUpdateAttemptNumber(update: TelegramSpooledUpdate) return (update.attempts ?? 0) + 1; } +export function resolveSpooledUpdatePersistenceRetryDelayMs(attempt: number): number { + return computeBackoff(TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY, attempt); +} + export function shouldDeadLetterRetryableSpooledUpdate( update: TelegramSpooledUpdate, attempt: number, diff --git a/extensions/telegram/src/telegram-ingress-spool.test.ts b/extensions/telegram/src/telegram-ingress-spool.test.ts index bc877ab48294..ffe75b41fe0a 100644 --- a/extensions/telegram/src/telegram-ingress-spool.test.ts +++ b/extensions/telegram/src/telegram-ingress-spool.test.ts @@ -13,6 +13,7 @@ import { claimNextTelegramSpooledUpdate, claimTelegramSpooledUpdate, completeTelegramSpooledUpdate, + completeTelegramSpooledUpdateWithRetry, failTelegramSpooledUpdateClaim, isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess, listTelegramSpooledUpdateClaims, @@ -134,6 +135,47 @@ describe("Telegram ingress spool", () => { }); }); + it("does not tombstone a claim after its token loses ownership", async () => { + await withTempSpool(async (spoolDir) => { + await writeTelegramSpooledUpdate({ + spoolDir, + update: { update_id: 21, message: { text: "claimed" } }, + }); + const pending = (await listTelegramSpooledUpdates({ spoolDir }))[0]; + if (!pending) { + throw new Error("Expected a spooled update"); + } + const firstClaim = await claimTelegramSpooledUpdate(pending); + if (!firstClaim) { + throw new Error("Expected the first claim"); + } + await releaseTelegramSpooledUpdateClaim(firstClaim); + const retryPending = (await listTelegramSpooledUpdates({ spoolDir }))[0]; + if (!retryPending) { + throw new Error("Expected the released update"); + } + const secondClaim = await claimTelegramSpooledUpdate(retryPending); + if (!secondClaim) { + throw new Error("Expected the replacement claim"); + } + + await expect(completeTelegramSpooledUpdateWithRetry({ update: firstClaim })).rejects.toThrow( + "lost claim ownership", + ); + expect( + (await listTelegramSpooledUpdateClaims({ spoolDir })).map((claim) => ({ + updateId: claim.updateId, + claimToken: claim.claim?.claimToken, + })), + ).toEqual([ + { + updateId: 21, + claimToken: secondClaim.claim?.claimToken, + }, + ]); + }); + }); + it("claims next update through the native ingress queue in update id order", async () => { await withTempSpool(async (spoolDir) => { await writeTelegramSpooledUpdate({ diff --git a/extensions/telegram/src/telegram-ingress-spool.ts b/extensions/telegram/src/telegram-ingress-spool.ts index 5f17f272c59c..520c2e0c8050 100644 --- a/extensions/telegram/src/telegram-ingress-spool.ts +++ b/extensions/telegram/src/telegram-ingress-spool.ts @@ -8,10 +8,12 @@ import type { ChannelIngressQueueClaimRef, ChannelIngressQueueRecord, } from "openclaw/plugin-sdk/channel-outbound"; +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { resolveStateDir } from "openclaw/plugin-sdk/state-paths"; import type { TelegramBotInfo } from "./bot-info.js"; import { getTelegramRuntime } from "./runtime.js"; import { getTelegramSequentialKey } from "./sequential-key.js"; +import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./spooled-update-retry-policy.js"; import { normalizeTelegramStateAccountId } from "./state-account-id.js"; const SPOOL_VERSION = 1; @@ -53,6 +55,13 @@ export type ClaimedTelegramSpooledUpdate = TelegramSpooledUpdate & { pendingPath: string; }; +export class TelegramSpooledUpdateCompletionOwnershipError extends Error { + constructor(updateId: number) { + super(`Telegram spooled update ${updateId} lost claim ownership before completion.`); + this.name = "TelegramSpooledUpdateCompletionOwnershipError"; + } +} + function isValidUpdateId(value: unknown): value is number { return typeof value === "number" && Number.isSafeInteger(value) && value >= 0; } @@ -282,11 +291,44 @@ export async function listTelegramSpooledUpdates(params: { ); } -export async function completeTelegramSpooledUpdate(update: TelegramSpooledUpdate): Promise { +export async function completeTelegramSpooledUpdate( + update: TelegramSpooledUpdate, +): Promise { const queue = createTelegramIngressQueue(path.dirname(update.path)); // Successful rows stay as bounded tombstones: Telegram can refetch an update // after dispatch, and callbacks have side effects that plain delete would rerun. - await queue.complete(queueMutationTarget(update)); + return await queue.complete(queueMutationTarget(update)); +} + +export async function completeTelegramSpooledUpdateWithRetry(params: { + update: ClaimedTelegramSpooledUpdate; + abortSignal?: AbortSignal; + onRetry?: (retry: { attempt: number; delayMs: number; error: unknown }) => void; +}): Promise { + if (!params.update.claim?.claimToken) { + throw new TelegramSpooledUpdateCompletionOwnershipError(params.update.updateId); + } + let attempt = 0; + while (true) { + try { + const completed = await completeTelegramSpooledUpdate(params.update); + if (!completed) { + throw new TelegramSpooledUpdateCompletionOwnershipError(params.update.updateId); + } + return; + } catch (err) { + if ( + err instanceof TelegramSpooledUpdateCompletionOwnershipError || + params.abortSignal?.aborted + ) { + throw err; + } + attempt += 1; + const delayMs = resolveSpooledUpdatePersistenceRetryDelayMs(attempt); + params.onRetry?.({ attempt, delayMs, error: err }); + await sleepWithAbort(delayMs, params.abortSignal); + } + } } export async function claimTelegramSpooledUpdate( diff --git a/extensions/telegram/src/webhook.test.ts b/extensions/telegram/src/webhook.test.ts index b03f58cc9e34..38af02712812 100644 --- a/extensions/telegram/src/webhook.test.ts +++ b/extensions/telegram/src/webhook.test.ts @@ -16,6 +16,7 @@ import { clearTelegramRuntime, setTelegramRuntime } from "./runtime.js"; import type { TelegramRuntime } from "./runtime.types.js"; import { TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS } from "./spooled-update-retry-policy.js"; import { + listTelegramSpooledUpdateClaims, listTelegramSpooledUpdates, writeTelegramSpooledUpdate, } from "./telegram-ingress-spool.js"; @@ -950,6 +951,152 @@ describe("startTelegramWebhook", () => { ); }); + it("keeps a webhook lane guarded while claimed completion retries", async () => { + let completeAttempts = 0; + let claimNextCalls = 0; + let releaseCompletion: (() => void) | undefined; + let markCompletionRetryStarted: (() => void) | undefined; + const completionGate = new Promise((resolve) => { + releaseCompletion = resolve; + }); + const completionRetryStarted = new Promise((resolve) => { + markCompletionRetryStarted = resolve; + }); + setTelegramRuntime({ + state: { + resolveStateDir: () => webhookStateDir ?? os.tmpdir(), + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => { + const queue = createChannelIngressQueue({ ...options, channelId: "telegram" }); + return { + ...queue, + claimNext: async (...args: Parameters) => { + claimNextCalls += 1; + return await queue.claimNext(...args); + }, + complete: async (...args: Parameters) => { + completeAttempts += 1; + if (completeAttempts === 1) { + throw new Error("transient completion write failure"); + } + if (completeAttempts === 2) { + markCompletionRetryStarted?.(); + await completionGate; + } + return await queue.complete(...args); + }, + }; + }, + }, + } as TelegramRuntime); + const firstUpdate = { update_id: 50, message: { chat: { id: 123 }, text: "first" } }; + const secondUpdate = { update_id: 51, message: { chat: { id: 123 }, text: "second" } }; + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: firstUpdate, + }); + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: secondUpdate, + }); + const seenUpdateIds: number[] = []; + handleUpdateSpy.mockImplementation(async (update: unknown) => { + seenUpdateIds.push((update as { update_id: number }).update_id); + }); + + const started = await startTelegramWebhook({ + token: TELEGRAM_TOKEN, + port: 0, + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + spoolDir: requireWebhookSpoolDir(), + runtime: { log: vi.fn(), error: vi.fn(), exit: vi.fn() }, + }); + try { + await completionRetryStarted; + const claimNextCallsBeforeLaterDrain = claimNextCalls; + await vi.waitFor( + () => { + expect(claimNextCalls).toBeGreaterThan(claimNextCallsBeforeLaterDrain); + }, + { timeout: 2_000 }, + ); + expect(seenUpdateIds).toEqual([50]); + expect( + (await listTelegramSpooledUpdateClaims({ spoolDir: requireWebhookSpoolDir() })).map( + (claim) => claim.updateId, + ), + ).toEqual([50]); + expect( + (await listTelegramSpooledUpdates({ spoolDir: requireWebhookSpoolDir() })).map( + (update) => update.updateId, + ), + ).toEqual([51]); + + releaseCompletion?.(); + await vi.waitFor(() => expect(seenUpdateIds).toEqual([50, 51])); + await vi.waitFor(async () => + expect( + await listTelegramSpooledUpdateClaims({ spoolDir: requireWebhookSpoolDir() }), + ).toEqual([]), + ); + expect(await listTelegramSpooledUpdates({ spoolDir: requireWebhookSpoolDir() })).toEqual([]); + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: firstUpdate, + }); + expect(await listTelegramSpooledUpdates({ spoolDir: requireWebhookSpoolDir() })).toEqual([]); + expect(seenUpdateIds).toEqual([50, 51]); + } finally { + releaseCompletion?.(); + await started.stop(); + } + }); + + it("stops claimed completion retries when the webhook stops", async () => { + let completeAttempts = 0; + setTelegramRuntime({ + state: { + resolveStateDir: () => webhookStateDir ?? os.tmpdir(), + openChannelIngressQueue: ( + options?: Omit[0], "channelId">, + ) => { + const queue = createChannelIngressQueue({ ...options, channelId: "telegram" }); + return { + ...queue, + complete: async () => { + completeAttempts += 1; + throw new Error("persistent completion write failure"); + }, + }; + }, + }, + } as unknown as TelegramRuntime); + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: { update_id: 52, message: { chat: { id: 123 }, text: "stop retry" } }, + }); + const runtimeLog = vi.fn(); + const started = await startTelegramWebhook({ + token: TELEGRAM_TOKEN, + port: 0, + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + spoolDir: requireWebhookSpoolDir(), + runtime: { log: runtimeLog, error: vi.fn(), exit: vi.fn() }, + }); + + await vi.waitFor(() => + expect(mockMessages(runtimeLog).join("\n")).toContain("completion retry 1 scheduled"), + ); + await started.stop(); + const attemptsAfterStop = completeAttempts; + await sleep(400); + + expect(completeAttempts).toBe(attemptsAfterStop); + }); + it("keeps retry-limit webhook updates pending until they are old enough to dead-letter", async () => { vi.useFakeTimers(); try { diff --git a/extensions/telegram/src/webhook.ts b/extensions/telegram/src/webhook.ts index c5bc6fe7b758..823115ad334e 100644 --- a/extensions/telegram/src/webhook.ts +++ b/extensions/telegram/src/webhook.ts @@ -51,7 +51,7 @@ import { } from "./spooled-update-retry-policy.js"; import { claimNextTelegramSpooledUpdate, - completeTelegramSpooledUpdate, + completeTelegramSpooledUpdateWithRetry, failTelegramSpooledUpdateClaim, isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess, listTelegramSpooledUpdateClaims, @@ -558,6 +558,7 @@ async function waitForWebhookSpooledDeferredWork(params: { async function handleWebhookSpooledUpdate(params: { accountId: string; + abortSignal?: AbortSignal; bot: ReturnType; log: (line: string) => void; update: ClaimedTelegramSpooledUpdate; @@ -642,13 +643,15 @@ async function handleWebhookSpooledUpdate(params: { return {}; } } - try { - await completeTelegramSpooledUpdate(params.update); - } catch (err) { - params.log( - `[telegram][diag] webhook spooled update ${params.update.updateId} completed but processing marker cleanup failed: ${formatErrorMessage(err)}`, - ); - } + await completeTelegramSpooledUpdateWithRetry({ + update: params.update, + abortSignal: params.abortSignal, + onRetry: ({ attempt, delayMs, error }) => { + params.log( + `[telegram][diag] webhook spooled update ${params.update.updateId} completion retry ${attempt} scheduled in ${formatDurationPrecise(delayMs)}: ${formatErrorMessage(error)}`, + ); + }, + }); return {}; } @@ -689,6 +692,10 @@ export async function startTelegramWebhook(opts: { const diagnosticsEnabled = isDiagnosticsEnabled(opts.config); const spoolDir = opts.spoolDir ?? resolveTelegramIngressSpoolDir({ accountId: opts.accountId }); let shutDown = false; + const shutdownAbortController = new AbortController(); + const webhookAbortSignal = opts.abortSignal + ? AbortSignal.any([shutdownAbortController.signal, opts.abortSignal]) + : shutdownAbortController.signal; const telegramAccountConfig = opts.config ? mergeTelegramAccountConfig(opts.config, opts.accountId ?? "default") : undefined; @@ -803,6 +810,7 @@ export async function startTelegramWebhook(opts: { let retainLaneGuardTask: Promise | undefined; void handleWebhookSpooledUpdate({ accountId: opts.accountId ?? "default", + abortSignal: webhookAbortSignal, bot, log, update: claimedUpdate, @@ -977,6 +985,7 @@ export async function startTelegramWebhook(opts: { return; } shutDown = true; + shutdownAbortController.abort(); if (drainTimer) { clearInterval(drainTimer); } diff --git a/src/agents/embedded-agent-runner/runs.test.ts b/src/agents/embedded-agent-runner/runs.test.ts index c0129d54a197..a41afd1afee7 100644 --- a/src/agents/embedded-agent-runner/runs.test.ts +++ b/src/agents/embedded-agent-runner/runs.test.ts @@ -579,9 +579,7 @@ describe("embedded-agent runner run registry", () => { expect(queueMessage).not.toHaveBeenCalled(); }); - it("keeps reply-run fallback reachable for transcript-commit wait requests", async () => { - // Some callers queue through the broader reply-run operation when the - // embedded handle cannot prove transcript commit support directly. + it("rejects transcript-commit waits before reply-run fallback without an active handle", async () => { const queueMessage = vi.fn(async () => {}); const operation = createReplyOperation({ sessionKey: "agent:main:main", @@ -606,22 +604,13 @@ describe("embedded-agent runner run registry", () => { { waitForTranscriptCommit: true, userTurnTranscriptRecorder: recorder }, ); - expect(outcome.queued).toBe(true); - if (!outcome.queued) { - throw new Error("expected reply-run fallback to queue"); - } - expect(outcome).toMatchObject({ - queued: true, + expect(outcome).toEqual({ + queued: false, sessionId: "session-reply-run", - target: "reply_run", + reason: "transcript_commit_wait_unsupported", gatewayHealth: "live", }); - expect(outcome.enqueuedAtMs).toEqual(expect.any(Number)); - expect(outcome.deliveredAtMs).toBeUndefined(); - expect(queueMessage).toHaveBeenCalledWith("completion from child", { - waitForTranscriptCommit: true, - userTurnTranscriptRecorder: recorder, - }); + expect(queueMessage).not.toHaveBeenCalled(); }); it("force-clears an aborted run that does not drain", async () => { diff --git a/src/agents/embedded-agent-runner/runs.ts b/src/agents/embedded-agent-runner/runs.ts index 96b3d4ec06a9..eb84e5968698 100644 --- a/src/agents/embedded-agent-runner/runs.ts +++ b/src/agents/embedded-agent-runner/runs.ts @@ -441,6 +441,15 @@ function prepareEmbeddedAgentQueueMessage( ): PreparedEmbeddedAgentQueueMessage { const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId); if (!handle) { + if (options?.waitForTranscriptCommit === true) { + diag.debug( + `queue message failed: sessionId=${sessionId} reason=transcript_commit_wait_unsupported`, + ); + return { + kind: "complete", + outcome: createQueueFailureOutcome(sessionId, "transcript_commit_wait_unsupported"), + }; + } const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text, options); if (queuedReplyRunMessage) { logMessageQueued({ sessionId, source: "embedded-agent-runner" }); @@ -455,15 +464,6 @@ function prepareEmbeddedAgentQueueMessage( }, }; } - if (options?.waitForTranscriptCommit === true) { - diag.debug( - `queue message failed: sessionId=${sessionId} reason=transcript_commit_wait_unsupported`, - ); - return { - kind: "complete", - outcome: createQueueFailureOutcome(sessionId, "transcript_commit_wait_unsupported"), - }; - } diag.debug(`queue message failed: sessionId=${sessionId} reason=no_active_run`); return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "no_active_run") }; } diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts index b458d231cc0a..ffe7739c6ae1 100644 --- a/src/auto-reply/get-reply-options.types.ts +++ b/src/auto-reply/get-reply-options.types.ts @@ -50,7 +50,7 @@ export type QueuedReplyLifecycle = { /** Retires this source's cancellation ownership while retaining its live identity. */ onCancellationRetired?: () => void; /** Called after the queued turn owns the reply lane, before model/tool execution. */ - onAdmitted?: () => void; + onAdmitted?: () => void | Promise; onComplete?: () => void; }; diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index 982e053ffbc8..e537fcc1bc49 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts @@ -317,6 +317,41 @@ describe("runReplyAgent active steering", () => { }), ); }); + + it("waits for transcript commit and keeps a rejected adoption finalizer irrevocably adopted", async () => { + const finalizerError = new Error("dedupe finalizer failed"); + const events: string[] = []; + state.queueEmbeddedAgentMessageMock.mockImplementationOnce( + (_sessionId: string, _prompt: string, options: unknown) => { + expect(requireRecord(options, "embedded queue options")).toMatchObject({ + steeringMode: "all", + waitForTranscriptCommit: true, + }); + events.push("transcript-committed"); + return true; + }, + ); + const onTurnAdopted = vi.fn(async () => { + events.push("adoption-finalizer"); + throw finalizerError; + }); + const { run, typing } = createMinimalRun({ + opts: { onTurnAdopted }, + isActive: true, + isStreaming: true, + shouldSteer: true, + resolvedQueueMode: "steer", + }); + + await expect(run()).resolves.toBeUndefined(); + + expect(events).toEqual(["transcript-committed", "adoption-finalizer"]); + expect(onTurnAdopted).toHaveBeenCalledTimes(1); + expect(state.queueEmbeddedAgentMessageMock).toHaveBeenCalledTimes(1); + expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled(); + expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled(); + expect(typing.cleanup).toHaveBeenCalledTimes(1); + }); }); describe("runReplyAgent heartbeat followup guard", () => { diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index 75ce403794c5..ecdddb1c301a 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -1269,6 +1269,7 @@ export async function runReplyAgent(params: { followupRun.prompt, { steeringMode: "all", + ...(opts?.onTurnAdopted ? { waitForTranscriptCommit: true } : {}), ...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}), ...(followupRun.userTurnTranscriptRecorder ? { userTurnTranscriptRecorder: followupRun.userTurnTranscriptRecorder } @@ -1276,6 +1277,17 @@ export async function runReplyAgent(params: { }, ); if (steerOutcome.queued) { + try { + await opts?.onTurnAdopted?.(); + } catch (error) { + // Transcript-backed steering is already irrevocably queued here. + // Replaying ingress would duplicate the injected user turn. + logVerbose( + `queue: active session ${steerSessionId} adoption finalizer failed after transcript commit: ${String( + error, + )}`, + ); + } if (followupRun.currentInboundAudio === true) { activeReplyOperation?.markAcceptedSteeredInboundAudio(); } diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts index c3929695d2e2..42bb9800c05a 100644 --- a/src/auto-reply/reply/followup-runner.test.ts +++ b/src/auto-reply/reply/followup-runner.test.ts @@ -387,6 +387,9 @@ async function loadFreshFollowupRunnerModuleForTest() { runCliAgent: (params: unknown) => runCliAgentMock(params), })); vi.doMock("./queue.js", () => ({ + admitFollowupRunLifecycle: async (run: Pick) => { + await run.queuedLifecycle?.onAdmitted?.(); + }, clearFollowupQueue: clearFollowupQueueForFollowupTest, completeFollowupRunLifecycle: (run: Pick) => run.queuedLifecycle?.onComplete?.(), @@ -689,8 +692,12 @@ describe("createFollowupRunner reply-lane admission", () => { expect(context.text).not.toContain("Active goal:"); }); - it("notifies queued owners after admission and before model execution", async () => { + it("awaits queued-owner admission before model execution", async () => { const events: string[] = []; + let releaseAdmission!: () => void; + const admissionBarrier = new Promise((resolve) => { + releaseAdmission = resolve; + }); runEmbeddedAgentMock.mockImplementationOnce(async () => { events.push("run"); return { payloads: [], meta: {} }; @@ -702,17 +709,27 @@ describe("createFollowupRunner reply-lane admission", () => { defaultModel: "anthropic/claude", }); - await runner( + const pending = runner( createQueuedRun({ queuedLifecycle: { - onAdmitted: () => events.push("admitted"), + onAdmitted: async () => { + events.push("admission-started"); + await admissionBarrier; + events.push("admitted"); + }, onComplete: () => events.push("complete"), }, run: { provider: "anthropic", model: "claude" }, }), ); - expect(events).toEqual(["admitted", "run", "complete"]); + await vi.waitFor(() => expect(events).toEqual(["admission-started"])); + expect(runEmbeddedAgentMock).not.toHaveBeenCalled(); + + releaseAdmission(); + await pending; + + expect(events).toEqual(["admission-started", "admitted", "run", "complete"]); }); it("passes prepared media user turns to embedded runtime dispatch", async () => { diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index 4637db8eebc1..a087cc58087a 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/src/auto-reply/reply/followup-runner.ts @@ -99,6 +99,7 @@ import { resolveFollowupDeliveryPayloads } from "./followup-delivery.js"; import { refreshActiveGoalContext } from "./inbound-meta.js"; import { resolveOriginMessageProvider } from "./origin-routing.js"; import { + admitFollowupRunLifecycle, completeFollowupRunLifecycle, FollowupRunDeferredError, isFollowupRunAborted, @@ -682,7 +683,7 @@ export function createFollowupRunner(params: { replyOperation.retainFailureUntilComplete(); // Multi-source collected turns become atomic at reply-lane admission. // Their queue owner uses this boundary to retire source cancellation ids. - effectiveQueued.queuedLifecycle?.onAdmitted?.(); + await admitFollowupRunLifecycle(effectiveQueued); if (replyOperation.sessionId !== run.sessionId) { run = { ...run, sessionId: replyOperation.sessionId }; effectiveQueued = { ...effectiveQueued, run }; diff --git a/src/auto-reply/reply/queue.collect.test.ts b/src/auto-reply/reply/queue.collect.test.ts index 2df0e108a271..5294288d5b03 100644 --- a/src/auto-reply/reply/queue.collect.test.ts +++ b/src/auto-reply/reply/queue.collect.test.ts @@ -6,7 +6,9 @@ import { describe, expect, it, vi } from "vitest"; import { createUserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.js"; import type { FollowupRun, QueueSettings } from "./queue.js"; import { + admitFollowupRunLifecycle, clearFollowupQueue, + completeFollowupRunLifecycle, enqueueFollowupRun, FollowupRunDeferredError, refreshQueuedFollowupSession, @@ -23,6 +25,56 @@ import { getExistingFollowupQueue } from "./queue/state.js"; installQueueRuntimeErrorSilencer(); describe("followup queue collect routing", () => { + it("retries lifecycle admission after a callback rejection", async () => { + const onAdmitted = vi + .fn<() => Promise>() + .mockRejectedValueOnce(new Error("admission failed")) + .mockResolvedValueOnce(); + const run = createRun({ prompt: "retry admission" }); + run.queuedLifecycle = { onAdmitted }; + + await expect(admitFollowupRunLifecycle(run)).rejects.toThrow("admission failed"); + await expect(admitFollowupRunLifecycle(run)).resolves.toBeUndefined(); + await expect(admitFollowupRunLifecycle(run)).resolves.toBeUndefined(); + + expect(onAdmitted).toHaveBeenCalledTimes(2); + }); + + it("serializes completion behind rejected admission and blocks later admission", async () => { + const admissionStarted = createDeferred(); + const releaseAdmission = createDeferred(); + const admissionError = new Error("admission failed"); + const events: string[] = []; + const onAdmitted = vi.fn(async () => { + events.push("admission-started"); + admissionStarted.resolve(); + await releaseAdmission.promise; + events.push("admission-rejected"); + throw admissionError; + }); + const onComplete = vi.fn(() => { + events.push("complete"); + }); + const run = createRun({ prompt: "complete during admission" }); + run.queuedLifecycle = { onAdmitted, onComplete }; + + const admission = admitFollowupRunLifecycle(run); + await admissionStarted.promise; + + completeFollowupRunLifecycle(run); + expect(onComplete).not.toHaveBeenCalled(); + + releaseAdmission.resolve(); + await expect(admission).rejects.toBe(admissionError); + await vi.waitFor(() => expect(onComplete).toHaveBeenCalledTimes(1)); + + await expect(admitFollowupRunLifecycle(run)).rejects.toThrow( + "followup run lifecycle completed before admission", + ); + expect(onAdmitted).toHaveBeenCalledTimes(1); + expect(events).toEqual(["admission-started", "admission-rejected", "complete"]); + }); + it("does not enqueue when the external lifecycle rejects the run identity", () => { const key = `test-rejected-lifecycle-${Date.now()}`; const onEnqueued = vi.fn(() => false); @@ -1329,6 +1381,48 @@ describe("followup queue collect routing", () => { expect(calls[2]?.originatingChatType).toBe("channel"); }); + it("does not deliver a context group again after concurrent overflow summarizes it", async () => { + const key = `test-collect-overflow-stale-context-${Date.now()}`; + const calls: FollowupRun[] = []; + const firstStarted = createDeferred(); + const releaseFirst = createDeferred(); + const settings: QueueSettings = { + mode: "collect", + debounceMs: 0, + cap: 2, + dropPolicy: "summarize", + }; + const createContextRun = (prompt: string, chatType: "direct" | "channel") => + createRun({ + prompt, + originatingChannel: "slack", + originatingTo: "same-target", + originatingChatType: chatType, + }); + + enqueueFollowupRun(key, createContextRun("context A", "direct"), settings); + enqueueFollowupRun(key, createContextRun("context B", "channel"), settings); + + scheduleFollowupDrain(key, async (run) => { + calls.push(run); + if (calls.length === 1) { + firstStarted.resolve(); + await releaseFirst.promise; + } + }); + await firstStarted.promise; + + enqueueFollowupRun(key, createContextRun("context C", "channel"), settings); + enqueueFollowupRun(key, createContextRun("context D", "channel"), settings); + releaseFirst.resolve(); + + await vi.waitFor(() => expect(getExistingFollowupQueue(key)).toBeUndefined()); + const contextBCalls = calls.filter((run) => run.prompt.includes("context B")); + + expect(contextBCalls).toHaveLength(1); + expect(contextBCalls[0]?.prompt).toContain("[Queue overflow] Dropped 1 message due to cap."); + }); + it("retries split overflow summaries after transient failure", async () => { const key = `test-collect-overflow-split-retry-${Date.now()}`; const prompts: string[] = []; @@ -3042,7 +3136,7 @@ describe("followup queue collect routing", () => { calls.push(run); if (calls.length === 1) { expect(run.prompt).toContain("[Queue overflow] Dropped 2 messages due to cap."); - run.queuedLifecycle?.onAdmitted?.(); + await run.queuedLifecycle?.onAdmitted?.(); expect(sourceCancellationRetirements[0]).toHaveBeenCalledTimes(1); expect(sourceCancellationRetirements[1]).not.toHaveBeenCalled(); expect(sourceCompletions[0]).not.toHaveBeenCalled(); @@ -3060,7 +3154,60 @@ describe("followup queue collect routing", () => { expect(calls[1]?.prompt).toBe("live followup"); }); - it("completes summarized room-event lifecycle when overflow summary delivery fails", async () => { + it("admits one lifecycle-owned overflow source before delivery", async () => { + const key = `test-overflow-summary-single-admission-${Date.now()}`; + const events: string[] = []; + const done = createDeferred(); + const sourceComplete = vi.fn(() => { + events.push("source-complete"); + }); + const settings: QueueSettings = { + mode: "followup", + debounceMs: 0, + cap: 1, + dropPolicy: "summarize", + }; + + enqueueFollowupRun( + key, + { + ...createRun({ prompt: "dropped lifecycle source" }), + queuedLifecycle: { + onAdmitted: async () => { + events.push("source-admitted"); + }, + onComplete: sourceComplete, + }, + }, + settings, + ); + enqueueFollowupRun(key, createRun({ prompt: "live followup" }), settings); + + scheduleFollowupDrain(key, async (run) => { + if (run.prompt.includes("[Queue overflow]")) { + events.push("summary-started"); + expect(run.queuedLifecycle?.onAdmitted).toEqual(expect.any(Function)); + await run.queuedLifecycle?.onAdmitted?.(); + events.push("model"); + run.queuedLifecycle?.onComplete?.(); + return; + } + events.push("live-followup"); + done.resolve(); + }); + await done.promise; + + expect(sourceComplete).toHaveBeenCalledTimes(1); + expect(events).toEqual([ + "summary-started", + "source-admitted", + "model", + "source-complete", + "live-followup", + ]); + }); + + it("keeps one onComplete-only overflow source retryable after delivery fails", async () => { const key = `test-overflow-summary-lifecycle-failure-${Date.now()}`; const calls: FollowupRun[] = []; const firstAttempt = createDeferred(); @@ -3070,6 +3217,7 @@ describe("followup queue collect routing", () => { let attempts = 0; const runFollowup = async (run: FollowupRun) => { calls.push(run); + expect(run.queuedLifecycle).toBeUndefined(); attempts += 1; if (attempts === 1) { firstAttempt.resolve(); @@ -3165,6 +3313,66 @@ describe("followup queue collect routing", () => { expect(secondComplete).toHaveBeenCalledTimes(1); }); + it("runs distinct collected admission lifecycles independently when one retries", async () => { + const key = `test-collect-admission-isolation-${Date.now()}`; + const events: string[] = []; + const done = createDeferred(); + const secondAdmissionError = new Error("second admission failed"); + const settings: QueueSettings = { mode: "collect", debounceMs: 0 }; + + const first = createRun({ prompt: "first" }); + first.queuedLifecycle = { + onAdmitted: async () => { + events.push("first-admitted"); + }, + }; + const second = createRun({ prompt: "second" }); + second.queuedLifecycle = { + onAdmitted: vi + .fn<() => Promise>() + .mockImplementationOnce(async () => { + events.push("second-rejected"); + throw secondAdmissionError; + }) + .mockImplementationOnce(async () => { + events.push("second-admitted"); + }), + }; + + enqueueFollowupRun(key, first, settings); + enqueueFollowupRun(key, second, settings); + + scheduleFollowupDrain(key, async (run) => { + const prompt = run.prompt.includes("first") ? "first" : "second"; + events.push(`run:${prompt}`); + try { + await admitFollowupRunLifecycle(run); + } catch (error) { + events.push(`error:${prompt}`); + throw error; + } + events.push(`model:${prompt}`); + if (prompt === "second") { + done.resolve(); + } + }); + + await done.promise; + + expect(events).toEqual([ + "run:first", + "first-admitted", + "model:first", + "run:second", + "second-rejected", + "error:second", + "run:second", + "second-admitted", + "model:second", + ]); + expect(second.queuedLifecycle.onAdmitted).toHaveBeenCalledTimes(2); + }); + it("collects transcript-owned turns under one aggregate recorder", async () => { const key = `test-collect-transcript-owner-${Date.now()}`; const calls: FollowupRun[] = []; @@ -3311,7 +3519,7 @@ describe("followup queue collect routing", () => { if (calls.length === 1) { expect(run.abortSignal).toBeDefined(); expect(run.abortSignal).not.toBe(survivor.signal); - run.queuedLifecycle?.onAdmitted?.(); + await run.queuedLifecycle?.onAdmitted?.(); expect(sourceCancellationRetirements[0]).toHaveBeenCalledTimes(1); expect(sourceCancellationRetirements[1]).not.toHaveBeenCalled(); expect(sourceCompletions[0]).not.toHaveBeenCalled(); @@ -3345,7 +3553,7 @@ describe("followup queue collect routing", () => { scheduleFollowupDrain(key, async (run) => { expect(run.abortSignal).toBeUndefined(); expect(run.queueAbortSignal?.aborted).toBe(false); - run.queuedLifecycle?.onAdmitted?.(); + await run.queuedLifecycle?.onAdmitted?.(); clearFollowupQueue(key); expect(run.queueAbortSignal?.aborted).toBe(true); done.resolve(); @@ -3535,7 +3743,7 @@ describe("followup queue collect routing", () => { if (calls.length === 1) { expect(run.prompt).toContain("Dropped 2 messages"); expect(run.prompt).toContain("retained source"); - run.queuedLifecycle?.onAdmitted?.(); + await run.queuedLifecycle?.onAdmitted?.(); expect(getExistingFollowupQueue(key)?.summaryElisions).toEqual([]); expect(getExistingFollowupQueue(key)?.droppedCount).toBe(0); throw new Error("admitted summary failure"); @@ -3549,6 +3757,74 @@ describe("followup queue collect routing", () => { expect(elidedComplete).toHaveBeenCalledOnce(); expect(retainedComplete).toHaveBeenCalledOnce(); }); + + it("runs distinct overflow admission lifecycles independently when one retries", async () => { + const key = `test-overflow-admission-isolation-${Date.now()}`; + const events: string[] = []; + const done = createDeferred(); + const secondAdmissionError = new Error("second overflow admission failed"); + const settings: QueueSettings = { + mode: "followup", + debounceMs: 0, + cap: 1, + dropPolicy: "summarize", + }; + + const first = createRun({ prompt: "first dropped" }); + first.queuedLifecycle = { + onAdmitted: async () => { + events.push("first-admitted"); + }, + }; + const second = createRun({ prompt: "second dropped" }); + second.queuedLifecycle = { + onAdmitted: vi + .fn<() => Promise>() + .mockImplementationOnce(async () => { + events.push("second-rejected"); + throw secondAdmissionError; + }) + .mockImplementationOnce(async () => { + events.push("second-admitted"); + }), + }; + + enqueueFollowupRun(key, first, settings); + enqueueFollowupRun(key, second, settings); + enqueueFollowupRun(key, createRun({ prompt: "live followup" }), settings); + + scheduleFollowupDrain(key, async (run) => { + if (run.prompt.includes("[Queue overflow]")) { + events.push("summary-run"); + try { + await admitFollowupRunLifecycle(run); + } catch (error) { + events.push("summary-error"); + throw error; + } + events.push("summary-model"); + return; + } + events.push("live-followup"); + done.resolve(); + }); + + await done.promise; + + expect(events).toEqual([ + "summary-run", + "first-admitted", + "summary-model", + "summary-run", + "second-rejected", + "summary-error", + "summary-run", + "second-admitted", + "summary-model", + "live-followup", + ]); + expect(second.queuedLifecycle.onAdmitted).toHaveBeenCalledTimes(2); + }); }); describe("resolveFollowupAuthorizationKey", () => { diff --git a/src/auto-reply/reply/queue.ts b/src/auto-reply/reply/queue.ts index 34e7c883ca5c..55cce8a404f6 100644 --- a/src/auto-reply/reply/queue.ts +++ b/src/auto-reply/reply/queue.ts @@ -18,5 +18,5 @@ export type { QueueSettings, } from "./queue/types.js"; export { isFollowupRunAborted } from "./queue/types.js"; -export { completeFollowupRunLifecycle } from "./queue/types.js"; +export { admitFollowupRunLifecycle, completeFollowupRunLifecycle } from "./queue/types.js"; export { FollowupRunDeferredError, isFollowupRunDeferredError } from "./queue/types.js"; diff --git a/src/auto-reply/reply/queue/drain.ts b/src/auto-reply/reply/queue/drain.ts index dfb0f2bb1e79..b29fff4214c9 100644 --- a/src/auto-reply/reply/queue/drain.ts +++ b/src/auto-reply/reply/queue/drain.ts @@ -14,7 +14,7 @@ import { buildPersistedUserTurnMediaInputsFromFields, createUserTurnTranscriptRecorder, } from "../../../sessions/user-turn-transcript.js"; -import { resolveGlobalMap } from "../../../shared/global-singleton.js"; +import { resolveGlobalMap, resolveGlobalSingleton } from "../../../shared/global-singleton.js"; import { buildCollectPrompt, beginQueueDrain, @@ -28,6 +28,7 @@ import { import { isRoutableChannel } from "../route-reply.js"; import { FOLLOWUP_QUEUES, trimSummaryElisionsToCap } from "./state.js"; import { + admitFollowupRunLifecycle, completeFollowupRunLifecycle, isFollowupRunAborted, isFollowupRunDeferredError, @@ -43,6 +44,39 @@ const FOLLOWUP_RUN_CALLBACKS = resolveGlobalMap Pr FOLLOWUP_DRAIN_CALLBACKS_KEY, ); +const QUEUED_ADMISSION_OWNER_STATE_KEY = Symbol.for("openclaw.queuedAdmissionOwnerState"); +const queuedAdmissionOwnerState = resolveGlobalSingleton(QUEUED_ADMISSION_OWNER_STATE_KEY, () => ({ + keys: new WeakMap, string>(), + nextId: 1, +})); + +function resolveQueuedLifecycleDeliveryKey(lifecycle: FollowupRun["queuedLifecycle"]): string { + if (!lifecycle) { + return ""; + } + const explicitOwnerKey = lifecycle.ownerKey ?? ""; + if (!lifecycle.onAdmitted) { + return explicitOwnerKey; + } + let admissionOwnerKey = queuedAdmissionOwnerState.keys.get(lifecycle); + if (!admissionOwnerKey) { + admissionOwnerKey = `admission:${queuedAdmissionOwnerState.nextId++}`; + queuedAdmissionOwnerState.keys.set(lifecycle, admissionOwnerKey); + } + // Durable admission callbacks own separate ingress identities. Combining + // them would let one source commit before a sibling rejects the aggregate. + return JSON.stringify([explicitOwnerKey, admissionOwnerKey]); +} + +function assertSingleAdmissionOwner(items: readonly FollowupRun[]): void { + const owners = new Set( + items.flatMap((item) => (item.queuedLifecycle?.onAdmitted ? [item.queuedLifecycle] : [])), + ); + if (owners.size > 1) { + throw new Error("followup queue cannot aggregate distinct admission lifecycles"); + } +} + export function rememberFollowupDrainCallback( key: string, runFollowup: (run: FollowupRun) => Promise, @@ -168,6 +202,7 @@ export function resolveFollowupDeliveryContextKey(run: FollowupRun): string { execution.suppressNextUserMessagePersistence === true, execution.suppressTranscriptOnlyAssistantPersistence === true, execution.blockReplyBreak, + resolveQueuedLifecycleDeliveryKey(run.queuedLifecycle), ]); } @@ -630,9 +665,13 @@ function dropAbortedQueueSummarySources(queue: FollowupQueueSummaryState): numbe async function runQueueSummaryDelivery( queue: FollowupQueueSummaryState, delivery: QueueSummaryDelivery, - run: (params: { abortSignal?: AbortSignal; onAdmitted?: () => void }) => Promise, + run: (params: { + abortSignal?: AbortSignal; + onAdmitted?: () => void | Promise; + }) => Promise, protectedSources: FollowupRun[] = delivery.sources, ): Promise { + assertSingleAdmissionOwner(protectedSources); const inheritedActiveSources = new Set( protectedSources.filter((source) => queue.activeSummarySources.has(source)), ); @@ -642,25 +681,28 @@ async function runQueueSummaryDelivery( let admitted = false; let deferredBeforeAdmission = false; const cancellation = createAggregateCancellation(protectedSources); - const onAdmitted = - protectedSources.length > 1 - ? () => { - if (admitted) { - return; - } - cancellation.admit(); - admitted = true; - // A multi-source summary is atomic once it owns the reply lane. - // Retire sibling ids while the latest source owns aggregate cancel. - consumeQueueSummaryDelivery(queue, { ...delivery, sources: protectedSources }, false); - const aggregateOwner = resolveAggregateOwner(protectedSources); - for (const source of protectedSources) { - if (source !== aggregateOwner) { - retireFollowupRunCancellation(source); - } + const needsAdmission = + protectedSources.length > 1 || + protectedSources.some((source) => source.queuedLifecycle?.onAdmitted); + const onAdmitted = needsAdmission + ? async () => { + if (admitted) { + return; + } + await Promise.all(protectedSources.map((source) => admitFollowupRunLifecycle(source))); + cancellation.admit(); + admitted = true; + // A multi-source summary is atomic once it owns the reply lane. + // Retire sibling ids while the latest source owns aggregate cancel. + consumeQueueSummaryDelivery(queue, { ...delivery, sources: protectedSources }, false); + const aggregateOwner = resolveAggregateOwner(protectedSources); + for (const source of protectedSources) { + if (source !== aggregateOwner) { + retireFollowupRunCancellation(source); } } - : undefined; + } + : undefined; try { try { await run({ abortSignal: cancellation.signal, onAdmitted }); @@ -835,7 +877,7 @@ async function runSyntheticOverflowSummary(params: { sources: FollowupRun[]; prompt: string; abortSignal?: AbortSignal; - onAdmitted?: () => void; + onAdmitted?: () => void | Promise; runFollowup: (run: FollowupRun) => Promise; }): Promise { const promptHash = createHash("sha256").update(params.prompt).digest("hex"); @@ -880,9 +922,9 @@ async function runSyntheticOverflowSummary(params: { ...(params.onAdmitted ? { queuedLifecycle: { - onAdmitted: () => { + onAdmitted: async () => { + await params.onAdmitted?.(); admitted = true; - params.onAdmitted?.(); }, onComplete: () => { if (admitted) { @@ -1045,6 +1087,11 @@ export function scheduleFollowupDrain( return; } const effectiveRunFollowup = FOLLOWUP_RUN_CALLBACKS.get(key) ?? runFollowup; + const reserveOptions = { + shouldRestoreOnError: () => + FOLLOWUP_QUEUES.get(key) === queue && !queue.abortController.signal.aborted, + onDiscard: (item: FollowupRun) => completeFollowupRunLifecycle(item), + }; // Cache callback only when a drain actually starts. Avoid keeping stale // callbacks around from finalize calls where no queue work is pending. rememberFollowupDrainCallback(key, effectiveRunFollowup); @@ -1090,6 +1137,7 @@ export function scheduleFollowupDrain( isCrossChannel, items: queue.items, run: effectiveRunFollowup, + reserveOptions, }); if (collectDrainResult === "empty") { break; @@ -1105,17 +1153,23 @@ export function scheduleFollowupDrain( } for (const groupItems of contextGroups) { - const abortedGroupItems = groupItems.filter(isFollowupRunAborted); + // Earlier groups await model work. Recheck membership so overflow + // eviction cannot leave a stale snapshot eligible for delivery. + const currentGroupItems = groupItems.filter((item) => queue.items.includes(item)); + const abortedGroupItems = currentGroupItems.filter(isFollowupRunAborted); if (abortedGroupItems.length > 0) { removeQueuedItemsByRef(queue.items, abortedGroupItems); for (const item of abortedGroupItems) { completeFollowupRunLifecycle(item); } } - const activeGroupItems = groupItems.filter((item) => !isFollowupRunAborted(item)); + const activeGroupItems = currentGroupItems.filter( + (item) => !isFollowupRunAborted(item), + ); if (activeGroupItems.length === 0) { continue; } + assertSingleAdmissionOwner(activeGroupItems); const groupSource = activeGroupItems.at(-1); const run = groupSource?.run ?? queue.lastRun; if (!run) { @@ -1134,6 +1188,14 @@ export function scheduleFollowupDrain( const aggregateOwner = resolveAggregateOwner(activeGroupItems); const cancellation = createAggregateCancellation(activeGroupItems); let admitted = false; + removeQueuedItemsByRef(queue.items, activeGroupItems); + const restoreGroupItems = (items: FollowupRun[]) => { + const missingItems = items.filter((item) => !queue.items.includes(item)); + queue.items.unshift(...missingItems); + }; + const needsGroupAdmission = + activeGroupItems.length > 1 || + activeGroupItems.some((item) => item.queuedLifecycle?.onAdmitted); const consumeAdmittedGroup = () => { cancellation.admit(); admitted = true; @@ -1144,6 +1206,10 @@ export function scheduleFollowupDrain( } } }; + const admitGroupSources = async () => { + await Promise.all(activeGroupItems.map((item) => admitFollowupRunLifecycle(item))); + consumeAdmittedGroup(); + }; const completeGroup = () => { removeQueuedItemsByRef(queue.items, activeGroupItems); for (const item of activeGroupItems) { @@ -1162,10 +1228,10 @@ export function scheduleFollowupDrain( enqueuedAt: Date.now(), ...routing, ...collectRuntimeMetadata(activeGroupItems, cancellation.signal), - ...(activeGroupItems.length > 1 + ...(needsGroupAdmission ? { queuedLifecycle: { - onAdmitted: consumeAdmittedGroup, + onAdmitted: admitGroupSources, onComplete: () => { if (admitted) { completeGroup(); @@ -1182,6 +1248,15 @@ export function scheduleFollowupDrain( } catch (err) { if (admitted) { completeGroup(); + } else if ( + FOLLOWUP_QUEUES.get(key) === queue && + !queue.abortController.signal.aborted + ) { + restoreGroupItems(activeGroupItems); + } else { + for (const item of activeGroupItems) { + completeFollowupRunLifecycle(item); + } } throw err; } finally { @@ -1190,10 +1265,22 @@ export function scheduleFollowupDrain( if (!admitted) { const canceledSources = activeGroupItems.filter(isFollowupRunAborted); if (canceledSources.length > 0) { - removeQueuedItemsByRef(queue.items, canceledSources); for (const item of canceledSources) { completeFollowupRunLifecycle(item); } + const survivors = activeGroupItems.filter( + (item) => !canceledSources.includes(item), + ); + if (FOLLOWUP_QUEUES.get(key) === queue && !queue.abortController.signal.aborted) { + restoreGroupItems(survivors); + if (survivors.length > 0) { + break; + } + } else { + for (const item of survivors) { + completeFollowupRunLifecycle(item); + } + } continue; } } @@ -1202,7 +1289,7 @@ export function scheduleFollowupDrain( continue; } - if (!(await drainNextQueueItem(queue.items, effectiveRunFollowup))) { + if (!(await drainNextQueueItem(queue.items, effectiveRunFollowup, reserveOptions))) { break; } } diff --git a/src/auto-reply/reply/queue/types.ts b/src/auto-reply/reply/queue/types.ts index 51186a64d55d..6c601efefaf1 100644 --- a/src/auto-reply/reply/queue/types.ts +++ b/src/auto-reply/reply/queue/types.ts @@ -171,8 +171,11 @@ export function isFollowupRunAborted( } const enqueuedFollowupLifecycles = new WeakSet(); +const admittedFollowupLifecycles = new WeakSet(); +const admittingFollowupLifecycles = new WeakMap>(); const retiredFollowupCancellationLifecycles = new WeakSet(); const completedFollowupLifecycles = new WeakSet(); +const completedFollowupLifecycleCallbacks = new WeakSet(); export function markFollowupRunEnqueued(run: Pick): boolean { const lifecycle = run.queuedLifecycle; @@ -195,13 +198,55 @@ export function retireFollowupRunCancellation(run: Pick, +): Promise { + const lifecycle = run.queuedLifecycle; + if (!lifecycle || admittedFollowupLifecycles.has(lifecycle)) { + return; + } + const existing = admittingFollowupLifecycles.get(lifecycle); + if (existing) { + await existing; + return; + } + if (completedFollowupLifecycles.has(lifecycle)) { + throw new Error("followup run lifecycle completed before admission"); + } + const admission = Promise.resolve() + .then(async () => await lifecycle.onAdmitted?.()) + .then(() => { + admittedFollowupLifecycles.add(lifecycle); + }); + admittingFollowupLifecycles.set(lifecycle, admission); + try { + await admission; + } finally { + admittingFollowupLifecycles.delete(lifecycle); + } +} + export function completeFollowupRunLifecycle(run: Pick): void { const lifecycle = run.queuedLifecycle; if (!lifecycle || completedFollowupLifecycles.has(lifecycle)) { return; } completedFollowupLifecycles.add(lifecycle); - lifecycle.onComplete?.(); + const finish = () => { + if (completedFollowupLifecycleCallbacks.has(lifecycle)) { + return; + } + completedFollowupLifecycleCallbacks.add(lifecycle); + lifecycle.onComplete?.(); + }; + const admission = admittingFollowupLifecycles.get(lifecycle); + if (!admission) { + finish(); + return; + } + // Completion closes future admission immediately, but the callback waits for + // the in-flight admission attempt so adoption and abandonment cannot race. + void admission.then(finish, finish).catch(() => {}); } export type ResolveQueueSettingsParams = { diff --git a/src/utils/queue-helpers.test.ts b/src/utils/queue-helpers.test.ts index 7f15614c9636..d6345d959659 100644 --- a/src/utils/queue-helpers.test.ts +++ b/src/utils/queue-helpers.test.ts @@ -196,7 +196,7 @@ describe("drainNextQueueItem", () => { ) {} expect(delivered).toEqual(["m1", "m6", "m7", "m8"]); - expect(dropped).toEqual(["m1", "m2", "m3", "m4", "m5"]); + expect(dropped).toEqual(["m2", "m3", "m4", "m5"]); expect(queue.items).toEqual([]); }); }); diff --git a/src/utils/queue-helpers.ts b/src/utils/queue-helpers.ts index 9735f4b270cf..d5808413b118 100644 --- a/src/utils/queue-helpers.ts +++ b/src/utils/queue-helpers.ts @@ -96,6 +96,11 @@ export function shouldSkipQueueItem(params: { return params.dedupe(params.item, params.items); } +type DrainQueueItemOptions = { + shouldRestoreOnError?: (item: T) => boolean; + onDiscard?: (item: T) => void; +}; + /** Apply overflow policy before enqueueing another item. */ export function applyQueueDropPolicy(params: { queue: QueueState; @@ -179,13 +184,22 @@ export function removeQueuedItemsByRef(items: T[], processed: readonly T[]): export async function drainNextQueueItem( items: T[], run: (item: T) => Promise, + options?: DrainQueueItemOptions, ): Promise { - const next = items[0]; + const next = items.shift(); if (!next) { return false; } - await run(next); - removeQueuedItemsByRef(items, [next]); + try { + await run(next); + } catch (error) { + if (options?.shouldRestoreOnError?.(next) ?? true) { + items.unshift(next); + } else { + options?.onDiscard?.(next); + } + throw error; + } return true; } @@ -196,6 +210,7 @@ async function drainCollectItemIfNeeded(params: { setForceIndividualCollect?: (next: boolean) => void; items: T[]; run: (item: T) => Promise; + reserveOptions?: DrainQueueItemOptions; }): Promise<"skipped" | "drained" | "empty"> { if (!params.forceIndividualCollect && !params.isCrossChannel) { return "skipped"; @@ -204,7 +219,7 @@ async function drainCollectItemIfNeeded(params: { // Once cross-channel items appear, future collection stays individual to preserve ordering. params.setForceIndividualCollect?.(true); } - const drained = await drainNextQueueItem(params.items, params.run); + const drained = await drainNextQueueItem(params.items, params.run, params.reserveOptions); return drained ? "drained" : "empty"; } @@ -214,6 +229,7 @@ export async function drainCollectQueueStep(params: { isCrossChannel: boolean; items: T[]; run: (item: T) => Promise; + reserveOptions?: DrainQueueItemOptions; }): Promise<"skipped" | "drained" | "empty"> { return await drainCollectItemIfNeeded({ forceIndividualCollect: params.collectState.forceIndividualCollect, @@ -223,6 +239,7 @@ export async function drainCollectQueueStep(params: { }, items: params.items, run: params.run, + reserveOptions: params.reserveOptions, }); }