fix(telegram): harden spooled turn adoption

This commit is contained in:
Vincent Koc
2026-07-10 04:30:24 -07:00
committed by Vincent Koc
parent 85db758c0d
commit ea88b0a60b
30 changed files with 2435 additions and 192 deletions
+147 -31
View File
@@ -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<TelegramMessageProcessingResult> => {
let dispatchDedupeCommitted = false;
let dispatchDedupeRollbackAttempted = false;
let spooledReplayFinalResult: TelegramMessageProcessingResult | undefined;
let spooledReplayFinalization: Promise<TelegramMessageProcessingResult> | 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<TelegramMessageProcessingResult> => {
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<void> =>
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<typeof processMessageWithReplyChain>[0]["msg"];
storeAllowFrom: Parameters<typeof processMessageWithReplyChain>[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({
@@ -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<
+41 -15
View File
@@ -248,6 +248,12 @@ type DispatchTelegramMessageParams = {
suppressFailureFallback?: boolean;
/** Fires after recovery-relevant session/run state is durably persisted. */
onTurnAdopted?: () => void | Promise<void>;
/** 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<TelegramDispatchResult> => {
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
+379 -6
View File
@@ -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<void>>(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<TelegramMessageProcessingResult> => {
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<TelegramMessageProcessingResult> =>
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<TelegramMessageProcessingResult> => 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<TelegramMessageProcessingResult>>()
.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<TelegramMessageProcessingResult>>()
.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<TelegramMessageProcessingResult> =>
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<TelegramMessageProcessingResult> =>
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<void>((resolve) => {
markDeferred = resolve;
});
let queuedAbortSignal: AbortSignal | undefined;
dispatchTelegramMessage.mockImplementationOnce(
async ({ onTurnAbandoned, onTurnDeferred, turnAbortSignal }) => {
queuedAbortSignal = turnAbortSignal;
onTurnDeferred?.();
markDeferred?.();
if (!turnAbortSignal?.aborted) {
await new Promise<void>((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",
+180 -30
View File
@@ -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> | 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<TelegramMessageProcessingResult>;
completeSpooledReplayAfterIrrevocableAdoption?: (
error: unknown,
) => Promise<TelegramMessageProcessingResult> | 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<void>;
onTurnDeferred?: () => void;
onTurnAbandoned?: () => void;
turnAbortSignal?: AbortSignal;
}): Promise<TelegramMessageProcessingResult> => {
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<TelegramMessageProcessingResult> | undefined;
const settle = async (
result: TelegramMessageProcessingResult,
phase: "adopted" | "terminal",
): Promise<TelegramMessageProcessingResult> => {
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({});
@@ -16,6 +16,7 @@ type TelegramSpooledReplayFrame = {
export type TelegramSpooledReplayDeferredParticipant = {
key: string;
abortSignal: AbortSignal;
task: Promise<TelegramMessageProcessingResult>;
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<TelegramMessageProcessingResult>((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);
},
};
@@ -58,6 +58,7 @@ type BuildModelsProviderDataMock = ReturnType<
typeof vi.fn<NonNullable<typeof telegramBotDepsForTest.buildModelsProviderData>>
>;
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<void>;
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<void>;
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<typeof setTimeout>,
);
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<void>((resolve) => {
markCommitStarted = resolve;
});
const commitGate = new Promise<void>((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<void>) | 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<void>;
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<typeof setTimeout>,
);
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<void>;
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<typeof setTimeout>,
);
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<string, unknown>,
) => Promise<void>;
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(
+22 -13
View File
@@ -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();
}
@@ -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<void>; resolve: () => void } {
let resolve!: () => void;
const promise = new Promise<void>((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<string>();
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({
@@ -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<void> {
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: {
@@ -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<void>((resolve) => {
releaseCompletion = resolve;
});
const completionRetryStarted = new Promise<void>((resolve) => {
markCompletionRetryStarted = resolve;
});
setTelegramRuntime({
state: {
resolveStateDir: () => tempDir,
openChannelIngressQueue: (
options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
) => {
const queue = createChannelIngressQueue({ ...options, channelId: "telegram" });
return {
...queue,
refreshClaim: async (...args: Parameters<NonNullable<typeof queue.refreshClaim>>) => {
refreshAttempts += 1;
if (failRefreshAfterAdoption) {
return false;
}
return (await queue.refreshClaim?.(...args)) ?? false;
},
complete: async (...args: Parameters<typeof queue.complete>) => {
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();
+45 -10
View File
@@ -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<boolean> {
@@ -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<void> => {
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,
});
@@ -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,
@@ -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({
@@ -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<void> {
export async function completeTelegramSpooledUpdate(
update: TelegramSpooledUpdate,
): Promise<boolean> {
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<void> {
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(
+147
View File
@@ -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<void>((resolve) => {
releaseCompletion = resolve;
});
const completionRetryStarted = new Promise<void>((resolve) => {
markCompletionRetryStarted = resolve;
});
setTelegramRuntime({
state: {
resolveStateDir: () => webhookStateDir ?? os.tmpdir(),
openChannelIngressQueue: (
options?: Omit<Parameters<typeof createChannelIngressQueue>[0], "channelId">,
) => {
const queue = createChannelIngressQueue({ ...options, channelId: "telegram" });
return {
...queue,
claimNext: async (...args: Parameters<typeof queue.claimNext>) => {
claimNextCalls += 1;
return await queue.claimNext(...args);
},
complete: async (...args: Parameters<typeof queue.complete>) => {
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<Parameters<typeof createChannelIngressQueue>[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 {
+17 -8
View File
@@ -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<typeof createTelegramBot>;
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<unknown> | 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);
}
+5 -16
View File
@@ -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 () => {
+9 -9
View File
@@ -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") };
}
+1 -1
View File
@@ -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<void>;
onComplete?: () => void;
};
@@ -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", () => {
+12
View File
@@ -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();
}
+21 -4
View File
@@ -387,6 +387,9 @@ async function loadFreshFollowupRunnerModuleForTest() {
runCliAgent: (params: unknown) => runCliAgentMock(params),
}));
vi.doMock("./queue.js", () => ({
admitFollowupRunLifecycle: async (run: Pick<FollowupRun, "queuedLifecycle">) => {
await run.queuedLifecycle?.onAdmitted?.();
},
clearFollowupQueue: clearFollowupQueueForFollowupTest,
completeFollowupRunLifecycle: (run: Pick<FollowupRun, "queuedLifecycle">) =>
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<void>((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 () => {
+2 -1
View File
@@ -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 };
+281 -5
View File
@@ -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<void>>()
.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<void>();
const releaseAdmission = createDeferred<void>();
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<void>();
const releaseFirst = createDeferred<void>();
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<void>();
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<void>();
@@ -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<void>();
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<void>>()
.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<void>();
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<void>>()
.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", () => {
+1 -1
View File
@@ -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";
+115 -28
View File
@@ -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<string, (run: FollowupRun) => 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<NonNullable<FollowupRun["queuedLifecycle"]>, 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<void>,
@@ -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<void>,
run: (params: {
abortSignal?: AbortSignal;
onAdmitted?: () => void | Promise<void>;
}) => Promise<void>,
protectedSources: FollowupRun[] = delivery.sources,
): Promise<boolean> {
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<void>;
runFollowup: (run: FollowupRun) => Promise<void>;
}): Promise<void> {
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;
}
}
+46 -1
View File
@@ -171,8 +171,11 @@ export function isFollowupRunAborted(
}
const enqueuedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
const admittedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
const admittingFollowupLifecycles = new WeakMap<QueuedReplyLifecycle, Promise<void>>();
const retiredFollowupCancellationLifecycles = new WeakSet<QueuedReplyLifecycle>();
const completedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
const completedFollowupLifecycleCallbacks = new WeakSet<QueuedReplyLifecycle>();
export function markFollowupRunEnqueued(run: Pick<FollowupRun, "queuedLifecycle">): boolean {
const lifecycle = run.queuedLifecycle;
@@ -195,13 +198,55 @@ export function retireFollowupRunCancellation(run: Pick<FollowupRun, "queuedLife
lifecycle.onCancellationRetired?.();
}
export async function admitFollowupRunLifecycle(
run: Pick<FollowupRun, "queuedLifecycle">,
): Promise<void> {
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<FollowupRun, "queuedLifecycle">): 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 = {
+1 -1
View File
@@ -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([]);
});
});
+21 -4
View File
@@ -96,6 +96,11 @@ export function shouldSkipQueueItem<T>(params: {
return params.dedupe(params.item, params.items);
}
type DrainQueueItemOptions<T> = {
shouldRestoreOnError?: (item: T) => boolean;
onDiscard?: (item: T) => void;
};
/** Apply overflow policy before enqueueing another item. */
export function applyQueueDropPolicy<T>(params: {
queue: QueueState<T>;
@@ -179,13 +184,22 @@ export function removeQueuedItemsByRef<T>(items: T[], processed: readonly T[]):
export async function drainNextQueueItem<T>(
items: T[],
run: (item: T) => Promise<void>,
options?: DrainQueueItemOptions<T>,
): Promise<boolean> {
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<T>(params: {
setForceIndividualCollect?: (next: boolean) => void;
items: T[];
run: (item: T) => Promise<void>;
reserveOptions?: DrainQueueItemOptions<T>;
}): Promise<"skipped" | "drained" | "empty"> {
if (!params.forceIndividualCollect && !params.isCrossChannel) {
return "skipped";
@@ -204,7 +219,7 @@ async function drainCollectItemIfNeeded<T>(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<T>(params: {
isCrossChannel: boolean;
items: T[];
run: (item: T) => Promise<void>;
reserveOptions?: DrainQueueItemOptions<T>;
}): Promise<"skipped" | "drained" | "empty"> {
return await drainCollectItemIfNeeded({
forceIndividualCollect: params.collectState.forceIndividualCollect,
@@ -223,6 +239,7 @@ export async function drainCollectQueueStep<T>(params: {
},
items: params.items,
run: params.run,
reserveOptions: params.reserveOptions,
});
}