From 71a472bf54daccd9c330183a71608b98514fe977 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 27 Jul 2026 01:26:47 -0400 Subject: [PATCH] refactor: consolidate channel and auto-reply tests (#114330) --- .../reply/get-reply-run.media-only.test.ts | 500 ++---- src/auto-reply/reply/queue.collect.test.ts | 1350 +++++------------ src/channels/message/ingress-drain.test.ts | 154 +- src/channels/message/ingress-queue.test.ts | 205 +-- 4 files changed, 582 insertions(+), 1627 deletions(-) diff --git a/src/auto-reply/reply/get-reply-run.media-only.test.ts b/src/auto-reply/reply/get-reply-run.media-only.test.ts index c76aa9f03bee..805bae6c72c6 100644 --- a/src/auto-reply/reply/get-reply-run.media-only.test.ts +++ b/src/auto-reply/reply/get-reply-run.media-only.test.ts @@ -179,22 +179,48 @@ function createGatewayDrainingError(): Error { const ROOM_EVENT_MESSAGE_TOOL_DIRECTIVE = "Treat this as observed room activity. Default: no reply; most room events need no response from you. Send a visible reply via message(action=send) only when you are directly addressed or have concrete value to add; your final text here stays private either way."; + +function createInboundBody(body: T) { + return { Body: body, RawBody: body, CommandBody: body }; +} + +function createSessionBody(body: T) { + return { Body: body, BodyStripped: body }; +} + +function createProviderSurface(provider: T) { + return { Provider: provider, Surface: provider }; +} + +function createInboundTurn< + TBody extends string, + TProvider extends string, + TChatType extends string, +>(body: TBody, provider: TProvider, chatType: TChatType) { + return { ...createInboundBody(body), ...createProviderSurface(provider), ChatType: chatType }; +} + +function createSessionTurn< + TBody extends string, + TProvider extends string, + TChatType extends string, +>(body: TBody, provider: TProvider, chatType: TChatType) { + return { ...createSessionBody(body), ...createProviderSurface(provider), ChatType: chatType }; +} + function baseParams( overrides: Partial[0]> = {}, ): Parameters[0] { const defaults = { ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier message in this thread", OriginatingChannel: "slack", OriginatingTo: "C123", ChatType: "group", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier message in this thread", media: [{ path: "/tmp/input.png" }], Provider: "slack", @@ -540,17 +566,14 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier direct message", OriginatingChannel: "slack", OriginatingTo: "D123", ChatType: "direct", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier direct message", media: [{ path: "/tmp/input.png" }], Provider: "slack", @@ -570,17 +593,14 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { sourceReplyDeliveryMode: "message_tool_only" }, ctx: { - Body: "yo", - RawBody: "yo", - CommandBody: "yo", + ...createInboundBody("yo"), ThreadHistoryBody: "Earlier direct message", OriginatingChannel: "telegram", OriginatingTo: "telegram-direct-test-id", ChatType: "direct", }, sessionCtx: { - Body: "yo", - BodyStripped: "yo", + ...createSessionBody("yo"), ThreadHistoryBody: "Earlier direct message", media: [{ path: "/tmp/input.png" }], Provider: "telegram", @@ -604,8 +624,7 @@ describe("runPreparedReply media-only handling", () => { expect(directContextParams?.sourceReplyDeliveryMode).toBe("message_tool_only"); expect(buildInboundUserContextPrefix).toHaveBeenCalledWith( { - Body: "yo", - BodyStripped: "yo", + ...createSessionBody("yo"), ThreadHistoryBody: "Earlier direct message", media: [{ path: "/tmp/input.png" }], Provider: "telegram", @@ -679,17 +698,14 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier direct message", OriginatingChannel: "slack", OriginatingTo: "D123", ChatType: chatType, }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier direct message", media: [{ path: "/tmp/input.png" }], Provider: "slack", @@ -715,9 +731,7 @@ describe("runPreparedReply media-only handling", () => { baseParams({ sessionKey: "agent:main:telegram:group:target", ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier direct message", OriginatingChannel: "telegram", OriginatingTo: "D123", @@ -727,8 +741,7 @@ describe("runPreparedReply media-only handling", () => { CommandTargetSessionKey: "agent:main:telegram:group:target", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier direct message", media: [{ path: "/tmp/input.png" }], Provider: "telegram", @@ -888,9 +901,7 @@ describe("runPreparedReply media-only handling", () => { baseParams({ isNewSession: false, ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadStarterBody: "starter message", ThreadHistoryBody: undefined, OriginatingChannel: "slack", @@ -898,8 +909,7 @@ describe("runPreparedReply media-only handling", () => { ChatType: "group", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadStarterBody: "starter message", ThreadHistoryBody: undefined, media: [{ path: "/tmp/input.png" }], @@ -922,9 +932,7 @@ describe("runPreparedReply media-only handling", () => { baseParams({ isNewSession: false, ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadStarterBody: "starter message", ThreadHistoryBody: "Earlier message in this thread", OriginatingChannel: "slack", @@ -932,8 +940,7 @@ describe("runPreparedReply media-only handling", () => { ChatType: "group", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadStarterBody: "starter message", ThreadHistoryBody: "Earlier message in this thread", media: [{ path: "/tmp/input.png" }], @@ -959,17 +966,14 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadStarterBody: "starter message", OriginatingChannel: "slack", OriginatingTo: "C123", ChatType: "group", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadStarterBody: "starter message", media: [{ path: "/tmp/input.png" }], Provider: "slack", @@ -990,13 +994,10 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), Provider: "slack", }, }), @@ -1021,13 +1022,10 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), Provider: "paperclip", OriginatingChannel: "paperclip", OriginatingTo: "paperclip:issue:abc", @@ -1059,15 +1057,12 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ChatType: "group", WasMentioned: true, }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), Provider: "feishu", OriginatingChannel: "feishu", OriginatingTo: "chat-1", @@ -1104,15 +1099,12 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ChatType: "group", WasMentioned: true, }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), Provider: "feishu", OriginatingChannel: "feishu", OriginatingTo: "chat-1", @@ -1142,13 +1134,10 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), Provider: "webchat", OriginatingChannel: "webchat", OriginatingTo: "webchat:local", @@ -1187,17 +1176,14 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "describe this", - RawBody: "describe this", - CommandBody: "describe this", + ...createInboundBody("describe this"), media: [{ path: imagePath, workspaceDir: tmpDir }], OriginatingChannel: "discord", OriginatingTo: "C123", ChatType: "group", }, sessionCtx: { - Body: "describe this", - BodyStripped: "describe this", + ...createSessionBody("describe this"), Provider: "discord", OriginatingChannel: "discord", OriginatingTo: "C123", @@ -1232,16 +1218,13 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "follow up without media", - RawBody: "follow up without media", - CommandBody: "follow up without media", + ...createInboundBody("follow up without media"), OriginatingChannel: "telegram", OriginatingTo: "42", ChatType: "direct", }, sessionCtx: { - Body: "follow up without media", - BodyStripped: "follow up without media", + ...createSessionBody("follow up without media"), Provider: "telegram", OriginatingChannel: "telegram", OriginatingTo: "42", @@ -1268,16 +1251,13 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "hello", - RawBody: "hello", - CommandBody: "hello", + ...createInboundBody("hello"), OriginatingChannel: "telegram", OriginatingTo: "chat-1", ChatType: chatType, }, sessionCtx: { - Body: "hello", - BodyStripped: "hello", + ...createSessionBody("hello"), Provider: "telegram", OriginatingChannel: "telegram", OriginatingTo: "chat-1", @@ -1315,17 +1295,14 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "timestamped followup", - RawBody: "timestamped followup", - CommandBody: "timestamped followup", + ...createInboundBody("timestamped followup"), OriginatingChannel: "whatsapp", OriginatingTo: "+15550001", ChatType: "direct", Timestamp: 1_710_000_000, }, sessionCtx: { - Body: "timestamped followup", - BodyStripped: "timestamped followup", + ...createSessionBody("timestamped followup"), Provider: "whatsapp", OriginatingChannel: "whatsapp", OriginatingTo: "+15550001", @@ -1365,9 +1342,7 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "describe this\n\n[Image]\nDescription:\na tiny dot image", - RawBody: "describe this\n\n[Image]\nDescription:\na tiny dot image", - CommandBody: "describe this\n\n[Image]\nDescription:\na tiny dot image", + ...createInboundBody("describe this\n\n[Image]\nDescription:\na tiny dot image"), media: [ { path: imagePath, contentType: "image/png", workspaceDir: tmpDir }, { path: secondImagePath, contentType: "image/png", workspaceDir: tmpDir }, @@ -1393,8 +1368,7 @@ describe("runPreparedReply media-only handling", () => { ChatType: "direct", }, sessionCtx: { - Body: "describe this\n\n[Image]\nDescription:\na tiny dot image", - BodyStripped: "describe this\n\n[Image]\nDescription:\na tiny dot image", + ...createSessionBody("describe this\n\n[Image]\nDescription:\na tiny dot image"), Provider: "webchat", OriginatingChannel: "webchat", OriginatingTo: "webchat:local", @@ -1440,9 +1414,7 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "describe this\n\n[Image]\nDescription:\na tiny dot image", - RawBody: "describe this\n\n[Image]\nDescription:\na tiny dot image", - CommandBody: "describe this\n\n[Image]\nDescription:\na tiny dot image", + ...createInboundBody("describe this\n\n[Image]\nDescription:\na tiny dot image"), media: [ { path: imagePath, contentType: "image/png", workspaceDir: tmpDir }, { path: secondImagePath, contentType: "image/png", workspaceDir: tmpDir }, @@ -1461,8 +1433,7 @@ describe("runPreparedReply media-only handling", () => { ChatType: "direct", }, sessionCtx: { - Body: "describe this\n\n[Image]\nDescription:\na tiny dot image", - BodyStripped: "describe this\n\n[Image]\nDescription:\na tiny dot image", + ...createSessionBody("describe this\n\n[Image]\nDescription:\na tiny dot image"), Provider: "webchat", OriginatingChannel: "webchat", OriginatingTo: "webchat:local", @@ -1503,9 +1474,7 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "/new", - RawBody: "/new", - CommandBody: "/new", + ...createInboundBody("/new"), }, command: { ...(baseParams().command as Record), @@ -1526,13 +1495,10 @@ describe("runPreparedReply media-only handling", () => { const result = await runPreparedReply( baseParams({ ctx: { - Body: "/reset soft re-read persona files", - RawBody: "/reset soft re-read persona files", - CommandBody: "/reset soft re-read persona files", + ...createInboundBody("/reset soft re-read persona files"), }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), Provider: "slack", }, command: { @@ -1860,22 +1826,13 @@ describe("runPreparedReply media-only handling", () => { baseParams({ isNewSession: false, ctx: { - Body: "second top-level DM", - RawBody: "second top-level DM", - CommandBody: "second top-level DM", - Provider: "slack", - Surface: "slack", - ChatType: "direct", + ...createInboundTurn("second top-level DM", "slack", "direct"), OriginatingChannel: "slack", OriginatingTo: "user:U1", ...threadContext, }, sessionCtx: { - Body: "second top-level DM", - BodyStripped: "second top-level DM", - Provider: "slack", - Surface: "slack", - ChatType: "direct", + ...createSessionTurn("second top-level DM", "slack", "direct"), OriginatingChannel: "slack", OriginatingTo: "user:U1", ...threadContext, @@ -1922,22 +1879,13 @@ describe("runPreparedReply media-only handling", () => { baseParams({ isNewSession: false, ctx: { - Body: "follow-up in another transport thread", - RawBody: "follow-up in another transport thread", - CommandBody: "follow-up in another transport thread", - Provider: "telegram", - Surface: "telegram", - ChatType: "direct", + ...createInboundTurn("follow-up in another transport thread", "telegram", "direct"), OriginatingChannel: "telegram", OriginatingTo: "user:1", MessageThreadId: 43, }, sessionCtx: { - Body: "follow-up in another transport thread", - BodyStripped: "follow-up in another transport thread", - Provider: "telegram", - Surface: "telegram", - ChatType: "direct", + ...createSessionTurn("follow-up in another transport thread", "telegram", "direct"), OriginatingChannel: "telegram", OriginatingTo: "user:1", MessageThreadId: 43, @@ -2299,19 +2247,10 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "what does this mean?", - RawBody: "what does this mean?", - CommandBody: "what does this mean?", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("what does this mean?", "telegram", "group"), }, sessionCtx: { - Body: "what does this mean?", - BodyStripped: "what does this mean?", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("what does this mean?", "telegram", "group"), ReplyToSender: "Jake", ReplyToBody: "quoted status body", ReplyToIsQuote: true, @@ -2351,19 +2290,16 @@ describe("runPreparedReply media-only handling", () => { Body: "", RawBody: "@bot", CommandBody: "@bot", - Provider: "telegram", - Surface: "telegram", + ...createProviderSurface("telegram"), ChatType: "group", ReplyToBody: "quoted status body", ReplyToSender: "Bot", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), RawBody: "@bot", CommandBody: "@bot", - Provider: "telegram", - Surface: "telegram", + ...createProviderSurface("telegram"), ChatType: "group", ReplyToBody: "quoted status body", ReplyToSender: "Bot", @@ -2406,20 +2342,15 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { sourceReplyDeliveryMode: "message_tool_only" }, ctx: { - Body: "No wtf", - RawBody: "No wtf", - CommandBody: "No wtf", - Provider: "telegram", - Surface: "telegram", + ...createInboundBody("No wtf"), + ...createProviderSurface("telegram"), OriginatingChannel: "telegram", OriginatingTo: "-100123", ChatType: "group", }, sessionCtx: { - Body: "No wtf", - BodyStripped: "No wtf", - Provider: "telegram", - Surface: "telegram", + ...createSessionBody("No wtf"), + ...createProviderSurface("telegram"), OriginatingChannel: "telegram", OriginatingTo: "-100123", ChatType: "group", @@ -2514,19 +2445,10 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { abortSignal: abortController.signal }, ctx: { - Body: "ambient", - RawBody: "ambient", - CommandBody: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("ambient", "telegram", "group"), }, sessionCtx: { - Body: "ambient", - BodyStripped: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("ambient", "telegram", "group"), InboundEventKind: "room_event", MessageSid: "992", SenderName: "Alice", @@ -2572,19 +2494,10 @@ describe("runPreparedReply media-only handling", () => { queuedFollowupAbortSignal?: AbortSignal; }, ctx: { - Body: "ambient", - RawBody: "ambient", - CommandBody: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("ambient", "telegram", "group"), }, sessionCtx: { - Body: "ambient", - BodyStripped: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("ambient", "telegram", "group"), InboundEventKind: "room_event", MessageSid: "993", SenderName: "Alice", @@ -2620,19 +2533,10 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { abortSignal: abortController.signal }, ctx: { - Body: "@bot keep this", - RawBody: "@bot keep this", - CommandBody: "@bot keep this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("@bot keep this", "telegram", "group"), }, sessionCtx: { - Body: "@bot keep this", - BodyStripped: "@bot keep this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("@bot keep this", "telegram", "group"), InboundEventKind: "user_request", MessageSid: "994", SenderName: "Alice", @@ -2666,19 +2570,10 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "ambient", - RawBody: "ambient", - CommandBody: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("ambient", "telegram", "group"), }, sessionCtx: { - Body: "ambient", - BodyStripped: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("ambient", "telegram", "group"), InboundEventKind: "room_event", MessageSid: "993", SenderName: "Alice", @@ -2702,19 +2597,10 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { sourceReplyDeliveryMode: "automatic" }, ctx: { - Body: "ambient", - RawBody: "ambient", - CommandBody: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("ambient", "telegram", "group"), }, sessionCtx: { - Body: "ambient", - BodyStripped: "ambient", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("ambient", "telegram", "group"), InboundEventKind: "room_event", MessageSid: "991", SenderName: "Alice", @@ -2735,19 +2621,10 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { sourceReplyDeliveryMode: "automatic" }, ctx: { - Body: "webchat prompt", - RawBody: "webchat prompt", - CommandBody: "webchat prompt", - Provider: "webchat", - Surface: "webchat", - ChatType: "direct", + ...createInboundTurn("webchat prompt", "webchat", "direct"), }, sessionCtx: { - Body: "webchat prompt", - BodyStripped: "webchat prompt", - Provider: "webchat", - Surface: "webchat", - ChatType: "direct", + ...createSessionTurn("webchat prompt", "webchat", "direct"), InboundEventKind: "room_event", MessageSid: "webchat-room-event", SenderName: "Operator", @@ -2769,16 +2646,13 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { sourceReplyDeliveryMode: "automatic" }, ctx: { - Body: "ambient", - RawBody: "ambient", - CommandBody: "ambient", + ...createInboundBody("ambient"), Provider: "webchat", Surface: "telegram", ChatType: "group", }, sessionCtx: { - Body: "ambient", - BodyStripped: "ambient", + ...createSessionBody("ambient"), Provider: "webchat", Surface: "telegram", ChatType: "group", @@ -2802,19 +2676,10 @@ describe("runPreparedReply media-only handling", () => { baseParams({ opts: { sourceReplyDeliveryMode: "message_tool_only" }, ctx: { - Body: "webchat prompt", - RawBody: "webchat prompt", - CommandBody: "webchat prompt", - Provider: "webchat", - Surface: "webchat", - ChatType: "direct", + ...createInboundTurn("webchat prompt", "webchat", "direct"), }, sessionCtx: { - Body: "webchat prompt", - BodyStripped: "webchat prompt", - Provider: "webchat", - Surface: "webchat", - ChatType: "direct", + ...createSessionTurn("webchat prompt", "webchat", "direct"), MessageSid: "webchat-direct", SenderName: "Operator", }, @@ -2840,15 +2705,13 @@ describe("runPreparedReply media-only handling", () => { Body: heartbeatPrompt, RawBody: heartbeatPrompt, CommandBody: heartbeatPrompt, - Provider: "heartbeat", - Surface: "heartbeat", + ...createProviderSurface("heartbeat"), ChatType: "direct", }, sessionCtx: { Body: heartbeatPrompt, BodyStripped: heartbeatPrompt, - Provider: "heartbeat", - Surface: "heartbeat", + ...createProviderSurface("heartbeat"), ChatType: "direct", }, }), @@ -2913,15 +2776,12 @@ describe("runPreparedReply media-only handling", () => { isNewSession: false, systemSent: true, ctx: { - Body: "scheduled wake", - RawBody: "scheduled wake", - CommandBody: "scheduled wake", + ...createInboundBody("scheduled wake"), Provider: "cron-event", SessionKey: "agent:main:discord:guild-1:channel-1", }, sessionCtx: { - Body: "scheduled wake", - BodyStripped: "scheduled wake", + ...createSessionBody("scheduled wake"), Provider: "cron-event", }, sessionEntry: { @@ -3014,20 +2874,11 @@ describe("runPreparedReply media-only handling", () => { systemSent: true, sessionEntry, ctx: { - Body: "@bot check this", - RawBody: "@bot check this", - CommandBody: "@bot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("@bot check this", "telegram", "group"), MessageSid: "msg-1", }, sessionCtx: { - Body: "@bot check this", - BodyStripped: "@bot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("@bot check this", "telegram", "group"), InboundEventKind: "room_event", MessageSid: "msg-1", }, @@ -3043,20 +2894,11 @@ describe("runPreparedReply media-only handling", () => { systemSent: true, sessionEntry, ctx: { - Body: "@bot check this", - RawBody: "@bot check this", - CommandBody: "@bot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("@bot check this", "telegram", "group"), MessageSid: "msg-2", }, sessionCtx: { - Body: "@bot check this", - BodyStripped: "@bot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("@bot check this", "telegram", "group"), MessageSid: "msg-2", }, }), @@ -3072,15 +2914,12 @@ describe("runPreparedReply media-only handling", () => { systemSent: true, sessionEntry, ctx: { - Body: "scheduled wake", - RawBody: "scheduled wake", - CommandBody: "scheduled wake", + ...createInboundBody("scheduled wake"), Provider: "cron-event", SessionKey: "agent:main:telegram:-100123", }, sessionCtx: { - Body: "scheduled wake", - BodyStripped: "scheduled wake", + ...createSessionBody("scheduled wake"), Provider: "cron-event", }, }), @@ -3125,19 +2964,10 @@ describe("runPreparedReply media-only handling", () => { isNewSession: false, systemSent: true, ctx: { - Body: "@bot check this", - RawBody: "@bot check this", - CommandBody: "@bot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("@bot check this", "telegram", "group"), }, sessionCtx: { - Body: "@bot check this", - BodyStripped: "@bot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("@bot check this", "telegram", "group"), InboundEventKind: "room_event" as const, }, }; @@ -3178,21 +3008,12 @@ describe("runPreparedReply media-only handling", () => { isNewSession: false, systemSent: true, ctx: { - Body: "@SirPinchALotBot check this", - RawBody: "@SirPinchALotBot check this", - CommandBody: "@SirPinchALotBot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("@SirPinchALotBot check this", "telegram", "group"), BotUsername: "SirPinchALotBot", ExplicitlyMentionedBot: true, }, sessionCtx: { - Body: "@SirPinchALotBot check this", - BodyStripped: "@SirPinchALotBot check this", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("@SirPinchALotBot check this", "telegram", "group"), BotUsername: "SirPinchALotBot", ExplicitlyMentionedBot: true, }, @@ -3242,19 +3063,10 @@ describe("runPreparedReply media-only handling", () => { systemSent: false, sessionEntry, ctx: { - Body: "@bot first", - RawBody: "@bot first", - CommandBody: "@bot first", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("@bot first", "telegram", "group"), }, sessionCtx: { - Body: "@bot first", - BodyStripped: "@bot first", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("@bot first", "telegram", "group"), }, }), ); @@ -3268,19 +3080,10 @@ describe("runPreparedReply media-only handling", () => { systemSent: true, sessionEntry, ctx: { - Body: "second", - RawBody: "second", - CommandBody: "second", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createInboundTurn("second", "telegram", "group"), }, sessionCtx: { - Body: "second", - BodyStripped: "second", - Provider: "telegram", - Surface: "telegram", - ChatType: "group", + ...createSessionTurn("second", "telegram", "group"), }, }), ); @@ -3310,18 +3113,13 @@ describe("runPreparedReply media-only handling", () => { Body: commandText, RawBody: commandText, CommandBody: commandText, - Provider: "webchat", - Surface: "webchat", + ...createProviderSurface("webchat"), ChatType: "direct", ReplyToBody: "quoted reset target", ReplyToSender: "Ada Lovelace", }, sessionCtx: { - Body: "", - BodyStripped: "", - Provider: "webchat", - Surface: "webchat", - ChatType: "direct", + ...createSessionTurn("", "webchat", "direct"), SenderId: "telegram-user-1", SenderName: "Ada Lovelace", ReplyToBody: "quoted reset target", @@ -3357,19 +3155,10 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "/reset summarize my workspace", - RawBody: "/reset summarize my workspace", - CommandBody: "/reset summarize my workspace", - Provider: "webchat", - Surface: "webchat", - ChatType: "direct", + ...createInboundTurn("/reset summarize my workspace", "webchat", "direct"), }, sessionCtx: { - Body: "", - BodyStripped: "", - Provider: "webchat", - Surface: "webchat", - ChatType: "direct", + ...createSessionTurn("", "webchat", "direct"), }, command: { surface: "webchat", @@ -3397,17 +3186,14 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier message in this thread", OriginatingChannel: "webchat", OriginatingTo: "session:abc", ChatType: "group", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier message in this thread", media: [{ path: "/tmp/input.png" }], Provider: "telegram", @@ -3434,9 +3220,7 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier message in this thread", OriginatingChannel: undefined, OriginatingTo: undefined, @@ -3445,8 +3229,7 @@ describe("runPreparedReply media-only handling", () => { ChatType: "group", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier message in this thread", media: [{ path: "/tmp/input.png" }], Provider: "webchat", @@ -3465,9 +3248,7 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier message in this thread", OriginatingChannel: "discord", OriginatingTo: "channel:24680", @@ -3475,8 +3256,7 @@ describe("runPreparedReply media-only handling", () => { AccountId: undefined, }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier message in this thread", media: [{ path: "/tmp/input.png" }], Provider: "discord", @@ -3503,9 +3283,7 @@ describe("runPreparedReply media-only handling", () => { agents: { defaults: {} }, }, ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier message in this thread", Provider: "slack", OriginatingChannel: undefined, @@ -3514,8 +3292,7 @@ describe("runPreparedReply media-only handling", () => { ReplyToMode: "off", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier message in this thread", media: [{ path: "/tmp/input.png" }], Provider: "slack", @@ -3548,17 +3325,14 @@ describe("runPreparedReply media-only handling", () => { }, opts: { isHeartbeat: true }, ctx: { - Body: "scheduled wake", - RawBody: "scheduled wake", - CommandBody: "scheduled wake", + ...createInboundBody("scheduled wake"), Provider: "cron-event", SessionKey: "agent:main:slack:direct:U1", OriginatingChannel: "slack", OriginatingTo: "user:U1", }, sessionCtx: { - Body: "scheduled wake", - BodyStripped: "scheduled wake", + ...createSessionBody("scheduled wake"), Provider: "cron-event", OriginatingChannel: "slack", OriginatingTo: "user:U1", @@ -3600,9 +3374,7 @@ describe("runPreparedReply media-only handling", () => { await runPreparedReply( baseParams({ ctx: { - Body: "", - RawBody: "", - CommandBody: "", + ...createInboundBody(""), ThreadHistoryBody: "Earlier message in this thread", OriginatingChannel: "slack", OriginatingTo: "user:U1", @@ -3611,8 +3383,7 @@ describe("runPreparedReply media-only handling", () => { TransportThreadId: "650.000", }, sessionCtx: { - Body: "", - BodyStripped: "", + ...createSessionBody(""), ThreadHistoryBody: "Earlier message in this thread", media: [{ path: "/tmp/input.png" }], Provider: "slack", @@ -3760,8 +3531,7 @@ describe("runPreparedReply media-only handling", () => { baseParams({ ctx: { Body: "low steer this conversation", RawBody: "low steer this conversation" }, sessionCtx: { - Body: "low steer this conversation", - BodyStripped: "low steer this conversation", + ...createSessionBody("low steer this conversation"), }, resolvedThinkLevel: undefined, }), diff --git a/src/auto-reply/reply/queue.collect.test.ts b/src/auto-reply/reply/queue.collect.test.ts index ee68723ddfbf..43be7fbb8b01 100644 --- a/src/auto-reply/reply/queue.collect.test.ts +++ b/src/auto-reply/reply/queue.collect.test.ts @@ -29,6 +29,36 @@ import { clearFollowupQueue, getExistingFollowupQueue } from "./queue/state.js"; installQueueRuntimeErrorSilencer(); +function createQueueSettings(overrides: Partial = {}): QueueSettings { + return { + mode: "collect", + debounceMs: 0, + cap: 50, + dropPolicy: "summarize", + ...overrides, + }; +} + +function enqueueTestRun( + key: string, + params: Parameters[0], + settings: QueueSettings, +) { + return enqueueFollowupRun(key, createRun(params), settings); +} + +function createDrainRecorder(expectedCalls = 1) { + const calls: FollowupRun[] = []; + const done = createDeferred(); + const runFollowup = async (run: FollowupRun) => { + calls.push(run); + if (calls.length >= expectedCalls) { + done.resolve(); + } + }; + return { calls, done, runFollowup }; +} + describe("followup queue collect routing", () => { it("marks exclusive admission without onAbandoned and isolates collect identity", () => { // Failure window: cancel-only used to be inferred from missing onAbandoned, @@ -145,38 +175,25 @@ describe("followup queue collect routing", () => { it("does not collect when destinations differ", async () => { const key = `test-collect-diff-to-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 2; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "one", originatingChannel: "slack", originatingTo: "channel:A", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "two", originatingChannel: "slack", originatingTo: "channel:B", - }), + }, settings, ); @@ -188,37 +205,27 @@ describe("followup queue collect routing", () => { it("collects when channel+destination match", async () => { const key = `test-collect-same-to-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "one", originatingChannel: "slack", originatingTo: "channel:A", originatingChatType: "channel", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "two", originatingChannel: "slack", originatingTo: "channel:A", originatingChatType: "channel", - }), + }, settings, ); @@ -232,22 +239,16 @@ describe("followup queue collect routing", () => { it("collects Slack top-level messages when reply anchors are disabled", async () => { const key = `test-collect-slack-reply-off-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); for (const [prompt, replyToId] of [ ["one", "101.001"], ["two", "101.002"], ] as const) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, messageId: replyToId, originatingChannel: "slack", @@ -255,15 +256,12 @@ describe("followup queue collect routing", () => { originatingReplyToId: replyToId, originatingReplyToMode: "off", originatingChatType: "channel", - }), + }, settings, ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - done.resolve(); - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(1); @@ -273,22 +271,16 @@ describe("followup queue collect routing", () => { it("splits collect batches when enabled reply anchors differ", async () => { const key = `test-collect-slack-reply-all-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); for (const [prompt, replyToId] of [ ["one", "101.001"], ["two", "101.002"], ] as const) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, messageId: replyToId, originatingChannel: "slack", @@ -296,17 +288,12 @@ describe("followup queue collect routing", () => { originatingReplyToId: replyToId, originatingReplyToMode: "all", originatingChatType: "channel", - }), + }, settings, ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length === 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls.map((call) => call.prompt)).toEqual(["one", "two"]); @@ -320,29 +307,23 @@ describe("followup queue collect routing", () => { "splits standalone Slack collect batches by message id in %s reply mode", async (replyToMode, originatingChannel) => { const key = `test-collect-slack-standalone-${replyToMode}-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done } = createDrainRecorder(); + const settings = createQueueSettings(); for (const [prompt, messageId] of [ ["one", "101.001"], ["two", "101.002"], ] as const) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, messageId, originatingChannel, originatingTo: "channel:A", originatingReplyToMode: replyToMode, originatingChatType: "channel", - }), + }, settings, ); } @@ -362,22 +343,16 @@ describe("followup queue collect routing", () => { it("collects distinct messages inside the same routed thread", async () => { const key = `test-collect-shared-thread-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); for (const [prompt, messageId] of [ ["one", "message-1"], ["two", "message-2"], ] as const) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, messageId, originatingChannel: "telegram", @@ -385,15 +360,12 @@ describe("followup queue collect routing", () => { originatingThreadId: "topic-1", originatingReplyToMode: "all", originatingChatType: "group", - }), + }, settings, ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - done.resolve(); - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(1); @@ -403,22 +375,16 @@ describe("followup queue collect routing", () => { it("does not collect when captured reply modes differ on the same anchor", async () => { const key = `test-collect-slack-reply-mode-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); for (const [prompt, messageId, replyToMode] of [ ["first", "message-1", "first"], ["all", "message-2", "all"], ] as const) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, messageId, originatingChannel: "slack", @@ -426,17 +392,12 @@ describe("followup queue collect routing", () => { originatingReplyToId: "101.001", originatingReplyToMode: replyToMode, originatingChatType: "channel", - }), + }, settings, ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length === 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls.map((call) => call.prompt)).toEqual(["first", "all"]); @@ -445,39 +406,27 @@ describe("followup queue collect routing", () => { it("does not collect when chat types differ on the same destination", async () => { const key = `test-collect-diff-chat-type-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "direct", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "direct", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "channel", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "channel", - }), + }, settings, ); @@ -490,14 +439,8 @@ describe("followup queue collect routing", () => { it("does not collect when source delivery policy differs", async () => { const key = `test-collect-diff-delivery-policy-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); const createPolicyRun = ( prompt: string, sourceReplyDeliveryMode: NonNullable, @@ -519,12 +462,7 @@ describe("followup queue collect routing", () => { enqueueFollowupRun(key, createPolicyRun("automatic", "automatic"), settings); enqueueFollowupRun(key, createPolicyRun("private", "message_tool_only"), settings); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls.map((call) => call.prompt)).toEqual([ @@ -539,14 +477,8 @@ describe("followup queue collect routing", () => { it("does not collect when task suggestion delivery differs", async () => { const key = `test-collect-diff-task-suggestion-delivery-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); const createTaskRun = (prompt: string, taskSuggestionDeliveryMode?: "gateway") => { const base = createRun({ prompt, @@ -565,12 +497,7 @@ describe("followup queue collect routing", () => { enqueueFollowupRun(key, createTaskRun("legacy client"), settings); enqueueFollowupRun(key, createTaskRun("actionable client", "gateway"), settings); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls.map((call) => call.run.taskSuggestionDeliveryMode)).toEqual([ @@ -581,39 +508,27 @@ describe("followup queue collect routing", () => { it("keeps overflow summaries on the dropped source chat type", async () => { const key = `test-collect-overflow-chat-type-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "private direct content", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "direct", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "public channel content", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "channel", - }), + }, settings, ); @@ -629,42 +544,31 @@ describe("followup queue collect routing", () => { it("keeps overflow summaries on the dropped source route", async () => { const key = `test-collect-overflow-route-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "channel A content", originatingChannel: "slack", originatingTo: "channel:A", originatingChatType: "channel", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "channel B content", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls[0]?.prompt).toContain("- channel A content"); @@ -676,52 +580,41 @@ describe("followup queue collect routing", () => { it("does not attribute elided private drops to a public summary", async () => { const key = `test-collect-overflow-elided-context-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "private direct content", originatingChannel: "slack", originatingTo: "direct:A", originatingChatType: "direct", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "older public content", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "newer public content", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 3) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls[0]?.prompt).toContain("[Queue overflow] Dropped 1 message due to cap."); @@ -737,22 +630,17 @@ describe("followup queue collect routing", () => { it("evicts oldest overflow context metadata when the item cap is reached", () => { const key = `test-collect-overflow-elision-bound-${Date.now()}`; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 2 }); const accepted = ["A", "B", "A", "B", "A", "B", "survivor"].map((target, index) => - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: `message ${index}`, originatingChannel: "slack", originatingTo: `channel:${target}`, originatingChatType: "channel", - }), + }, settings, ), ); @@ -775,12 +663,7 @@ describe("followup queue collect routing", () => { it("bounds retained overflow cancellation identities by the item cap", () => { const key = `test-collect-overflow-source-bound-${Date.now()}`; const completions = Array.from({ length: 8 }, () => vi.fn()); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 2 }); for (const [index, onComplete] of completions.entries()) { enqueueFollowupRun( @@ -814,12 +697,7 @@ describe("followup queue collect routing", () => { const key = `test-drop-new-lifecycle-${Date.now()}`; const onEnqueued = vi.fn(); const onComplete = vi.fn(); - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "new", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1, dropPolicy: "new" }); expect(enqueueFollowupRun(key, createRun({ prompt: "existing" }), settings)).toBe(true); expect( @@ -845,14 +723,8 @@ describe("followup queue collect routing", () => { it("keeps retained excess contexts isolated after evicting the oldest metadata", async () => { const key = `test-collect-overflow-evicted-context-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ cap: 1 }); const accepted = ["A", "B", "C", "D"].map((target) => enqueueFollowupRun( @@ -868,12 +740,7 @@ describe("followup queue collect routing", () => { ); expect(accepted).toEqual([true, true, true, true]); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 3) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(3); @@ -891,14 +758,8 @@ describe("followup queue collect routing", () => { it("keeps overflow summaries under the dropped sender authorization", async () => { const key = `test-collect-overflow-auth-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ cap: 1 }); const dropped = createRun({ prompt: "guest content", originatingChannel: "slack", @@ -937,12 +798,7 @@ describe("followup queue collect routing", () => { settings, ); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls[0]?.prompt).toContain("- guest content"); @@ -956,14 +812,8 @@ describe("followup queue collect routing", () => { it("uses the head item authorization for non-collect overflow delivery", async () => { const key = `test-followup-overflow-auth-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ mode: "followup", cap: 2 }); const guestRun = (prompt: string) => { const base = createRun({ prompt, @@ -1002,12 +852,7 @@ describe("followup queue collect routing", () => { settings, ); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 3) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(3); @@ -1024,14 +869,8 @@ describe("followup queue collect routing", () => { it("batches compatible overflow sources into one summary run", async () => { const key = `test-collect-overflow-group-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 3, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ cap: 3 }); for (const [prompt, model] of [ ["direct A", "model-a"], @@ -1057,24 +896,19 @@ describe("followup queue collect routing", () => { ); } for (const prompt of ["channel D", "channel E", "channel F"]) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "channel", - }), + }, settings, ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(2); @@ -1097,23 +931,17 @@ describe("followup queue collect routing", () => { }); it("scopes overflow transcript idempotency to the source route", async () => { - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 1 }); const drainRoute = async (to: string): Promise => { const key = `test-collect-overflow-route-key-${to}-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); for (const [prompt, messageId] of [ ["dropped", "provider-local-id"], ["survivor", "survivor-id"], ] as const) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, messageId, originatingChannel: "slack", @@ -1123,7 +951,7 @@ describe("followup queue collect routing", () => { originatingReplyToId: "reply", originatingReplyToMode: "all", originatingChatType: "channel", - }), + }, settings, ); } @@ -1154,14 +982,8 @@ describe("followup queue collect routing", () => { it("uses the newest run for a fully elided overflow segment", async () => { const key = `test-collect-overflow-elided-latest-run-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ cap: 1 }); for (const [prompt, model, authProfileId, chatType] of [ ["first", "model-a", "auth-a", "direct"], @@ -1189,12 +1011,7 @@ describe("followup queue collect routing", () => { ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 3) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls[0]?.prompt).toContain("Dropped 1 message"); @@ -1206,14 +1023,8 @@ describe("followup queue collect routing", () => { it("splits overflow groups when source delivery policy changes", async () => { const key = `test-collect-overflow-delivery-policy-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ cap: 2 }); const createSource = ( prompt: string, sourceReplyDeliveryMode: NonNullable, @@ -1236,24 +1047,19 @@ describe("followup queue collect routing", () => { enqueueFollowupRun(key, createSource("automatic source", "automatic"), settings); enqueueFollowupRun(key, createSource("private source", "message_tool_only"), settings); for (const prompt of ["survivor one", "survivor two"]) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 3) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(3); @@ -1267,14 +1073,8 @@ describe("followup queue collect routing", () => { it("splits overflow groups when runtime policy identity changes", async () => { const key = `test-collect-overflow-runtime-policy-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ cap: 2 }); const createSource = (prompt: string, runtimePolicySessionKey: string) => { const base = createRun({ prompt, @@ -1294,24 +1094,19 @@ describe("followup queue collect routing", () => { enqueueFollowupRun(key, createSource("policy one", "policy:one"), settings); enqueueFollowupRun(key, createSource("policy two", "policy:two"), settings); for (const prompt of ["survivor one", "survivor two"]) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt, originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 3) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(3); @@ -1325,43 +1120,32 @@ describe("followup queue collect routing", () => { it("preserves the source message id for standalone overflow summaries", async () => { const key = `test-collect-overflow-message-id-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "dropped source", messageId: "message-42", originatingChannel: "slack", originatingTo: "channel:A", originatingChatType: "channel", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "survivor", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls[0]?.prompt).toContain("- dropped source"); @@ -1376,39 +1160,27 @@ describe("followup queue collect routing", () => { "separates overflow when the %s chat type is missing", async (_missingSide, droppedChatType, survivingChatType) => { const key = `test-collect-overflow-missing-chat-${_missingSide}-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "dropped content", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: droppedChatType, - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "surviving content", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: survivingChatType, - }), + }, settings, ); @@ -1424,8 +1196,7 @@ describe("followup queue collect routing", () => { it("drops an aborted split summary before running the surviving item", async () => { const key = `test-collect-overflow-current-run-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done, runFollowup } = createDrainRecorder(); const controller = new AbortController(); const droppedBase = createRun({ prompt: "private direct content", @@ -1439,12 +1210,7 @@ describe("followup queue collect routing", () => { originatingTo: "same-target", originatingChatType: "channel", }); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 1 }); enqueueFollowupRun( key, @@ -1480,10 +1246,7 @@ describe("followup queue collect routing", () => { nextModel: "current-model", }); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - done.resolve(); - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(1); @@ -1499,31 +1262,26 @@ describe("followup queue collect routing", () => { const firstStarted = createDeferred(); const releaseFirst = createDeferred(); const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "source A", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "direct", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "source B", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "channel", - }), + }, settings, ); @@ -1540,14 +1298,14 @@ describe("followup queue collect routing", () => { }); await firstStarted.promise; - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "surviving C", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "channel", - }), + }, settings, ); releaseFirst.resolve(); @@ -1569,12 +1327,7 @@ describe("followup queue collect routing", () => { const calls: FollowupRun[] = []; const firstStarted = createDeferred(); const releaseFirst = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 2 }); const createContextRun = (prompt: string, chatType: "direct" | "channel") => createRun({ prompt, @@ -1612,12 +1365,7 @@ describe("followup queue collect routing", () => { const done = createDeferred(); const onComplete = vi.fn(); let attempt = 0; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 1 }); enqueueFollowupRun( key, @@ -1632,14 +1380,14 @@ describe("followup queue collect routing", () => { }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "public survivor", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "channel", - }), + }, settings, ); @@ -1664,47 +1412,41 @@ describe("followup queue collect routing", () => { it("keeps deferred overflow summary text paired with its source route", async () => { const key = `test-collect-overflow-deferred-pairs-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done } = createDrainRecorder(); + const settings = createQueueSettings({ cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "source A", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "direct", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "source B", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "direct", - }), + }, settings, ); scheduleFollowupDrain(key, async (run) => { calls.push(run); if (calls.length === 1) { - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "surviving C", originatingChannel: "slack", originatingTo: "same-target", originatingChatType: "channel", - }), + }, settings, ); throw new FollowupRunDeferredError(); @@ -1726,46 +1468,34 @@ describe("followup queue collect routing", () => { it("collects compatible items after one cross-channel drain", async () => { const key = `test-collect-after-cross-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "first route", originatingChannel: "slack", originatingTo: "channel:A", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "second route one", originatingChannel: "slack", originatingTo: "channel:B", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "second route two", originatingChannel: "slack", originatingTo: "channel:B", - }), + }, settings, ); @@ -1783,40 +1513,28 @@ describe("followup queue collect routing", () => { it("drains unresolved-origin items separately from a routed batch", async () => { const key = `test-collect-unresolved-origin-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); enqueueFollowupRun(key, createRun({ prompt: "unresolved origin" }), settings); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "keyed one", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "keyed two", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); @@ -1837,50 +1555,39 @@ describe("followup queue collect routing", () => { it("does not collect known route-less chat types into another destination", async () => { const key = `test-collect-known-chat-without-route-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "unresolved direct", originatingChatType: "direct", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "channel one", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "channel two", originatingChannel: "slack", originatingTo: "channel:B", originatingChatType: "channel", - }), + }, settings, ); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls[0]?.prompt).toBe("unresolved direct"); @@ -1893,37 +1600,27 @@ describe("followup queue collect routing", () => { it("collects ordinary user-request followups with current turn kind", async () => { const key = `test-collect-user-request-kind-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "one", currentInboundEventKind: "user_request", originatingChannel: "slack", originatingTo: "channel:A", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "two", currentInboundEventKind: "user_request", originatingChannel: "slack", originatingTo: "channel:A", - }), + }, settings, ); @@ -1938,32 +1635,19 @@ describe("followup queue collect routing", () => { it("drains runtime-context followups individually instead of collecting them", async () => { const key = `test-collect-runtime-context-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 2; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); const controller = new AbortController(); const begin = () => () => undefined; const lifecycle = { onAdopted: async () => {}, onSettled: () => undefined }; - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "[OpenClaw room event]", originatingChannel: "telegram", originatingTo: "-100123", - }), + }, settings, ); const first = getExistingFollowupQueue(key)?.items[0]; @@ -1976,13 +1660,13 @@ describe("followup queue collect routing", () => { first.abortSignal = controller.signal; first.deliveryCorrelations = [{ begin }]; first.turnAdoptionLifecycle = lifecycle; - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "second", originatingChannel: "telegram", originatingTo: "-100123", - }), + }, settings, ); @@ -2003,21 +1687,8 @@ describe("followup queue collect routing", () => { it("drains a disableCollectBatching retry individually instead of collecting it", async () => { const strandedReplyRetryMarker = "stranded-reply-retry"; const key = `test-collect-disable-batching-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 3; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings(); const route = { originatingChannel: "slack" as const, originatingTo: "channel:A" }; const retryPrompt = "[System] Please deliver this reply now by calling message(action=send)."; @@ -2053,12 +1724,7 @@ describe("followup queue collect routing", () => { it("can prepend priority followups before already queued items", () => { const key = `test-priority-followup-front-${Date.now()}`; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup" }); enqueueFollowupRun(key, createRun({ prompt: "queued later one" }), settings); enqueueFollowupRun(key, createRun({ prompt: "queued later two" }), settings); @@ -2082,12 +1748,7 @@ describe("followup queue collect routing", () => { it("preserves prepended priority followups during old-item overflow eviction", () => { const key = `test-priority-followup-overflow-${Date.now()}`; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 2, - dropPolicy: "old", - }; + const settings = createQueueSettings({ mode: "followup", cap: 2, dropPolicy: "old" }); enqueueFollowupRun(key, createRun({ prompt: "queued later one" }), settings); enqueueFollowupRun(key, createRun({ prompt: "queued later two" }), settings); @@ -2110,12 +1771,7 @@ describe("followup queue collect routing", () => { it("keeps a cap-one protected priority followup instead of evicting it", () => { const key = `test-priority-followup-cap-one-${Date.now()}`; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); const priorityAccepted = enqueueFollowupRun( key, @@ -2142,12 +1798,12 @@ describe("followup queue collect routing", () => { it("does not advance debounce stamp when overflow rejects an incoming message", () => { const key = `test-priority-followup-debounce-reject-${Date.now()}`; - const settings: QueueSettings = { + const settings = createQueueSettings({ mode: "followup", debounceMs: 5_000, cap: 1, dropPolicy: "old", - }; + }); const priorityAccepted = enqueueFollowupRun( key, @@ -2215,20 +1871,8 @@ describe("followup queue collect routing", () => { it("drains protected priority followups before overflow summaries", async () => { const key = `test-priority-followup-before-summary-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun(key, createRun({ prompt: "overflowed normal" }), settings); enqueueFollowupRun( @@ -2252,18 +1896,8 @@ describe("followup queue collect routing", () => { it("carries image payloads across collected batches", async () => { const key = `test-collect-images-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); const firstImage = { type: "image" as const, data: "first", mimeType: "image/png" }; const secondImage = { type: "image" as const, data: "second", mimeType: "image/png" }; @@ -2303,21 +1937,8 @@ describe("followup queue collect routing", () => { it("splits collect batches when sender authorization changes", async () => { const key = `test-collect-auth-split-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 2; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); const nonOwner = createRun({ prompt: "use the gateway tool", @@ -2368,8 +1989,7 @@ describe("followup queue collect routing", () => { it("splits collect batches when queued cancellation owners differ", async () => { const key = `test-collect-cancel-owner-split-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done, runFollowup } = createDrainRecorder(2); const settings: QueueSettings = { mode: "collect", debounceMs: 0 }; for (const [prompt, ownerKey] of [ @@ -2390,12 +2010,7 @@ describe("followup queue collect routing", () => { ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - if (calls.length === 2) { - done.resolve(); - } - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(2); @@ -2407,18 +2022,8 @@ describe("followup queue collect routing", () => { it("keeps one collect batch when authorization context matches", async () => { const key = `test-collect-auth-match-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); const first = createRun({ prompt: "first", @@ -2472,18 +2077,8 @@ describe("followup queue collect routing", () => { it("keeps one collect batch when only sender display fields drift", async () => { const key = `test-collect-auth-display-drift-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); const first = createRun({ prompt: "first", @@ -2537,21 +2132,8 @@ describe("followup queue collect routing", () => { it("splits collect batches when exec context changes", async () => { const key = `test-collect-exec-split-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 2; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); const base = createRun({ prompt: "first", @@ -2603,18 +2185,8 @@ describe("followup queue collect routing", () => { it("uses the newest run within a matching authorization batch", async () => { const key = `test-collect-latest-run-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); const first = createRun({ prompt: "first", originatingChannel: "slack", originatingTo: "A" }); const second = createRun({ @@ -2664,47 +2236,34 @@ describe("followup queue collect routing", () => { it("delivers summary-only collect work under its source route", async () => { const key = `test-collect-summary-only-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 3; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ cap: 2 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "first", originatingChannel: "slack", originatingTo: "channel:A", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "second", originatingChannel: "slack", originatingTo: "channel:B", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "third", originatingChannel: "slack", originatingTo: "channel:C", - }), + }, settings, ); @@ -2721,21 +2280,8 @@ describe("followup queue collect routing", () => { it("preserves collect order when authorization changes more than once", async () => { const key = `test-collect-auth-order-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 3; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings(); const first = createRun({ prompt: "first", originatingChannel: "slack", originatingTo: "A" }); const second = createRun({ prompt: "second", originatingChannel: "slack", originatingTo: "A" }); @@ -2778,37 +2324,27 @@ describe("followup queue collect routing", () => { it("collects Slack messages in same thread and preserves string thread id", async () => { const key = `test-collect-slack-thread-same-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "one", originatingChannel: "slack", originatingTo: "channel:A", originatingThreadId: "1706000000.000001", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "two", originatingChannel: "slack", originatingTo: "channel:A", originatingThreadId: "1706000000.000001", - }), + }, settings, ); @@ -2820,37 +2356,27 @@ describe("followup queue collect routing", () => { it("collects messages when numeric and string thread ids share the route key", async () => { const key = `test-collect-thread-normalized-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "one", originatingChannel: "telegram", originatingTo: "-100123", originatingThreadId: 42.9, - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "two", originatingChannel: "telegram", originatingTo: "-100123", originatingThreadId: "42", - }), + }, settings, ); @@ -2864,34 +2390,30 @@ describe("followup queue collect routing", () => { it("collects matching local webchat routes with distinct message ids", async () => { const key = `test-collect-local-webchat-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done, runFollowup } = createDrainRecorder(); const settings: QueueSettings = { mode: "collect", debounceMs: 0 }; - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "one", messageId: "webchat-message-1", originatingChannel: "webchat", originatingReplyToMode: "all", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "two", messageId: "webchat-message-2", originatingChannel: "webchat", originatingReplyToMode: "all", - }), + }, settings, ); - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - done.resolve(); - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(1); @@ -2901,40 +2423,27 @@ describe("followup queue collect routing", () => { it("does not collect Slack messages when thread ids differ", async () => { const key = `test-collect-slack-thread-diff-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 2; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings(); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "one", originatingChannel: "slack", originatingTo: "channel:A", originatingThreadId: "1706000000.000001", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "two", originatingChannel: "slack", originatingTo: "channel:A", originatingThreadId: "1706000000.000002", - }), + }, settings, ); @@ -2948,8 +2457,7 @@ describe("followup queue collect routing", () => { it("retries collect-mode batches without losing queued items", async () => { const key = `test-collect-retry-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); let attempt = 0; const runFollowup = async (run: FollowupRun) => { attempt += 1; @@ -2959,12 +2467,7 @@ describe("followup queue collect routing", () => { calls.push(run); done.resolve(); }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const settings = createQueueSettings(); enqueueFollowupRun(key, createRun({ prompt: "one" }), settings); enqueueFollowupRun(key, createRun({ prompt: "two" }), settings); @@ -2992,12 +2495,7 @@ describe("followup queue collect routing", () => { done.resolve(); } }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 50, - dropPolicy: "summarize", - }; + const settings = createQueueSettings(); const guest = createRun({ prompt: "guest message", @@ -3054,8 +2552,7 @@ describe("followup queue collect routing", () => { it("retries overflow summary delivery without losing dropped previews", async () => { const key = `test-overflow-summary-retry-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); let attempt = 0; const runFollowup = async (run: FollowupRun) => { attempt += 1; @@ -3065,12 +2562,7 @@ describe("followup queue collect routing", () => { calls.push(run); done.resolve(); }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun(key, createRun({ prompt: "first" }), settings); enqueueFollowupRun(key, createRun({ prompt: "second" }), settings); @@ -3086,14 +2578,8 @@ describe("followup queue collect routing", () => { const storePath = path.join(tempDir, "sessions.json"); const oldTranscriptPath = path.join(tempDir, "old-session.jsonl"); const key = `test-overflow-summary-session-rotation-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done } = createDrainRecorder(); + const settings = createQueueSettings({ mode: "followup", cap: 1 }); try { await replaceSessionEntry( @@ -3155,12 +2641,7 @@ describe("followup queue collect routing", () => { const key = `test-overflow-summary-aborted-${Date.now()}`; const calls: FollowupRun[] = []; const cleaned: FollowupRun[] = []; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); const controller = new AbortController(); const onComplete = vi.fn(); @@ -3196,21 +2677,8 @@ describe("followup queue collect routing", () => { it("delivers the overflow summary before split auth groups", async () => { const key = `test-collect-overflow-summary-once-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const expectedCalls = 3; - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= expectedCalls) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(3); + const settings = createQueueSettings({ cap: 2 }); const droppedGuest = createRun({ prompt: "dropped guest message", @@ -3282,8 +2750,7 @@ describe("followup queue collect routing", () => { it("does not re-deliver overflow summary on partial auth group failure retry", async () => { const key = `test-collect-overflow-partial-retry-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); let attempt = 0; const runFollowup = async (run: FollowupRun) => { attempt += 1; @@ -3297,12 +2764,7 @@ describe("followup queue collect routing", () => { done.resolve(); } }; - const settings: QueueSettings = { - mode: "collect", - debounceMs: 0, - cap: 2, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ cap: 2 }); const droppedGuest = createRun({ prompt: "dropped guest message", @@ -3375,39 +2837,29 @@ describe("followup queue collect routing", () => { it("preserves routing metadata on overflow summary followups", async () => { const key = `test-overflow-summary-routing-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - done.resolve(); - }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(); + const settings = createQueueSettings({ mode: "followup", cap: 1 }); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "first", originatingChannel: "discord", originatingTo: "channel:C1", originatingAccountId: "work", originatingThreadId: "1739142736.000100", - }), + }, settings, ); - enqueueFollowupRun( + enqueueTestRun( key, - createRun({ + { prompt: "second", originatingChannel: "discord", originatingTo: "channel:C1", originatingAccountId: "work", originatingThreadId: "1739142736.000100", - }), + }, settings, ); @@ -3423,8 +2875,7 @@ describe("followup queue collect routing", () => { it("keeps live item runtime metadata out of standalone overflow summaries", async () => { const key = `test-overflow-summary-runtime-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const controller = new AbortController(); const onComplete = vi.fn(); const begin = vi.fn(() => () => undefined); @@ -3434,12 +2885,7 @@ describe("followup queue collect routing", () => { done.resolve(); } }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -3483,20 +2929,8 @@ describe("followup queue collect routing", () => { it("keeps mixed overflow summaries as normal followups", async () => { const key = `test-overflow-summary-mixed-kind-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); - const runFollowup = async (run: FollowupRun) => { - calls.push(run); - if (calls.length >= 2) { - done.resolve(); - } - }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const { calls, done, runFollowup } = createDrainRecorder(2); + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -3527,20 +2961,14 @@ describe("followup queue collect routing", () => { it("drops an aborted summarized room event before overflow delivery", async () => { const key = `test-overflow-summary-lifecycle-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const controller = new AbortController(); const onComplete = vi.fn(); const runFollowup = async (run: FollowupRun) => { calls.push(run); done.resolve(); }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -3568,16 +2996,10 @@ describe("followup queue collect routing", () => { it("retains summarized source identities through admitted overflow delivery", async () => { const key = `test-overflow-summary-admitted-lifecycle-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const sourceCompletions = [vi.fn(), vi.fn()]; const sourceCancellationRetirements = [vi.fn(), vi.fn()]; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); for (const [index, prompt] of ["first dropped", "second dropped"].entries()) { enqueueFollowupRun( @@ -3625,12 +3047,7 @@ describe("followup queue collect routing", () => { const sourceComplete = vi.fn(() => { events.push("source-complete"); }); - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -3692,12 +3109,7 @@ describe("followup queue collect routing", () => { await releaseRetry.promise; done.resolve(); }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -3739,8 +3151,7 @@ describe("followup queue collect routing", () => { it("collects compatible cancelable turns and completes each source lifecycle", async () => { const key = `test-collect-cancelable-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const firstComplete = vi.fn(); const secondComplete = vi.fn(); const runFollowup = async (run: FollowupRun) => { @@ -3845,8 +3256,7 @@ describe("followup queue collect routing", () => { it("collects transcript-owned turns under one aggregate recorder", async () => { const key = `test-collect-transcript-owner-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done, runFollowup } = createDrainRecorder(); const firstComplete = vi.fn(); const secondComplete = vi.fn(); const firstCorrelation = { begin: vi.fn() }; @@ -3880,10 +3290,7 @@ describe("followup queue collect routing", () => { ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - done.resolve(); - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(1); @@ -3915,8 +3322,7 @@ describe("followup queue collect routing", () => { it("pairs differing inbound runtime contexts inside one collected turn", async () => { const key = `test-collect-runtime-context-split-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done, runFollowup } = createDrainRecorder(); const settings: QueueSettings = { mode: "collect", debounceMs: 0 }; for (const [prompt, contextText] of [ @@ -3933,10 +3339,7 @@ describe("followup queue collect routing", () => { ); } - scheduleFollowupDrain(key, async (run) => { - calls.push(run); - done.resolve(); - }); + scheduleFollowupDrain(key, runFollowup); await done.promise; expect(calls).toHaveLength(1); @@ -3948,8 +3351,7 @@ describe("followup queue collect routing", () => { it("does not let one source cancel an admitted collected run", async () => { const key = `test-collect-transcript-cancel-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const canceled = new AbortController(); const survivor = new AbortController(); const sourceCompletions = [vi.fn(), vi.fn()]; @@ -4040,8 +3442,7 @@ describe("followup queue collect routing", () => { const canceled = new AbortController(); const canceledComplete = vi.fn(); const survivorComplete = vi.fn(); - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const settings: QueueSettings = { mode: "collect", debounceMs: 0 }; const enqueueSource = (prompt: string, onComplete: () => void, abortSignal?: AbortSignal) => { @@ -4085,8 +3486,7 @@ describe("followup queue collect routing", () => { it("keeps summarized work when a different cancelable live item is aborted", async () => { const key = `test-summary-owner-isolation-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const summarizedComplete = vi.fn(); const abortedComplete = vi.fn(); const aborted = new AbortController(); @@ -4097,12 +3497,7 @@ describe("followup queue collect routing", () => { calls.push(run); done.resolve(); }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -4137,8 +3532,7 @@ describe("followup queue collect routing", () => { it("removes an aborted elided source without leaking it into the summary", async () => { const key = `test-elided-summary-cancel-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const elidedComplete = vi.fn(); const elided = new AbortController(); const runFollowup = async (run: FollowupRun) => { @@ -4150,12 +3544,7 @@ describe("followup queue collect routing", () => { done.resolve(); } }; - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -4181,16 +3570,10 @@ describe("followup queue collect routing", () => { it("does not replay elided sources after an admitted summary failure", async () => { const key = `test-elided-summary-admitted-failure-${Date.now()}`; - const calls: FollowupRun[] = []; - const done = createDeferred(); + const { calls, done } = createDrainRecorder(); const elidedComplete = vi.fn(); const retainedComplete = vi.fn(); - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); enqueueFollowupRun( key, @@ -4235,12 +3618,7 @@ describe("followup queue collect routing", () => { const events: string[] = []; const done = createDeferred(); const secondAdmissionError = new Error("second overflow admission failed"); - const settings: QueueSettings = { - mode: "followup", - debounceMs: 0, - cap: 1, - dropPolicy: "summarize", - }; + const settings = createQueueSettings({ mode: "followup", cap: 1 }); const first = createRun({ prompt: "first dropped" }); first.turnAdoptionLifecycle = { diff --git a/src/channels/message/ingress-drain.test.ts b/src/channels/message/ingress-drain.test.ts index 6364af2ec60b..77defa8a0b47 100644 --- a/src/channels/message/ingress-drain.test.ts +++ b/src/channels/message/ingress-drain.test.ts @@ -24,6 +24,21 @@ import { type Payload = { text: string }; +function createTestIngressQueue( + stateDir: string, + options: Omit< + Parameters[0], + "channelId" | "accountId" | "stateDir" + > = {}, +) { + return createChannelIngressQueue({ + channelId: "test", + accountId: "a", + stateDir, + ...options, + }); +} + async function withTempState(fn: (stateDir: string) => Promise): Promise { const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-ingress-drain-")); try { @@ -46,12 +61,7 @@ describe("channel ingress drain", () => { it("crash-window: lost claim is recovered and dispatched exactly once", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - now: () => 1_000, - }); + const queue = createTestIngressQueue(stateDir, { now: () => 1_000 }); await queue.enqueue("evt-1", { text: "hello" }, { laneKey: "lane-a" }); const orphanClaim = await queue.claim("evt-1", { ownerId: "999:1:dead-owner" }); expect(orphanClaim).not.toBeNull(); @@ -84,11 +94,7 @@ describe("channel ingress drain", () => { it("dispatches a resubmitted dead letter exactly once", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-replay", { text: "recover" }, { laneKey: "lane-a" }); const originalClaim = await queue.claim("evt-replay", { ownerId: "worker" }); if (!originalClaim) { @@ -125,11 +131,7 @@ describe("channel ingress drain", () => { it("complete-at-adoption: adoption tombstones; settle is not required", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-adopt", { text: "x" }, { laneKey: "l1" }); let settleResolve!: () => void; @@ -162,11 +164,7 @@ describe("channel ingress drain", () => { it("deferred holds claim without complete until adopted or abandoned", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-def", { text: "x" }, { laneKey: "l1" }); const capturedLifecycles: ChannelIngressDispatchLifecycle[] = []; @@ -200,11 +198,7 @@ describe("channel ingress drain", () => { it("lets callers await an abandoned claim release", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-await-abandon", { text: "x" }, { laneKey: "l1" }); let finishRelease!: () => void; @@ -248,11 +242,7 @@ describe("channel ingress drain", () => { it("abandoned via turnAdoptionLifecycle releases claim with attempt increment", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-q", { text: "x" }, { laneKey: "l1" }); const drain = createChannelIngressDrain({ @@ -279,11 +269,7 @@ describe("channel ingress drain", () => { it("queued deferral→admission completes the claim exactly once via turnAdoptionLifecycle", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-admit", { text: "x" }, { laneKey: "l1" }); let adoptCount = 0; @@ -319,12 +305,7 @@ describe("channel ingress drain", () => { it("watchdog only guillotines pre-adoption stalls with handler-timeout", async () => { await withTempState(async (stateDir) => { let clock = 10_000; - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - now: () => clock, - }); + const queue = createTestIngressQueue(stateDir, { now: () => clock }); await queue.enqueue("evt-stall", { text: "x" }, { laneKey: "l1" }); const drain = createChannelIngressDrain({ @@ -355,12 +336,7 @@ describe("channel ingress drain", () => { it("watchdog guillotines deferred phase (timer not cleared by deferral)", async () => { await withTempState(async (stateDir) => { let clock = 30_000; - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - now: () => clock, - }); + const queue = createTestIngressQueue(stateDir, { now: () => clock }); await queue.enqueue("evt-def-stall", { text: "x" }, { laneKey: "l1" }); const drain = createChannelIngressDrain({ @@ -392,12 +368,7 @@ describe("channel ingress drain", () => { it("watchdog does not kill healthy long turns after adoption", async () => { await withTempState(async (stateDir) => { let clock = 20_000; - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - now: () => clock, - }); + const queue = createTestIngressQueue(stateDir, { now: () => clock }); await queue.enqueue("evt-long", { text: "x" }, { laneKey: "l1" }); let settleResolve!: () => void; @@ -432,11 +403,7 @@ describe("channel ingress drain", () => { it("supersede tombstones the superseded claim (never re-dispatches)", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("old", { text: "old" }, { laneKey: "shared" }); const firstLifecycles: ChannelIngressDispatchLifecycle[] = []; @@ -497,11 +464,7 @@ describe("channel ingress drain", () => { it("does not supersede without predicate", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("a1", { text: "a" }, { laneKey: "lane" }); let hold!: () => void; @@ -542,12 +505,7 @@ describe("channel ingress drain", () => { await withTempState(async (stateDir) => { const receivedAt = 100; let clock = receivedAt; - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - now: () => clock, - }); + const queue = createTestIngressQueue(stateDir, { now: () => clock }); await queue.enqueue("poison", { text: "x" }, { laneKey: "l", receivedAt }); // Burn attempts without aging past the gate. @@ -635,12 +593,7 @@ describe("channel ingress drain", () => { it("refreshes active claims on claimLeaseMs/3 while deferred", async () => { await withTempState(async (stateDir) => { let clock = 1_000; - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - now: () => clock, - }); + const queue = createTestIngressQueue(stateDir, { now: () => clock }); await queue.enqueue("evt-refresh", { text: "x" }, { laneKey: "l1" }); const refreshClaim = vi.fn(async () => true); @@ -687,11 +640,7 @@ describe("channel ingress drain", () => { it("throws IngressAdoptionLostError when onAdopted races supersede", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("old", { text: "old" }, { laneKey: "shared" }); const lifecycles: ChannelIngressDispatchLifecycle[] = []; @@ -734,11 +683,7 @@ describe("channel ingress drain", () => { it("retries tombstone complete failures then commits", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-tombstone", { text: "x" }, { laneKey: "l1" }); let completeAttempts = 0; @@ -775,11 +720,7 @@ describe("channel ingress drain", () => { it("holds claim ownership when tombstone complete keeps failing", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-wedge", { text: "x" }, { laneKey: "l1" }); queue.complete = async () => { @@ -815,11 +756,7 @@ describe("channel ingress drain", () => { it("does not steal live peer-drain claims; recovers after owner abort", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-peer", { text: "x" }, { laneKey: "l1" }); let releaseFirst!: () => void; @@ -873,11 +810,7 @@ describe("channel ingress drain", () => { it("throws IngressAdoptionLostError when complete returns false (lease reclaimed)", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-reclaim", { text: "x" }, { laneKey: "l1" }); queue.complete = async () => false; @@ -913,11 +846,7 @@ describe("channel ingress drain", () => { // Failure window: dispatch returns completed (side effects ran) but complete() // write fails while phase was still dispatching — must not release for replay. await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("evt-completed-tombstone-fail", { text: "ran" }, { laneKey: "l1" }); queue.complete = async () => { @@ -959,12 +888,7 @@ describe("channel ingress drain", () => { it("refreshClaim false aborts the handler mid-dispatch (lease reclaimed)", async () => { await withTempState(async (stateDir) => { let clock = 1_000; - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - now: () => clock, - }); + const queue = createTestIngressQueue(stateDir, { now: () => clock }); await queue.enqueue("evt-refresh-false", { text: "x" }, { laneKey: "l1" }); const refreshClaim = vi.fn(async () => false); @@ -1018,11 +942,7 @@ describe("channel ingress drain", () => { // Failure window: async shouldSupersedePending resolves after the pending // handler has already adopted — must revalidate and no-op. await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "a", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); await queue.enqueue("old", { text: "old" }, { laneKey: "shared" }); let releaseOld!: () => void; diff --git a/src/channels/message/ingress-queue.test.ts b/src/channels/message/ingress-queue.test.ts index a54c691c098d..a03768132f4e 100644 --- a/src/channels/message/ingress-queue.test.ts +++ b/src/channels/message/ingress-queue.test.ts @@ -19,6 +19,21 @@ import { createChannelIngressQueue } from "./ingress-queue.js"; type ChannelIngressTestDatabase = Pick; +function createTestIngressQueue( + stateDir: string, + options: Omit< + Parameters[0], + "channelId" | "accountId" | "stateDir" + > = {}, +) { + return createChannelIngressQueue({ + channelId: "test", + accountId: "account", + stateDir, + ...options, + }); +} + async function withTempState(fn: (stateDir: string) => Promise): Promise { const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-ingress-queue-")); try { @@ -36,16 +51,11 @@ describe("channel ingress queue", () => { it("deduplicates pending and completed ingress events", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue< + const queue = createTestIngressQueue< { text: string }, { source: string }, { handledBy: string } - >({ - channelId: "test", - accountId: "account", - stateDir, - now: () => 100, - }); + >(stateDir, { now: () => 100 }); const accepted = await queue.enqueue( "event-1", @@ -137,12 +147,7 @@ describe("channel ingress queue", () => { it("can bound pending scans and prune stale pending rows", async () => { await withTempState(async (stateDir) => { let clock = 1; - const queue = createChannelIngressQueue<{ index: number }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => clock++, - }); + const queue = createTestIngressQueue<{ index: number }>(stateDir, { now: () => clock++ }); await queue.enqueue("0002", { index: 2 }); await queue.enqueue("0001", { index: 1 }); @@ -160,12 +165,7 @@ describe("channel ingress queue", () => { it("does not prune protected rows while enforcing max-entry limits", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ index: number }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => 10, - }); + const queue = createTestIngressQueue<{ index: number }>(stateDir, { now: () => 10 }); await queue.enqueue("z", { index: 1 }); await queue.enqueue("a", { index: 2 }); @@ -180,12 +180,7 @@ describe("channel ingress queue", () => { it("prunes max-entry overflow across bounded batches", async () => { await withTempState(async (stateDir) => { let clock = 1; - const queue = createChannelIngressQueue<{ index: number }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => clock++, - }); + const queue = createTestIngressQueue<{ index: number }>(stateDir, { now: () => clock++ }); for (let index = 0; index < 520; index += 1) { await queue.enqueue(String(index).padStart(4, "0"), { index }); @@ -202,12 +197,7 @@ describe("channel ingress queue", () => { it("claims, releases, and skips blocked lanes", async () => { await withTempState(async (stateDir) => { let clock = 1; - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => clock++, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => clock++ }); await queue.enqueue("a", { text: "blocked" }, { laneKey: "chat-1", receivedAt: 1 }); await queue.enqueue("b", { text: "open" }, { laneKey: "chat-2", receivedAt: 2 }); @@ -245,12 +235,7 @@ describe("channel ingress queue", () => { it("claims next pending row by id when requested", async () => { await withTempState(async (stateDir) => { let clock = 1; - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => clock++, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => clock++ }); await queue.enqueue("0002", { text: "second" }, { receivedAt: 1 }); await queue.enqueue("0001", { text: "first" }, { receivedAt: 2 }); @@ -267,12 +252,7 @@ describe("channel ingress queue", () => { it("claims next only from candidate ids when provided", async () => { await withTempState(async (stateDir) => { let clock = 1; - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => clock++, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => clock++ }); await queue.enqueue("a", { text: "outside snapshot" }, { receivedAt: 1 }); await queue.enqueue("b", { text: "inside snapshot" }, { receivedAt: 2 }); @@ -290,12 +270,7 @@ describe("channel ingress queue", () => { it("derives missing lane keys before claiming next", async () => { await withTempState(async (stateDir) => { let clock = 1; - const queue = createChannelIngressQueue<{ lane: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => clock++, - }); + const queue = createTestIngressQueue<{ lane: string }>(stateDir, { now: () => clock++ }); await queue.enqueue("a", { lane: "blocked" }, { receivedAt: 1 }); await queue.enqueue("b", { lane: "open" }, { receivedAt: 2 }); @@ -317,12 +292,7 @@ describe("channel ingress queue", () => { it("blocks lanes claimed by candidate rows before claiming later candidates", async () => { await withTempState(async (stateDir) => { let clock = 1; - const queue = createChannelIngressQueue<{ lane: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => clock++, - }); + const queue = createTestIngressQueue<{ lane: string }>(stateDir, { now: () => clock++ }); await queue.enqueue("a", { lane: "chat-1" }, { receivedAt: 1 }); await queue.enqueue("b", { lane: "chat-1" }, { receivedAt: 2 }); @@ -345,12 +315,7 @@ describe("channel ingress queue", () => { it("requires claim tokens before mutating claimed rows", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => 10, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 }); await queue.enqueue("event-1", { text: "claimed" }); const claimed = await queue.claim("event-1", { ownerId: "worker" }); @@ -371,12 +336,7 @@ describe("channel ingress queue", () => { it("refreshes claimed rows only with the active claim token", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => 10, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 }); await queue.enqueue("event-1", { text: "claimed" }); const claimed = await queue.claim("event-1", { ownerId: "worker" }); @@ -407,12 +367,7 @@ describe("channel ingress queue", () => { it("does not let old claim tokens refresh recovered and reclaimed rows", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => 10, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 }); await queue.enqueue("event-1", { text: "claimed" }); const oldClaim = await queue.claim("event-1", { ownerId: "worker-1" }); @@ -436,12 +391,7 @@ describe("channel ingress queue", () => { it("does not recover a claim refreshed after stale recovery snapshots it", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => 10, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 }); await queue.enqueue("event-1", { text: "claimed" }); const claimed = await queue.claim("event-1", { ownerId: "worker" }); @@ -470,12 +420,7 @@ describe("channel ingress queue", () => { it("recovers stale claims and prunes completed or failed rows", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - now: () => 10, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => 10 }); await queue.enqueue("old", { text: "old" }); await queue.enqueue("keep", { text: "keep" }); @@ -589,11 +534,7 @@ describe("channel ingress queue", () => { it("skips a pending row with corrupt payload_json in listPending", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); await queue.enqueue("good-1", { text: "hello" }); insertCorruptRow(stateDir, '["test","account"]', "bad-1", { @@ -609,11 +550,7 @@ describe("channel ingress queue", () => { it("applies listPending limits after excluding corrupt payloads", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); for (let index = 0; index < 100; index += 1) { insertCorruptRow( stateDir, @@ -632,11 +569,7 @@ describe("channel ingress queue", () => { it("uses the queue JSON contract when listing deeply nested payloads", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); const nestedJson = `${"[".repeat(1001)}0${"]".repeat(1001)}`; const payload = JSON.parse(nestedJson); @@ -648,11 +581,7 @@ describe("channel ingress queue", () => { it("skips corrupt metadata_json in listPending", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }, { source: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }, { source: string }>(stateDir); await queue.enqueue("ev-1", { text: "ok" }, { metadata: { source: "good" } }); insertCorruptRow(stateDir, '["test","account"]', "ev-bad-meta", { @@ -669,11 +598,7 @@ describe("channel ingress queue", () => { it("skips a claimed row with corrupt payload_json in listClaims", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); await queue.enqueue("claim-ok", { text: "ok" }); insertCorruptRow(stateDir, '["test","account"]', "claim-bad", { @@ -699,11 +624,9 @@ describe("channel ingress queue", () => { it("skips corrupt completed_metadata_json during duplicate detection", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }, unknown, { handler: string }>({ - channelId: "test", - accountId: "account", + const queue = createTestIngressQueue<{ text: string }, unknown, { handler: string }>( stateDir, - }); + ); await queue.enqueue("comp-1", { text: "first" }); await queue.complete("comp-1", { metadata: { handler: "worker" }, completedAt: 150 }); @@ -729,11 +652,7 @@ describe("channel ingress queue", () => { it("claimNext skips a corrupt first pending row without lane derivation", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); // Insert the bad row first so it sorts before the good row. const earlyTime = 10; @@ -774,11 +693,7 @@ describe("channel ingress queue", () => { it("makes durable progress when a corrupt prefix fills the claim scan limit", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); insertCorruptRow(stateDir, '["test","account"]', "bad-first", { payload_json: "{corrupt", }); @@ -794,11 +709,7 @@ describe("channel ingress queue", () => { it("bounds corrupt reconciliation work per claimNext call", async () => { await withTempState(async (stateDir) => { const queueName = '["test","account"]'; - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); for (let index = 0; index < 101; index += 1) { insertCorruptRow(stateDir, queueName, `bad-${index.toString().padStart(3, "0")}`, { payload_json: "{corrupt", @@ -825,11 +736,7 @@ describe("channel ingress queue", () => { it("claim returns null for a corrupt pending row", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); await queue.enqueue("good-1", { text: "hello" }); insertCorruptRow(stateDir, '["test","account"]', "bad-direct", { @@ -860,11 +767,7 @@ describe("channel ingress queue", () => { it("handles valid JSON null payload correctly", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue(stateDir); // Valid JSON null should parse as null, not be treated as corrupt. await queue.enqueue("null-ok", null); @@ -876,11 +779,7 @@ describe("channel ingress queue", () => { it("tombstones a corrupt pending row on duplicate enqueue", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); insertCorruptRow(stateDir, '["test","account"]', "dup-bad", { payload_json: "{corrupt", @@ -915,11 +814,7 @@ describe("channel ingress queue", () => { it("does not tombstone a corrupt actively claimed row on duplicate enqueue", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); insertCorruptRow(stateDir, '["test","account"]', "dup-claimed-bad", { payload_json: "{corrupt", status: "claimed", @@ -955,11 +850,7 @@ describe("channel ingress queue", () => { it("tombstones corrupt claimed rows during stale recovery", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); const oldTime = 10; insertCorruptRow(stateDir, '["test","account"]', "stale-bad", { @@ -998,11 +889,7 @@ describe("channel ingress queue", () => { it("does not bypass recovery policy for a corrupt stale claim", async () => { await withTempState(async (stateDir) => { - const queue = createChannelIngressQueue<{ text: string }>({ - channelId: "test", - accountId: "account", - stateDir, - }); + const queue = createTestIngressQueue<{ text: string }>(stateDir); insertCorruptRow(stateDir, '["test","account"]', "stale-policy-bad", { payload_json: "{corrupt", status: "claimed",