File size: 3,784 Bytes
fc93158 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | import { createDedupeCache } from "../../../infra/dedupe.js";
import { resolveGlobalSingleton } from "../../../shared/global-singleton.js";
import { applyQueueDropPolicy, shouldSkipQueueItem } from "../../../utils/queue-helpers.js";
import { kickFollowupDrainIfIdle } from "./drain.js";
import { getExistingFollowupQueue, getFollowupQueue } from "./state.js";
import type { FollowupRun, QueueDedupeMode, QueueSettings } from "./types.js";
/**
* Keep queued message-id dedupe shared across bundled chunks so redeliveries
* are rejected no matter which chunk receives the enqueue call.
*/
const RECENT_QUEUE_MESSAGE_IDS_KEY = Symbol.for("openclaw.recentQueueMessageIds");
const RECENT_QUEUE_MESSAGE_IDS = resolveGlobalSingleton(RECENT_QUEUE_MESSAGE_IDS_KEY, () =>
createDedupeCache({
ttlMs: 5 * 60 * 1000,
maxSize: 10_000,
}),
);
function buildRecentMessageIdKey(run: FollowupRun, queueKey: string): string | undefined {
const messageId = run.messageId?.trim();
if (!messageId) {
return undefined;
}
// Use JSON tuple serialization to avoid delimiter-collision edge cases when
// channel/to/account values contain "|" characters.
return JSON.stringify([
"queue",
queueKey,
run.originatingChannel ?? "",
run.originatingTo ?? "",
run.originatingAccountId ?? "",
run.originatingThreadId == null ? "" : String(run.originatingThreadId),
messageId,
]);
}
function isRunAlreadyQueued(
run: FollowupRun,
items: FollowupRun[],
allowPromptFallback = false,
): boolean {
const hasSameRouting = (item: FollowupRun) =>
item.originatingChannel === run.originatingChannel &&
item.originatingTo === run.originatingTo &&
item.originatingAccountId === run.originatingAccountId &&
item.originatingThreadId === run.originatingThreadId;
const messageId = run.messageId?.trim();
if (messageId) {
return items.some((item) => item.messageId?.trim() === messageId && hasSameRouting(item));
}
if (!allowPromptFallback) {
return false;
}
return items.some((item) => item.prompt === run.prompt && hasSameRouting(item));
}
export function enqueueFollowupRun(
key: string,
run: FollowupRun,
settings: QueueSettings,
dedupeMode: QueueDedupeMode = "message-id",
): boolean {
const queue = getFollowupQueue(key, settings);
const recentMessageIdKey = dedupeMode !== "none" ? buildRecentMessageIdKey(run, key) : undefined;
if (recentMessageIdKey && RECENT_QUEUE_MESSAGE_IDS.peek(recentMessageIdKey)) {
return false;
}
const dedupe =
dedupeMode === "none"
? undefined
: (item: FollowupRun, items: FollowupRun[]) =>
isRunAlreadyQueued(item, items, dedupeMode === "prompt");
// Deduplicate: skip if the same message is already queued.
if (shouldSkipQueueItem({ item: run, items: queue.items, dedupe })) {
return false;
}
queue.lastEnqueuedAt = Date.now();
queue.lastRun = run.run;
const shouldEnqueue = applyQueueDropPolicy({
queue,
summarize: (item) => item.summaryLine?.trim() || item.prompt.trim(),
});
if (!shouldEnqueue) {
return false;
}
queue.items.push(run);
if (recentMessageIdKey) {
RECENT_QUEUE_MESSAGE_IDS.check(recentMessageIdKey);
}
// If drain finished and deleted the queue before this item arrived, a new queue
// object was created (draining: false) but nobody scheduled a drain for it.
// Use the cached callback to restart the drain now.
if (!queue.draining) {
kickFollowupDrainIfIdle(key);
}
return true;
}
export function getFollowupQueueDepth(key: string): number {
const queue = getExistingFollowupQueue(key);
if (!queue) {
return 0;
}
return queue.items.length;
}
export function resetRecentQueuedMessageIdDedupe(): void {
RECENT_QUEUE_MESSAGE_IDS.clear();
}
|