/** * Shared queue overflow, debounce, and collection helpers. * * Queue owners use these helpers to cap pending work, summarize dropped items, * debounce drains, and force individual collection when cross-channel ordering matters. */ /** Mutable summary state for a capped queue. */ type QueueSummaryState = { dropPolicy: "summarize" | "old" | "new"; droppedCount: number; summaryLines: string[]; }; /** Queue overflow strategy. */ type QueueDropPolicy = QueueSummaryState["dropPolicy"]; /** Generic capped queue state with shared overflow summary fields. */ type QueueState = QueueSummaryState & { items: T[]; cap: number; }; /** Clear accumulated overflow summary state after it has been emitted. */ export function clearQueueSummaryState(state: QueueSummaryState): void { state.droppedCount = 0; state.summaryLines = []; } /** Build a summary prompt preview without mutating the source queue state. */ export function previewQueueSummaryPrompt(params: { state: QueueSummaryState; noun: string; title?: string; }): string | undefined { return buildQueueSummaryPrompt({ state: { dropPolicy: params.state.dropPolicy, droppedCount: params.state.droppedCount, summaryLines: [...params.state.summaryLines], }, noun: params.noun, title: params.title, }); } /** Apply runtime queue settings while preserving previous values for omitted fields. */ export function applyQueueRuntimeSettings(params: { target: { mode: TMode; debounceMs: number; cap: number; dropPolicy: QueueDropPolicy; }; settings: { mode: TMode; debounceMs?: number; cap?: number; dropPolicy?: QueueDropPolicy; }; }): void { params.target.mode = params.settings.mode; params.target.debounceMs = typeof params.settings.debounceMs === "number" ? Math.max(0, params.settings.debounceMs) : params.target.debounceMs; params.target.cap = typeof params.settings.cap === "number" && params.settings.cap > 0 ? Math.floor(params.settings.cap) : params.target.cap; params.target.dropPolicy = params.settings.dropPolicy ?? params.target.dropPolicy; } /** Trim queue summary text to a bounded single-line preview. */ function elideQueueText(text: string, limit = 140): string { if (text.length <= limit) { return text; } return `${text.slice(0, Math.max(0, limit - 1)).trimEnd()}…`; } /** Normalize whitespace and elide one dropped item for queue summaries. */ function buildQueueSummaryLine(text: string, limit = 160): string { const cleaned = text.replace(/\s+/g, " ").trim(); return elideQueueText(cleaned, limit); } /** Run optional duplicate detection before an item enters a queue. */ export function shouldSkipQueueItem(params: { item: T; items: T[]; dedupe?: (item: T, items: T[]) => boolean; }): boolean { if (!params.dedupe) { return false; } return params.dedupe(params.item, params.items); } /** Apply overflow policy before enqueueing another item. */ export function applyQueueDropPolicy(params: { queue: QueueState; summarize: (item: T) => string; summaryLimit?: number; onDrop?: (items: T[]) => void; }): boolean { const cap = params.queue.cap; if (cap <= 0 || params.queue.items.length < cap) { return true; } if (params.queue.dropPolicy === "new") { return false; } const dropCount = params.queue.items.length - cap + 1; const dropped = params.queue.items.splice(0, dropCount); params.onDrop?.(dropped); if (params.queue.dropPolicy === "summarize") { for (const item of dropped) { params.queue.droppedCount += 1; params.queue.summaryLines.push(buildQueueSummaryLine(params.summarize(item))); } // Summary memory is bounded independently from the item cap to avoid prompt blowups. const limit = Math.max(0, params.summaryLimit ?? cap); while (params.queue.summaryLines.length > limit) { params.queue.summaryLines.shift(); } } return true; } /** Wait until the queue has been quiet for its debounce window. */ export function waitForQueueDebounce(queue: { debounceMs: number; lastEnqueuedAt: number; }): Promise { if (process.env.OPENCLAW_TEST_FAST === "1") { // Tests use this escape hatch so debounce logic does not slow deterministic queue specs. return Promise.resolve(); } const debounceMs = Math.max(0, queue.debounceMs); if (debounceMs <= 0) { return Promise.resolve(); } return new Promise((resolve) => { const check = () => { const since = Date.now() - queue.lastEnqueuedAt; if (since >= debounceMs) { resolve(); return; } setTimeout(check, debounceMs - since); }; check(); }); } /** Mark one queue as draining unless another drain is already active. */ export function beginQueueDrain( map: Map, key: string, ): T | undefined { const queue = map.get(key); if (!queue || queue.draining) { return undefined; } queue.draining = true; return queue; } export function removeQueuedItemsByRef(items: T[], processed: readonly T[]): void { for (const item of processed) { const idx = items.indexOf(item); if (idx !== -1) { items.splice(idx, 1); } } } /** Run and remove the next queued item, returning false when empty. */ export async function drainNextQueueItem( items: T[], run: (item: T) => Promise, ): Promise { const next = items[0]; if (!next) { return false; } await run(next); removeQueuedItemsByRef(items, [next]); return true; } /** Drain one item when collect mode requires individual processing. */ async function drainCollectItemIfNeeded(params: { forceIndividualCollect: boolean; isCrossChannel: boolean; setForceIndividualCollect?: (next: boolean) => void; items: T[]; run: (item: T) => Promise; }): Promise<"skipped" | "drained" | "empty"> { if (!params.forceIndividualCollect && !params.isCrossChannel) { return "skipped"; } if (params.isCrossChannel) { // Once cross-channel items appear, future collection stays individual to preserve ordering. params.setForceIndividualCollect?.(true); } const drained = await drainNextQueueItem(params.items, params.run); return drained ? "drained" : "empty"; } /** Drain one collect step using mutable queue collection state. */ export async function drainCollectQueueStep(params: { collectState: { forceIndividualCollect: boolean }; isCrossChannel: boolean; items: T[]; run: (item: T) => Promise; }): Promise<"skipped" | "drained" | "empty"> { return await drainCollectItemIfNeeded({ forceIndividualCollect: params.collectState.forceIndividualCollect, isCrossChannel: params.isCrossChannel, setForceIndividualCollect: (next) => { params.collectState.forceIndividualCollect = next; }, items: params.items, run: params.run, }); } /** Build and consume the queue overflow summary prompt. */ function buildQueueSummaryPrompt(params: { state: QueueSummaryState; noun: string; title?: string; }): string | undefined { if (params.state.dropPolicy !== "summarize" || params.state.droppedCount <= 0) { return undefined; } const noun = params.noun; const title = params.title ?? `[Queue overflow] Dropped ${params.state.droppedCount} ${noun}${params.state.droppedCount === 1 ? "" : "s"} due to cap.`; const lines = [title]; if (params.state.summaryLines.length > 0) { lines.push("Summary:"); for (const line of params.state.summaryLines) { lines.push(`- ${line}`); } } clearQueueSummaryState(params.state); return lines.join("\n"); } /** Render a collect prompt from queued items and optional overflow summary. */ export function buildCollectPrompt(params: { title: string; items: T[]; summary?: string; renderItem: (item: T, index: number) => string; }): string { const blocks: string[] = [params.title]; if (params.summary) { blocks.push(params.summary); } params.items.forEach((item, idx) => { blocks.push(params.renderItem(item, idx)); }); return blocks.join("\n\n"); } /** Return true when queued items span keys or explicitly mark cross-channel state. */ export function hasCrossChannelItems( items: T[], resolveKey: (item: T) => { key?: string; cross?: boolean }, ): boolean { const keys = new Set(); for (const item of items) { const resolved = resolveKey(item); if (resolved.cross) { return true; } if (!resolved.key) { continue; } keys.add(resolved.key); } return keys.size > 1; }