From f5e3b5ef547144121d81199ff4595f3370361f99 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 6 Aug 2026 14:47:36 -0700 Subject: [PATCH] refactor(plugins): single-source question reactions and preflight audio (#119987) --- .../.generated/plugin-sdk-api-baseline.sha256 | 4 +- .../monitor/message-handler.preflight.test.ts | 17 +- .../src/monitor/preflight-audio.runtime.ts | 10 -- .../src/monitor/preflight-audio.test.ts | 17 +- .../discord/src/monitor/preflight-audio.ts | 55 +++--- .../imessage/src/question-reactions.test.ts | 11 +- extensions/imessage/src/question-reactions.ts | 134 ++++---------- .../monitor/handler.audio-preflight.test.ts | 47 +++-- .../matrix/monitor/preflight-audio.runtime.ts | 18 -- .../matrix/monitor/preflight-audio.test.ts | 135 ++------------ .../src/matrix/monitor/preflight-audio.ts | 88 ++-------- .../signal/src/question-reactions.test.ts | 10 +- extensions/signal/src/question-reactions.ts | 134 +++++--------- .../preflight-audio.runtime.ts | 3 - .../message-handler/preflight-audio.test.ts | 81 ++------- .../message-handler/preflight-audio.ts | 82 ++------- .../monitor/message-handler/prepare.test.ts | 27 ++- .../whatsapp/src/question-reactions.test.ts | 57 +++--- extensions/whatsapp/src/question-reactions.ts | 132 ++++---------- scripts/plugin-sdk-surface-report.mjs | 6 +- src/media-understanding/echo-transcript.ts | 7 +- .../media-understanding-runtime.test.ts | 146 ++++++++++++++++ src/plugin-sdk/media-understanding-runtime.ts | 111 ++++++++++++ .../question-gateway-runtime.test.ts | 164 ++++++++++++++++++ src/plugin-sdk/question-gateway-runtime.ts | 134 ++++++++++++++ 25 files changed, 870 insertions(+), 760 deletions(-) delete mode 100644 extensions/discord/src/monitor/preflight-audio.runtime.ts delete mode 100644 extensions/matrix/src/matrix/monitor/preflight-audio.runtime.ts delete mode 100644 extensions/slack/src/monitor/message-handler/preflight-audio.runtime.ts create mode 100644 src/plugin-sdk/media-understanding-runtime.test.ts create mode 100644 src/plugin-sdk/question-gateway-runtime.test.ts diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 43a41ac65677..b76270b671e2 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -86,7 +86,7 @@ a5f59c9acbcaa3f82247bf806eb5ba08032373fb853719f0ec9457690f16fc70 module/media-m 14e6c4c8471207a204710322e88ae2cda12b9e37c726f508f0da45e136262470 module/media-runtime 7a5a1de743fff9d5a86515b501ffd88e7fa49970c9f46a5153816e2ebaef7bb0 module/media-store c5e3eb1a584f4b8126d9d6c177a840ec9103671e8d1242634ee67db9b5b9e573 module/media-understanding -c0ffaed532578cf33493992e1ff806b2268b8e3774a92edbaede5cf5bda162a6 module/media-understanding-runtime +2288d50521f3323c070a0b74b17e0e647f7ee0e422ccc12989a9ffb8eefacdd4 module/media-understanding-runtime bebd2931dc51d67c063ff19fa1c278f8dcfe00ab23cfbd480d47329ea8e5088e module/meeting-runtime ea56ea0455c62f292e1c7b3e56cac6eb3fea00e548f9ab2f4a69e0b41b0a2d96 module/memory-core-host-engine-foundation 00d6f8bc78256558972d431b7983d7b774e0a98d802a1eec533d2cae1e1f1986 module/memory-host-core @@ -103,7 +103,7 @@ d7dd3c82a4b1df9e4144b078caf0e6ab95e7b11c5046b53a38f01143a7fa0eec module/plugin- de508df6d9cfa9d21c4524989dd8bd142fb60cb4ae980193e0907767d76e6022 module/provider-auth 71bebeac51e701cd7c8e63d22754b9bcbd82b024aced303aa055d229781129d7 module/provider-catalog-runtime 56151035047a69e6163d5578023d00f51a2413b777f3784af88e06261c039345 module/proxy-capture -aa2a56b4448c8ebdec9d06aac95d809995f533093d42fa32cd75e1d852967245 module/question-gateway-runtime +cc3f586640ca1f1988de64ab2b6c9a119024a911a27289fdacfa0901d093c8fe module/question-gateway-runtime 2e09c3181e79e157ed5366b144d116ef8cc06023256ace3fa59b35c43cab513a module/reply-chunking 7994045066b29af1fc6b36ae32068f2a6f277195971af84701cb739cc23d0579 module/reply-dispatch-runtime ac2b199e95c5c8b1e2a65e62bd41d1b6322e531bca294ef4979a297a12640bce module/reply-history diff --git a/extensions/discord/src/monitor/message-handler.preflight.test.ts b/extensions/discord/src/monitor/message-handler.preflight.test.ts index 05392cd0781f..631cbcb6d8c7 100644 --- a/extensions/discord/src/monitor/message-handler.preflight.test.ts +++ b/extensions/discord/src/monitor/message-handler.preflight.test.ts @@ -12,9 +12,20 @@ const saveRemoteMediaMock = vi.hoisted(() => vi.fn()); vi.mock("../pluralkit.js", () => ({ fetchPluralKitMessageInfo: (...args: unknown[]) => fetchPluralKitMessageInfoMock(...args), })); -vi.mock("./preflight-audio.runtime.js", () => ({ - transcribeFirstAudio: transcribeFirstAudioMock, -})); +vi.mock("openclaw/plugin-sdk/media-understanding-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + createChannelPreflightAudio: ( + params: Parameters[0], + ) => + actual.createChannelPreflightAudio({ + ...params, + transcribeFirstAudio: transcribeFirstAudioMock, + }), + }; +}); vi.mock("./dm-command-auth.js", async (importOriginal) => ({ ...(await importOriginal()), resolveDiscordDmCommandAccess: resolveDiscordDmCommandAccessMock, diff --git a/extensions/discord/src/monitor/preflight-audio.runtime.ts b/extensions/discord/src/monitor/preflight-audio.runtime.ts deleted file mode 100644 index 24d71e783b5d..000000000000 --- a/extensions/discord/src/monitor/preflight-audio.runtime.ts +++ /dev/null @@ -1,10 +0,0 @@ -// Discord plugin module implements preflight audio behavior. -import { transcribeFirstAudio as transcribeFirstAudioImpl } from "openclaw/plugin-sdk/media-runtime"; - -type TranscribeFirstAudio = typeof import("openclaw/plugin-sdk/media-runtime").transcribeFirstAudio; - -export async function transcribeFirstAudio( - ...args: Parameters -): ReturnType { - return await transcribeFirstAudioImpl(...args); -} diff --git a/extensions/discord/src/monitor/preflight-audio.test.ts b/extensions/discord/src/monitor/preflight-audio.test.ts index 073090d5f0a4..d9203cdbf35c 100644 --- a/extensions/discord/src/monitor/preflight-audio.test.ts +++ b/extensions/discord/src/monitor/preflight-audio.test.ts @@ -3,9 +3,20 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const transcribeFirstAudioMock = vi.hoisted(() => vi.fn()); -vi.mock("./preflight-audio.runtime.js", () => ({ - transcribeFirstAudio: transcribeFirstAudioMock, -})); +vi.mock("openclaw/plugin-sdk/media-understanding-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + createChannelPreflightAudio: ( + params: Parameters[0], + ) => + actual.createChannelPreflightAudio({ + ...params, + transcribeFirstAudio: transcribeFirstAudioMock, + }), + }; +}); import { resolveDiscordPreflightAudioMentionContext } from "./preflight-audio.js"; diff --git a/extensions/discord/src/monitor/preflight-audio.ts b/extensions/discord/src/monitor/preflight-audio.ts index b5cd728b8500..08e15ec26340 100644 --- a/extensions/discord/src/monitor/preflight-audio.ts +++ b/extensions/discord/src/monitor/preflight-audio.ts @@ -1,14 +1,9 @@ // Discord plugin module implements preflight audio behavior. import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; import { getFileExtension } from "openclaw/plugin-sdk/media-mime"; -import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { createChannelPreflightAudio } from "openclaw/plugin-sdk/media-understanding-runtime"; import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; -const loadDiscordPreflightAudioRuntime = createLazyRuntimeModule( - () => import("./preflight-audio.runtime.js"), -); - type DiscordAudioAttachment = { content_type?: string; duration_secs?: number; @@ -44,15 +39,22 @@ function inferAudioAttachmentMime(attachment: DiscordAudioAttachment): string | return ext ? AUDIO_ATTACHMENT_MIME_BY_EXT.get(ext) : undefined; } +const discordPreflightAudio = createChannelPreflightAudio({ + channel: "discord", + isAudio: (attachment: DiscordAudioAttachment) => + Boolean(normalizeOptionalString(attachment.url) && inferAudioAttachmentMime(attachment)), + // Discord uses this transcript only for mention admission and has no deferred + // admitted-message echo, so its transcription config must remain unchanged. + deferTranscriptEcho: false, +}); + function collectAudioAttachments( attachments: DiscordAudioAttachment[] | undefined, ): DiscordAudioAttachment[] { if (!Array.isArray(attachments)) { return []; } - return attachments.filter( - (att) => normalizeOptionalString(att.url) && inferAudioAttachmentMime(att), - ); + return attachments.filter(discordPreflightAudio.isAudio); } export async function resolveDiscordPreflightAudioMentionContext(params: { @@ -87,32 +89,19 @@ export async function resolveDiscordPreflightAudioMentionContext(params: { hasTypedText, }; } - try { - const { transcribeFirstAudio } = await loadDiscordPreflightAudioRuntime(); - if (params.abortSignal?.aborted) { - return { - hasAudioAttachment, - hasTypedText, - }; - } - const media = audioAttachments.flatMap((attachment) => { - const url = normalizeOptionalString(attachment.url); - return url ? [{ url, contentType: inferAudioAttachmentMime(attachment) }] : []; - }); - if (media.length > 0) { - transcript = await transcribeFirstAudio({ - ctx: { - media, - }, + const media = audioAttachments.flatMap((attachment) => { + const url = normalizeOptionalString(attachment.url); + return url ? [{ url, contentType: inferAudioAttachmentMime(attachment) }] : []; + }); + if (media.length > 0) { + transcript = await discordPreflightAudio.resolve({ + request: { + ctx: { media }, cfg: params.cfg, agentDir: undefined, - }); - if (params.abortSignal?.aborted) { - transcript = undefined; - } - } - } catch (err) { - logVerbose(`discord: audio preflight transcription failed: ${String(err)}`); + }, + abortSignal: params.abortSignal, + }); } } diff --git a/extensions/imessage/src/question-reactions.test.ts b/extensions/imessage/src/question-reactions.test.ts index b2635808a71e..280d3413421a 100644 --- a/extensions/imessage/src/question-reactions.test.ts +++ b/extensions/imessage/src/question-reactions.test.ts @@ -16,7 +16,6 @@ vi.mock("openclaw/plugin-sdk/question-gateway-runtime", async (importOriginal) = import { questionGatewayRuntime } from "openclaw/plugin-sdk/question-gateway-runtime"; import { - clearIMessageQuestionReactionTargetsForTest, hasIMessageQuestionReactionTarget, maybeResolveIMessageQuestionReaction, registerIMessageQuestionReactionTargetForDeliveredPayload, @@ -45,7 +44,6 @@ function buildPayload() { describe("iMessage question reactions", () => { beforeEach(() => { - clearIMessageQuestionReactionTargetsForTest(); hoisted.resolve.mockReset().mockResolvedValue({ status: "answered", questionId: "choice", @@ -53,7 +51,7 @@ describe("iMessage question reactions", () => { }); }); - it("recognizes a stable GUID and consumes a stale duplicate", async () => { + it("normalizes stable GUIDs and routes their numbered reactions", async () => { expect( registerIMessageQuestionReactionTargetForDeliveredPayload({ accountId: "default", @@ -101,13 +99,6 @@ describe("iMessage question reactions", () => { logDebug: vi.fn(), }; - await expect( - maybeResolveIMessageQuestionReaction({ - ...params, - message: { ...message, reaction_emoji: "4️⃣" }, - }), - ).resolves.toBe(true); - await expect(maybeResolveIMessageQuestionReaction(params)).resolves.toBe(true); await expect(maybeResolveIMessageQuestionReaction(params)).resolves.toBe(true); expect(hoisted.resolve).toHaveBeenCalledOnce(); expect(hoisted.resolve).toHaveBeenCalledWith( diff --git a/extensions/imessage/src/question-reactions.ts b/extensions/imessage/src/question-reactions.ts index d3cd54d19d89..08b1479399ae 100644 --- a/extensions/imessage/src/question-reactions.ts +++ b/extensions/imessage/src/question-reactions.ts @@ -1,59 +1,34 @@ // iMessage transport binding for numbered ask_user reactions. import type { OutboundDeliveryResult } from "openclaw/plugin-sdk/channel-send-result"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { questionGatewayRuntime } from "openclaw/plugin-sdk/question-gateway-runtime"; +import { + createQuestionReactionTargetStore, + questionGatewayRuntime, +} from "openclaw/plugin-sdk/question-gateway-runtime"; import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; import { resolveIMessageReactionContext } from "./monitor/reaction-context.js"; import type { IMessagePayload } from "./monitor/types.js"; -const TARGET_TTL_MS = 24 * 60 * 60 * 1_000; - -type IMessageQuestionReactionTarget = { - questionId: string; - optionValues: string[]; - terminal: boolean; - expiresAtMs: number; - cleanupTimer: ReturnType; -}; - -const targets = new Map(); - -function storeTarget(key: string, binding: { questionId: string; optionValues: string[] }): void { - const existing = targets.get(key); - if (existing) { - clearTimeout(existing.cleanupTimer); - } - const target: IMessageQuestionReactionTarget = { - ...binding, - terminal: false, - expiresAtMs: Date.now() + TARGET_TTL_MS, - cleanupTimer: setTimeout(() => { - if (targets.get(key) === target) { - targets.delete(key); - } - }, TARGET_TTL_MS), - }; - target.cleanupTimer.unref?.(); - targets.set(key, target); - questionGatewayRuntime.registerChannelDelivery({ - questionId: binding.questionId, - deliveryId: `imessage-reaction:${key}`, - finalize: () => { - target.terminal = true; - }, - }); -} - function normalizeGuid(value: string): string { return value.trim().replace(/^p:\d+\//iu, ""); } -function buildKey(accountId: string, messageGuid: string): string | null { - const account = accountId.trim(); - const guid = normalizeGuid(messageGuid); +type IMessageQuestionReactionIdentity = { accountId: string; messageGuid: string }; + +function buildKey(identity: IMessageQuestionReactionIdentity): string | null { + const account = identity.accountId.trim(); + const guid = normalizeGuid(identity.messageGuid); return account && guid ? `${account}:${guid}` : null; } +const questionReactionTargets = createQuestionReactionTargetStore({ + channel: "imessage", + channelDisplayName: "iMessage", + buildKey, + registerChannelDelivery: questionGatewayRuntime.registerChannelDelivery, + resolveReaction: questionGatewayRuntime.resolveReaction, +}); + function reactionCandidates( message: IMessagePayload, bodyText: string, @@ -95,12 +70,14 @@ export function registerIMessageQuestionReactionTargetForDeliveredPayload(params typeof result.meta?.imessageMessageGuid === "string" ? result.meta.imessageMessageGuid : result.messageId; - const key = buildKey(params.accountId, guid); - if (!key || /^\d+$/u.test(normalizeGuid(guid))) { + if (/^\d+$/u.test(normalizeGuid(guid))) { continue; } - storeTarget(key, binding); - registered = true; + registered = + questionReactionTargets.register(binding, { + accountId: params.accountId, + messageGuid: guid, + }) || registered; } return registered; } @@ -118,10 +95,9 @@ export function hasIMessageQuestionReactionTarget(params: { ) { return false; } - return reaction.guids.some((guid) => { - const key = buildKey(params.accountId, guid); - return key ? targets.has(key) : false; - }); + return questionReactionTargets.has( + reaction.guids.map((messageGuid) => ({ accountId: params.accountId, messageGuid })), + ); } export async function maybeResolveIMessageQuestionReaction(params: { @@ -140,51 +116,15 @@ export async function maybeResolveIMessageQuestionReaction(params: { if (!reaction || reaction.action === "removed" || optionIndex === undefined) { return false; } - let target: IMessageQuestionReactionTarget | undefined; - for (const guid of reaction.guids) { - const key = buildKey(params.accountId, guid); - target = key ? targets.get(key) : undefined; - if (target) { - break; - } - } - if (!target) { - return false; - } - if (target.expiresAtMs <= Date.now() || target.terminal) { - target.terminal = true; - params.logDebug?.(`imessage: stale question reaction ignored id=${target.questionId}`); - return true; - } - const optionValue = target.optionValues[optionIndex]; - if (!optionValue) { - params.logDebug?.(`imessage: out-of-range question reaction ignored id=${target.questionId}`); - return true; - } - try { - const result = await questionGatewayRuntime.resolveReaction({ - cfg: params.cfg, - questionId: target.questionId, - optionValue, - senderId: params.senderId, - gatewayUrl: params.gatewayUrl, - clientDisplayName: `iMessage question (${params.senderId})`, - }); - target.terminal = result?.status === "answered" || result?.status === "already-terminal"; - if (result?.status === "already-terminal") { - params.logDebug?.(`imessage: stale question reaction ignored id=${target.questionId}`); - } - } catch (error) { - params.logDebug?.( - `imessage: question reaction failed id=${target.questionId}: ${String(error)}`, - ); - } - return true; -} - -export function clearIMessageQuestionReactionTargetsForTest(): void { - for (const target of targets.values()) { - clearTimeout(target.cleanupTimer); - } - targets.clear(); + return await questionReactionTargets.resolve({ + identities: reaction.guids.map((messageGuid) => ({ + accountId: params.accountId, + messageGuid, + })), + optionIndex, + cfg: params.cfg, + senderId: params.senderId, + gatewayUrl: params.gatewayUrl, + logDebug: params.logDebug, + }); } diff --git a/extensions/matrix/src/matrix/monitor/handler.audio-preflight.test.ts b/extensions/matrix/src/matrix/monitor/handler.audio-preflight.test.ts index cb8b39da3628..595e16f604dd 100644 --- a/extensions/matrix/src/matrix/monitor/handler.audio-preflight.test.ts +++ b/extensions/matrix/src/matrix/monitor/handler.audio-preflight.test.ts @@ -6,12 +6,13 @@ import { createMatrixRoomMessageEvent, } from "./handler.test-helpers.js"; -const { downloadMatrixMediaMock, sendDurableMessageBatchMock, transcribeFirstAudioMock } = - vi.hoisted(() => ({ +const { downloadMatrixMediaMock, sendTranscriptEchoMock, transcribeFirstAudioMock } = vi.hoisted( + () => ({ downloadMatrixMediaMock: vi.fn(), - sendDurableMessageBatchMock: vi.fn(), + sendTranscriptEchoMock: vi.fn(), transcribeFirstAudioMock: vi.fn(), - })); + }), +); vi.mock("./media.js", async () => { const actual = await vi.importActual("./media.js"); @@ -21,10 +22,21 @@ vi.mock("./media.js", async () => { }; }); -vi.mock("./preflight-audio.runtime.js", () => ({ - sendDurableMessageBatch: sendDurableMessageBatchMock, - transcribeFirstAudio: transcribeFirstAudioMock, -})); +vi.mock("openclaw/plugin-sdk/media-understanding-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + createChannelPreflightAudio: ( + params: Parameters[0], + ) => + actual.createChannelPreflightAudio({ + ...params, + sendTranscriptEcho: sendTranscriptEchoMock, + transcribeFirstAudio: transcribeFirstAudioMock, + }), + }; +}); function createAudioPreflightHarness( overrides: Parameters[0] = {}, @@ -80,7 +92,7 @@ function expectLatestInboundContext( describe("createMatrixRoomMessageHandler audio preflight", () => { beforeEach(() => { downloadMatrixMediaMock.mockReset(); - sendDurableMessageBatchMock.mockReset(); + sendTranscriptEchoMock.mockReset(); transcribeFirstAudioMock.mockReset(); installMatrixMonitorTestRuntime(); }); @@ -190,7 +202,7 @@ describe("createMatrixRoomMessageHandler audio preflight", () => { contentType: "audio/ogg", placeholder: "[matrix audio attachment]", }); - sendDurableMessageBatchMock.mockResolvedValue({ status: "sent", results: [] }); + sendTranscriptEchoMock.mockResolvedValue(undefined); transcribeFirstAudioMock.mockResolvedValue("hello bot"); const { handler } = createAudioPreflightHarness({ cfg: { @@ -209,14 +221,15 @@ describe("createMatrixRoomMessageHandler audio preflight", () => { }), ); - expect(sendDurableMessageBatchMock).toHaveBeenCalledWith( + expect(sendTranscriptEchoMock).toHaveBeenCalledWith( expect.objectContaining({ - channel: "matrix", - to: "room:!room:example.org", - accountId: "ops", - payloads: [{ text: '📝 "hello bot"' }], - bestEffort: true, - durability: "best_effort", + ctx: expect.objectContaining({ + Provider: "matrix", + OriginatingTo: "room:!room:example.org", + AccountId: "ops", + }), + transcript: "hello bot", + format: '📝 "{transcript}"', }), ); }); diff --git a/extensions/matrix/src/matrix/monitor/preflight-audio.runtime.ts b/extensions/matrix/src/matrix/monitor/preflight-audio.runtime.ts deleted file mode 100644 index 880359696b4d..000000000000 --- a/extensions/matrix/src/matrix/monitor/preflight-audio.runtime.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { sendDurableMessageBatch as sendDurableMessageBatchImpl } from "openclaw/plugin-sdk/channel-outbound"; -import { transcribeFirstAudio as transcribeFirstAudioImpl } from "openclaw/plugin-sdk/media-runtime"; - -type TranscribeFirstAudio = typeof import("openclaw/plugin-sdk/media-runtime").transcribeFirstAudio; -type SendDurableMessageBatch = - typeof import("openclaw/plugin-sdk/channel-outbound").sendDurableMessageBatch; - -export async function transcribeFirstAudio( - ...args: Parameters -): ReturnType { - return await transcribeFirstAudioImpl(...args); -} - -export async function sendDurableMessageBatch( - ...args: Parameters -): ReturnType { - return await sendDurableMessageBatchImpl(...args); -} diff --git a/extensions/matrix/src/matrix/monitor/preflight-audio.test.ts b/extensions/matrix/src/matrix/monitor/preflight-audio.test.ts index 9e1444fce684..0d9ce01cca71 100644 --- a/extensions/matrix/src/matrix/monitor/preflight-audio.test.ts +++ b/extensions/matrix/src/matrix/monitor/preflight-audio.test.ts @@ -1,20 +1,28 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { sendDurableMessageBatchMock, transcribeFirstAudioMock } = vi.hoisted(() => ({ - sendDurableMessageBatchMock: vi.fn(), +const { transcribeFirstAudioMock } = vi.hoisted(() => ({ transcribeFirstAudioMock: vi.fn(), })); -vi.mock("./preflight-audio.runtime.js", () => ({ - sendDurableMessageBatch: sendDurableMessageBatchMock, - transcribeFirstAudio: transcribeFirstAudioMock, -})); +vi.mock("openclaw/plugin-sdk/media-understanding-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + createChannelPreflightAudio: ( + params: Parameters[0], + ) => + actual.createChannelPreflightAudio({ + ...params, + transcribeFirstAudio: transcribeFirstAudioMock, + }), + }; +}); import { formatMatrixAudioTranscript, isMatrixAudioContent, resolveMatrixPreflightAudioTranscript, - sendMatrixPreflightAudioTranscriptEcho, } from "./preflight-audio.js"; const cfg = {} as import("openclaw/plugin-sdk/config-contracts").OpenClawConfig; @@ -43,7 +51,6 @@ describe("formatMatrixAudioTranscript", () => { describe("resolveMatrixPreflightAudioTranscript", () => { beforeEach(() => { - sendDurableMessageBatchMock.mockReset(); transcribeFirstAudioMock.mockReset(); }); @@ -79,116 +86,4 @@ describe("resolveMatrixPreflightAudioTranscript", () => { ); expect(transcript).toBe("hello from voice"); }); - - it("suppresses shared echo during pre-mention transcription", async () => { - const echoCfg = { - tools: { media: { audio: { echoTranscript: true, echoFormat: "echo: {transcript}" } } }, - } as import("openclaw/plugin-sdk/config-contracts").OpenClawConfig; - transcribeFirstAudioMock.mockResolvedValue("hello from voice"); - - await resolveMatrixPreflightAudioTranscript({ - mediaPath: "/tmp/inbound/voice.ogg", - mediaContentType: "audio/ogg", - cfg: echoCfg, - accountId: "ops", - chatType: "channel", - originatingTo: "room:!room:example.org", - sessionKey: "agent:main:matrix:channel:!room:example.org", - }); - - const callCfg = transcribeFirstAudioMock.mock.calls[0]?.[0]?.cfg as - | { tools?: { media?: { audio?: { echoTranscript?: unknown } } } } - | undefined; - expect(callCfg?.tools?.media?.audio?.echoTranscript).toBe(false); - }); - - it("swallows provider failures and aborts", async () => { - transcribeFirstAudioMock.mockRejectedValue(new Error("STT down")); - await expect( - resolveMatrixPreflightAudioTranscript({ - mediaPath: "/tmp/inbound/voice.ogg", - cfg, - accountId: "ops", - chatType: "direct", - originatingTo: "room:!dm:example.org", - sessionKey: "agent:main:matrix:direct:@frank:example.org", - }), - ).resolves.toBeUndefined(); - - const controller = new AbortController(); - controller.abort(); - transcribeFirstAudioMock.mockClear(); - await expect( - resolveMatrixPreflightAudioTranscript({ - mediaPath: "/tmp/inbound/voice.ogg", - cfg, - accountId: "ops", - chatType: "direct", - originatingTo: "room:!dm:example.org", - sessionKey: "agent:main:matrix:direct:@frank:example.org", - abortSignal: controller.signal, - }), - ).resolves.toBeUndefined(); - expect(transcribeFirstAudioMock).not.toHaveBeenCalled(); - }); -}); - -describe("sendMatrixPreflightAudioTranscriptEcho", () => { - beforeEach(() => { - sendDurableMessageBatchMock.mockReset(); - transcribeFirstAudioMock.mockReset(); - }); - - it("sends accepted Matrix preflight transcript echoes through durable delivery", async () => { - sendDurableMessageBatchMock.mockResolvedValue({ status: "sent", results: [] }); - await sendMatrixPreflightAudioTranscriptEcho({ - transcript: "hello bot", - cfg: { - tools: { media: { audio: { echoTranscript: true, echoFormat: "heard: {transcript}" } } }, - } as import("openclaw/plugin-sdk/config-contracts").OpenClawConfig, - accountId: "ops", - originatingTo: "room:!room:example.org", - messageThreadId: "$thread", - }); - - expect(sendDurableMessageBatchMock).toHaveBeenCalledWith({ - cfg: expect.any(Object), - channel: "matrix", - to: "room:!room:example.org", - accountId: "ops", - threadId: "$thread", - payloads: [{ text: "heard: hello bot" }], - bestEffort: true, - durability: "best_effort", - }); - }); - - it("keeps dollar sequences in the transcript literal", async () => { - sendDurableMessageBatchMock.mockResolvedValue({ status: "sent", results: [] }); - await sendMatrixPreflightAudioTranscriptEcho({ - transcript: "tickets cost $$40, confirm with $&", - cfg: { - tools: { media: { audio: { echoTranscript: true, echoFormat: "heard: {transcript}" } } }, - } as import("openclaw/plugin-sdk/config-contracts").OpenClawConfig, - accountId: "ops", - originatingTo: "room:!room:example.org", - }); - - expect(sendDurableMessageBatchMock).toHaveBeenCalledWith( - expect.objectContaining({ - payloads: [{ text: "heard: tickets cost $$40, confirm with $&" }], - }), - ); - }); - - it("does not echo when transcript echo is disabled", async () => { - await sendMatrixPreflightAudioTranscriptEcho({ - transcript: "hello bot", - cfg, - accountId: "ops", - originatingTo: "room:!room:example.org", - }); - - expect(sendDurableMessageBatchMock).not.toHaveBeenCalled(); - }); }); diff --git a/extensions/matrix/src/matrix/monitor/preflight-audio.ts b/extensions/matrix/src/matrix/monitor/preflight-audio.ts index 419f1ebb0264..a12146a4d528 100644 --- a/extensions/matrix/src/matrix/monitor/preflight-audio.ts +++ b/extensions/matrix/src/matrix/monitor/preflight-audio.ts @@ -1,42 +1,10 @@ import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; -import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; -const MATRIX_DEFAULT_ECHO_TRANSCRIPT_FORMAT = '📝 "{transcript}"'; - -const loadMatrixPreflightAudioRuntime = createLazyRuntimeModule( - () => import("./preflight-audio.runtime.js"), -); +import { createChannelPreflightAudio } from "openclaw/plugin-sdk/media-understanding-runtime"; export function formatMatrixAudioTranscript(transcript: string): string { return `[Audio transcript (machine-generated, untrusted)]: ${JSON.stringify(transcript)}`; } -function formatMatrixAudioTranscriptEcho(transcript: string, format: string): string { - // Function replacer keeps `$` sequences in the transcript literal instead of - // being parsed as String.prototype.replace substitution patterns. - return format.replace("{transcript}", () => transcript); -} - -function suppressMatrixPreflightAudioEcho(cfg: OpenClawConfig): OpenClawConfig { - const audio = cfg.tools?.media?.audio; - if (!audio?.echoTranscript) { - return cfg; - } - return { - ...cfg, - tools: { - ...cfg.tools, - media: { - ...cfg.tools?.media, - audio: { - ...audio, - echoTranscript: false, - }, - }, - }, - }; -} - export function isMatrixAudioContent(params: { msgtype?: string; mimetype?: string }): boolean { if (params.msgtype === "m.audio") { return true; @@ -47,6 +15,11 @@ export function isMatrixAudioContent(params: { msgtype?: string; mimetype?: stri return false; } +const matrixPreflightAudio = createChannelPreflightAudio({ + channel: "matrix", + isAudio: isMatrixAudioContent, +}); + export async function resolveMatrixPreflightAudioTranscript(params: { mediaPath: string; mediaContentType?: string; @@ -58,15 +31,8 @@ export async function resolveMatrixPreflightAudioTranscript(params: { sessionKey: string; abortSignal?: AbortSignal; }): Promise { - if (params.abortSignal?.aborted) { - return undefined; - } - try { - const { transcribeFirstAudio } = await loadMatrixPreflightAudioRuntime(); - if (params.abortSignal?.aborted) { - return undefined; - } - const transcript = await transcribeFirstAudio({ + return await matrixPreflightAudio.resolve({ + request: { ctx: { media: [{ path: params.mediaPath, contentType: params.mediaContentType }], Provider: "matrix", @@ -78,13 +44,10 @@ export async function resolveMatrixPreflightAudioTranscript(params: { ChatType: params.chatType, SessionKey: params.sessionKey, }, - cfg: suppressMatrixPreflightAudioEcho(params.cfg), - }); - return params.abortSignal?.aborted ? undefined : transcript; - } catch (err) { - logVerbose(`matrix: audio preflight transcription failed: ${String(err)}`); - return undefined; - } + cfg: params.cfg, + }, + abortSignal: params.abortSignal, + }); } export async function sendMatrixPreflightAudioTranscriptEcho(params: { @@ -94,30 +57,5 @@ export async function sendMatrixPreflightAudioTranscriptEcho(params: { originatingTo: string; messageThreadId?: string; }): Promise { - const audio = params.cfg.tools?.media?.audio; - if (!audio?.echoTranscript) { - return; - } - const text = formatMatrixAudioTranscriptEcho( - params.transcript, - audio.echoFormat ?? MATRIX_DEFAULT_ECHO_TRANSCRIPT_FORMAT, - ); - try { - const { sendDurableMessageBatch } = await loadMatrixPreflightAudioRuntime(); - const send = await sendDurableMessageBatch({ - cfg: params.cfg, - channel: "matrix", - to: params.originatingTo, - accountId: params.accountId, - threadId: params.messageThreadId, - payloads: [{ text }], - bestEffort: true, - durability: "best_effort", - }); - if (send.status === "failed") { - throw send.error; - } - } catch (err) { - logVerbose(`matrix: audio transcript echo failed: ${String(err)}`); - } + await matrixPreflightAudio.send(params); } diff --git a/extensions/signal/src/question-reactions.test.ts b/extensions/signal/src/question-reactions.test.ts index a78845f400a7..49afa3dc82ff 100644 --- a/extensions/signal/src/question-reactions.test.ts +++ b/extensions/signal/src/question-reactions.test.ts @@ -50,7 +50,7 @@ describe("Signal question reactions", () => { }); }); - it("matches the bot-authored target and ignores a duplicate", async () => { + it("matches the bot-authored target before routing a numbered reaction", async () => { const payload = buildPayload(); expect( registerSignalQuestionReactionTargetForDeliveredPayload({ @@ -77,16 +77,12 @@ describe("Signal question reactions", () => { false, ); await expect( - maybeResolveSignalQuestionReaction({ ...params, reactionKey: "4️⃣" }), - ).resolves.toBe(true); - await expect(maybeResolveSignalQuestionReaction(params)).resolves.toBe(true); + maybeResolveSignalQuestionReaction({ ...params, targetAuthor: "+15550009999" }), + ).resolves.toBe(false); await expect(maybeResolveSignalQuestionReaction(params)).resolves.toBe(true); expect(hoisted.resolve).toHaveBeenCalledOnce(); expect(hoisted.resolve).toHaveBeenCalledWith( expect.objectContaining({ questionId, optionValue: "One", senderId: "+15550002222" }), ); - expect(params.logDebug).toHaveBeenCalledWith( - expect.stringContaining("stale question reaction ignored"), - ); }); }); diff --git a/extensions/signal/src/question-reactions.ts b/extensions/signal/src/question-reactions.ts index 73365ee67222..74966ae2e838 100644 --- a/extensions/signal/src/question-reactions.ts +++ b/extensions/signal/src/question-reactions.ts @@ -1,7 +1,10 @@ // Signal transport binding for numbered ask_user reactions. import type { OutboundDeliveryResult } from "openclaw/plugin-sdk/channel-send-result"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { questionGatewayRuntime } from "openclaw/plugin-sdk/question-gateway-runtime"; +import { + createQuestionReactionTargetStore, + questionGatewayRuntime, +} from "openclaw/plugin-sdk/question-gateway-runtime"; import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; import { normalizeAccountId } from "openclaw/plugin-sdk/routing"; import { resolveSignalTarget } from "./aliases.js"; @@ -10,55 +13,32 @@ import { resolveSignalApprovalTargetAuthorKeys, } from "./approval-reactions.js"; -const TARGET_TTL_MS = 24 * 60 * 60 * 1_000; - -type SignalQuestionReactionTarget = { - questionId: string; - optionValues: string[]; - targetAuthorKeys: string[]; - terminal: boolean; - expiresAtMs: number; - cleanupTimer: ReturnType; +type SignalQuestionReactionIdentity = { + accountId: string; + conversationKey: string; + messageId: string; }; -const targets = new Map(); - -function storeTarget( - key: string, - binding: { questionId: string; optionValues: string[] }, - targetAuthorKeys: string[], -): void { - const existing = targets.get(key); - if (existing) { - clearTimeout(existing.cleanupTimer); - } - const target: SignalQuestionReactionTarget = { - ...binding, - targetAuthorKeys, - terminal: false, - expiresAtMs: Date.now() + TARGET_TTL_MS, - cleanupTimer: setTimeout(() => { - if (targets.get(key) === target) { - targets.delete(key); - } - }, TARGET_TTL_MS), - }; - target.cleanupTimer.unref?.(); - targets.set(key, target); - questionGatewayRuntime.registerChannelDelivery({ - questionId: binding.questionId, - deliveryId: `signal-reaction:${key}`, - finalize: () => { - target.terminal = true; - }, - }); -} - -function buildKey(accountId: string, conversationKey: string, messageId: string): string | null { - const values = [accountId, conversationKey, messageId].map((value) => value.trim()); +function buildKey(identity: SignalQuestionReactionIdentity): string | null { + const values = [identity.accountId, identity.conversationKey, identity.messageId].map((value) => + value.trim(), + ); return values.every(Boolean) ? values.join(":") : null; } +const questionReactionTargets = createQuestionReactionTargetStore< + SignalQuestionReactionIdentity, + string[] +>({ + channel: "signal", + channelDisplayName: "Signal", + buildKey, + identityMatches: (stored, incoming) => + Boolean(stored && incoming?.some((authorKey) => stored.includes(authorKey))), + registerChannelDelivery: questionGatewayRuntime.registerChannelDelivery, + resolveReaction: questionGatewayRuntime.resolveReaction, +}); + function resolveConversationKey(params: { cfg: OpenClawConfig; accountId?: string | null; @@ -98,13 +78,15 @@ export function registerSignalQuestionReactionTargetForDeliveredPayload(params: let registered = false; for (const result of params.results) { const messageId = result.channel === "signal" ? result.messageId.trim() : ""; - const key = - messageId && messageId !== "unknown" ? buildKey(accountId, conversationKey, messageId) : null; - if (!key) { + if (!messageId || messageId === "unknown") { continue; } - storeTarget(key, binding, targetAuthorKeys); - registered = true; + registered = + questionReactionTargets.register( + binding, + { accountId, conversationKey, messageId }, + targetAuthorKeys, + ) || registered; } return registered; } @@ -126,43 +108,23 @@ export async function maybeResolveSignalQuestionReaction(params: { return false; } const optionIndex = questionGatewayRuntime.resolveReactionIndex(params.reactionKey); - const key = buildKey(params.accountId, params.conversationKey, params.messageId); - if (optionIndex === undefined || !key) { - return false; - } - const target = targets.get(key); - if (!target) { + if (optionIndex === undefined) { return false; } const authorKeys = resolveSignalApprovalTargetAuthorKeys(params); - if (!authorKeys.some((authorKey) => target.targetAuthorKeys.includes(authorKey))) { - return false; - } - if (target.expiresAtMs <= Date.now() || target.terminal) { - target.terminal = true; - params.logDebug?.(`signal: stale question reaction ignored id=${target.questionId}`); - return true; - } - const optionValue = target.optionValues[optionIndex]; - if (!optionValue) { - params.logDebug?.(`signal: out-of-range question reaction ignored id=${target.questionId}`); - return true; - } - try { - const result = await questionGatewayRuntime.resolveReaction({ - cfg: params.cfg, - questionId: target.questionId, - optionValue, - senderId: params.actorId, - gatewayUrl: params.gatewayUrl, - clientDisplayName: `Signal question (${params.actorId})`, - }); - target.terminal = result?.status === "answered" || result?.status === "already-terminal"; - if (result?.status === "already-terminal") { - params.logDebug?.(`signal: stale question reaction ignored id=${target.questionId}`); - } - } catch (error) { - params.logDebug?.(`signal: question reaction failed id=${target.questionId}: ${String(error)}`); - } - return true; + return await questionReactionTargets.resolve({ + identities: [ + { + accountId: params.accountId, + conversationKey: params.conversationKey, + messageId: params.messageId, + }, + ], + optionIndex, + cfg: params.cfg, + senderId: params.actorId, + gatewayUrl: params.gatewayUrl, + metadata: authorKeys, + logDebug: params.logDebug, + }); } diff --git a/extensions/slack/src/monitor/message-handler/preflight-audio.runtime.ts b/extensions/slack/src/monitor/message-handler/preflight-audio.runtime.ts deleted file mode 100644 index 3d4e6e78fa9d..000000000000 --- a/extensions/slack/src/monitor/message-handler/preflight-audio.runtime.ts +++ /dev/null @@ -1,3 +0,0 @@ -// Slack plugin module implements audio preflight runtime behavior. -export { sendDurableMessageBatch } from "openclaw/plugin-sdk/channel-outbound"; -export { transcribeFirstAudio } from "openclaw/plugin-sdk/media-runtime"; diff --git a/extensions/slack/src/monitor/message-handler/preflight-audio.test.ts b/extensions/slack/src/monitor/message-handler/preflight-audio.test.ts index 79e8fd3248a2..54655e683fe0 100644 --- a/extensions/slack/src/monitor/message-handler/preflight-audio.test.ts +++ b/extensions/slack/src/monitor/message-handler/preflight-audio.test.ts @@ -10,18 +10,26 @@ import { findCaptionlessSlackAudioFile, formatSlackAudioTranscriptForAgent, resolveSlackPreflightAudioTranscript, - sendSlackPreflightAudioTranscriptEcho, } from "./preflight-audio.js"; -const { sendDurableMessageBatchMock, transcribeFirstAudioMock } = vi.hoisted(() => ({ - sendDurableMessageBatchMock: vi.fn(), +const { transcribeFirstAudioMock } = vi.hoisted(() => ({ transcribeFirstAudioMock: vi.fn(), })); -vi.mock("./preflight-audio.runtime.js", () => ({ - sendDurableMessageBatch: sendDurableMessageBatchMock, - transcribeFirstAudio: transcribeFirstAudioMock, -})); +vi.mock("openclaw/plugin-sdk/media-understanding-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + createChannelPreflightAudio: ( + params: Parameters[0], + ) => + actual.createChannelPreflightAudio({ + ...params, + transcribeFirstAudio: transcribeFirstAudioMock, + }), + }; +}); function createSlackMessage(overrides: Partial): SlackMessageEvent { return { @@ -35,24 +43,8 @@ function createSlackMessage(overrides: Partial): SlackMessage } as SlackMessageEvent; } -function createAudioConfig(overrides: Record = {}): OpenClawConfig { - return { - tools: { - media: { - audio: { - enabled: true, - echoTranscript: true, - ...overrides, - }, - }, - }, - } as OpenClawConfig; -} - describe("Slack captionless audio preflight", () => { beforeEach(() => { - sendDurableMessageBatchMock.mockReset(); - sendDurableMessageBatchMock.mockResolvedValue({ status: "sent", messageIds: ["1"] }); transcribeFirstAudioMock.mockReset(); }); @@ -91,9 +83,9 @@ describe("Slack captionless audio preflight", () => { ); }); - it("transcribes the first audio attachment once and suppresses speculative echo", async () => { + it("transcribes the first audio attachment and returns its ordered media index", async () => { transcribeFirstAudioMock.mockResolvedValue("Bill please review this"); - const cfg = createAudioConfig(); + const cfg = {} as OpenClawConfig; const media: SlackMediaResult[] = [ { path: "/tmp/image.png", contentType: "image/png", placeholder: "[image]" }, { path: "/tmp/voice.mp4", contentType: "audio/mp4", placeholder: "[voice]" }, @@ -120,45 +112,8 @@ describe("Slack captionless audio preflight", () => { MessageThreadId: "1.000", SessionKey: "agent:main:slack:channel:c1", }), - cfg: expect.objectContaining({ - tools: expect.objectContaining({ - media: expect.objectContaining({ - audio: expect.objectContaining({ echoTranscript: false }), - }), - }), - }), + cfg, }); - expect(cfg.tools?.media?.audio?.echoTranscript).toBe(true); - }); - - it("echoes only an admitted transcript and preserves literal replacement tokens", async () => { - await sendSlackPreflightAudioTranscriptEcho({ - transcript: "cost is $& and $1", - cfg: createAudioConfig({ echoFormat: "heard: {transcript}" }), - accountId: "work", - originatingTo: "channel:C1", - messageThreadId: "1.000", - }); - - expect(sendDurableMessageBatchMock).toHaveBeenCalledWith({ - cfg: expect.any(Object), - channel: "slack", - to: "channel:C1", - accountId: "work", - threadId: "1.000", - payloads: [{ text: "heard: cost is $& and $1" }], - bestEffort: true, - durability: "best_effort", - }); - - sendDurableMessageBatchMock.mockClear(); - await sendSlackPreflightAudioTranscriptEcho({ - transcript: "not echoed", - cfg: createAudioConfig({ echoTranscript: false }), - accountId: "work", - originatingTo: "channel:C1", - }); - expect(sendDurableMessageBatchMock).not.toHaveBeenCalled(); }); it("removes preflight downloads when the transcript does not admit the message", async () => { diff --git a/extensions/slack/src/monitor/message-handler/preflight-audio.ts b/extensions/slack/src/monitor/message-handler/preflight-audio.ts index 6f570bbf836a..86d55997867f 100644 --- a/extensions/slack/src/monitor/message-handler/preflight-audio.ts +++ b/extensions/slack/src/monitor/message-handler/preflight-audio.ts @@ -1,18 +1,11 @@ // Slack plugin module implements captionless audio mention preflight behavior. import fs from "node:fs/promises"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; import { mimeTypeFromFilePath } from "openclaw/plugin-sdk/media-mime"; -import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { createChannelPreflightAudio } from "openclaw/plugin-sdk/media-understanding-runtime"; import type { SlackFile, SlackMessageEvent } from "../../types.js"; import { MAX_SLACK_MEDIA_FILES, type SlackMediaResult } from "../media-types.js"; -const SLACK_DEFAULT_ECHO_TRANSCRIPT_FORMAT = '📝 "{transcript}"'; - -const loadSlackPreflightAudioRuntime = createLazyRuntimeModule( - () => import("./preflight-audio.runtime.js"), -); - function isSlackAudioFile(file: SlackFile): boolean { if (file.subtype === "slack_audio") { return true; @@ -24,6 +17,11 @@ function isSlackAudioFile(file: SlackFile): boolean { return Boolean(mimeTypeFromFilePath(file.name)?.startsWith("audio/")); } +const slackPreflightAudio = createChannelPreflightAudio({ + channel: "slack", + isAudio: isSlackAudioFile, +}); + export function findCaptionlessSlackAudioFile(message: SlackMessageEvent): SlackFile | undefined { if (message.text?.trim()) { return undefined; @@ -39,26 +37,6 @@ export function formatSlackAudioTranscriptForAgent(params: { return [framed, params.rawBody].filter(Boolean).join("\n"); } -function suppressSlackPreflightAudioEcho(cfg: OpenClawConfig): OpenClawConfig { - const audio = cfg.tools?.media?.audio; - if (!audio?.echoTranscript) { - return cfg; - } - return { - ...cfg, - tools: { - ...cfg.tools, - media: { - ...cfg.tools?.media, - audio: { - ...audio, - echoTranscript: false, - }, - }, - }, - }; -} - export async function resolveSlackPreflightAudioTranscript(params: { media: readonly SlackMediaResult[]; cfg: OpenClawConfig; @@ -73,9 +51,8 @@ export async function resolveSlackPreflightAudioTranscript(params: { if (mediaIndex < 0) { return null; } - try { - const { transcribeFirstAudio } = await loadSlackPreflightAudioRuntime(); - const transcript = await transcribeFirstAudio({ + const transcript = await slackPreflightAudio.resolve({ + request: { ctx: { media: [...params.media], Provider: "slack", @@ -87,18 +64,10 @@ export async function resolveSlackPreflightAudioTranscript(params: { ChatType: "channel", SessionKey: params.sessionKey, }, - cfg: suppressSlackPreflightAudioEcho(params.cfg), - }); - return transcript ? { transcript, mediaIndex } : null; - } catch (err) { - logVerbose(`slack: audio preflight transcription failed: ${String(err)}`); - return null; - } -} - -function formatSlackAudioTranscriptEcho(transcript: string, format: string): string { - // Function replacement preserves literal `$` sequences in provider output. - return format.replace("{transcript}", () => transcript); + cfg: params.cfg, + }, + }); + return transcript ? { transcript, mediaIndex } : null; } export async function sendSlackPreflightAudioTranscriptEcho(params: { @@ -108,32 +77,7 @@ export async function sendSlackPreflightAudioTranscriptEcho(params: { originatingTo: string; messageThreadId?: string; }): Promise { - const audio = params.cfg.tools?.media?.audio; - if (!audio?.echoTranscript) { - return; - } - const text = formatSlackAudioTranscriptEcho( - params.transcript, - audio.echoFormat ?? SLACK_DEFAULT_ECHO_TRANSCRIPT_FORMAT, - ); - try { - const { sendDurableMessageBatch } = await loadSlackPreflightAudioRuntime(); - const send = await sendDurableMessageBatch({ - cfg: params.cfg, - channel: "slack", - to: params.originatingTo, - accountId: params.accountId, - threadId: params.messageThreadId, - payloads: [{ text }], - bestEffort: true, - durability: "best_effort", - }); - if (send.status === "failed") { - throw send.error; - } - } catch (err) { - logVerbose(`slack: audio transcript echo failed: ${String(err)}`); - } + await slackPreflightAudio.send(params); } export async function discardSlackPreflightMedia( diff --git a/extensions/slack/src/monitor/message-handler/prepare.test.ts b/extensions/slack/src/monitor/message-handler/prepare.test.ts index 73df33f06ac3..3a4ecc7d6f37 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.test.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.test.ts @@ -34,21 +34,32 @@ import { const { enqueueSystemEventMock, logVerboseMock, - sendDurableMessageBatchMock, + sendTranscriptEchoMock, shouldLogVerboseMock, transcribeFirstAudioMock, } = vi.hoisted(() => ({ enqueueSystemEventMock: vi.fn(), logVerboseMock: vi.fn(), - sendDurableMessageBatchMock: vi.fn(), + sendTranscriptEchoMock: vi.fn(), shouldLogVerboseMock: vi.fn(() => false), transcribeFirstAudioMock: vi.fn(), })); -vi.mock("./preflight-audio.runtime.js", () => ({ - sendDurableMessageBatch: sendDurableMessageBatchMock, - transcribeFirstAudio: transcribeFirstAudioMock, -})); +vi.mock("openclaw/plugin-sdk/media-understanding-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + createChannelPreflightAudio: ( + params: Parameters[0], + ) => + actual.createChannelPreflightAudio({ + ...params, + sendTranscriptEcho: sendTranscriptEchoMock, + transcribeFirstAudio: transcribeFirstAudioMock, + }), + }; +}); vi.mock("openclaw/plugin-sdk/runtime-env", async (importOriginal) => { const actual = await importOriginal(); @@ -79,8 +90,8 @@ describe("slack prepareSlackMessage inbound contract", () => { clearSlackThreadParticipationCache(); enqueueSystemEventMock.mockClear(); logVerboseMock.mockClear(); - sendDurableMessageBatchMock.mockReset(); - sendDurableMessageBatchMock.mockResolvedValue({ status: "sent", messageIds: ["1"] }); + sendTranscriptEchoMock.mockReset(); + sendTranscriptEchoMock.mockResolvedValue(undefined); shouldLogVerboseMock.mockReset(); shouldLogVerboseMock.mockReturnValue(false); transcribeFirstAudioMock.mockReset(); diff --git a/extensions/whatsapp/src/question-reactions.test.ts b/extensions/whatsapp/src/question-reactions.test.ts index 420fadbfdfa2..964c7f7a8e1b 100644 --- a/extensions/whatsapp/src/question-reactions.test.ts +++ b/extensions/whatsapp/src/question-reactions.test.ts @@ -16,7 +16,6 @@ vi.mock("openclaw/plugin-sdk/question-gateway-runtime", async (importOriginal) = import { questionGatewayRuntime } from "openclaw/plugin-sdk/question-gateway-runtime"; import { - clearWhatsAppQuestionReactionTargetsForTest, maybeResolveWhatsAppQuestionReaction, registerWhatsAppQuestionReactionTargetForDeliveredPayload, } from "./question-reactions.js"; @@ -44,7 +43,6 @@ function buildPayload() { describe("WhatsApp question reactions", () => { beforeEach(() => { - clearWhatsAppQuestionReactionTargetsForTest(); hoisted.resolve.mockReset().mockResolvedValue({ status: "answered", questionId: "choice", @@ -52,7 +50,7 @@ describe("WhatsApp question reactions", () => { }); }); - it("round-trips a delivered message and silently consumes a stale second tap", async () => { + it("matches receipt identities through reaction-target JID aliases", async () => { const payload = buildPayload(); expect(payload).not.toBeNull(); expect( @@ -60,60 +58,51 @@ describe("WhatsApp question reactions", () => { cfg: {}, target: { channel: "whatsapp", accountId: "default" }, payload: payload!, - results: [{ channel: "whatsapp", messageId: "wa-1", toJid: "1555@s.whatsapp.net" }], + results: [ + { + channel: "whatsapp", + messageId: "summary", + toJid: "group@g.us", + receipt: { + platformMessageIds: ["wa-1"], + sentAt: 1, + parts: [ + { + platformMessageId: "wa-1", + kind: "text", + index: 0, + raw: { messageId: "wa-1", toJid: "1555@s.whatsapp.net" }, + }, + ], + }, + }, + ], }), ).toBe(true); const msg = { - key: { remoteJid: "1555@s.whatsapp.net", participant: "1555@s.whatsapp.net" }, + key: { remoteJid: "group@g.us", participant: "1555@s.whatsapp.net" }, message: { reactionMessage: { text: "2️⃣", - key: { id: "wa-1", remoteJid: "1555@s.whatsapp.net" }, + key: { id: "wa-1", remoteJid: "group@g.us" }, }, }, }; const debug = vi.fn(); - await expect( - maybeResolveWhatsAppQuestionReaction({ - cfg: {}, - accountId: "default", - msg: { - ...msg, - message: { - reactionMessage: { - ...msg.message.reactionMessage, - text: "4️⃣", - }, - }, - }, - senderId: "+1555", - logDebug: debug, - }), - ).resolves.toBe(true); await expect( maybeResolveWhatsAppQuestionReaction({ cfg: {}, accountId: "default", msg, senderId: "+1555", + resolveReactionTargetJids: async () => ["1555@s.whatsapp.net"], logDebug: debug, }), ).resolves.toBe(true); expect(hoisted.resolve).toHaveBeenCalledWith( expect.objectContaining({ questionId, optionValue: "Two", senderId: "+1555" }), ); - - await expect( - maybeResolveWhatsAppQuestionReaction({ - cfg: {}, - accountId: "default", - msg, - senderId: "+1555", - logDebug: debug, - }), - ).resolves.toBe(true); expect(hoisted.resolve).toHaveBeenCalledOnce(); - expect(debug).toHaveBeenCalledWith(expect.stringContaining("stale question reaction ignored")); }); }); diff --git a/extensions/whatsapp/src/question-reactions.ts b/extensions/whatsapp/src/question-reactions.ts index cb32cf00d750..59373e4f49f9 100644 --- a/extensions/whatsapp/src/question-reactions.ts +++ b/extensions/whatsapp/src/question-reactions.ts @@ -2,53 +2,34 @@ import type { WAMessage } from "baileys"; import type { OutboundDeliveryResult } from "openclaw/plugin-sdk/channel-send-result"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { questionGatewayRuntime } from "openclaw/plugin-sdk/question-gateway-runtime"; +import { + createQuestionReactionTargetStore, + questionGatewayRuntime, +} from "openclaw/plugin-sdk/question-gateway-runtime"; import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; import { resolveWhatsAppAccount } from "./accounts.js"; -const TARGET_TTL_MS = 24 * 60 * 60 * 1_000; - -type WhatsAppQuestionReactionTarget = { - questionId: string; - optionValues: string[]; - terminal: boolean; - expiresAtMs: number; - cleanupTimer: ReturnType; +type WhatsAppQuestionReactionIdentity = { + accountId: string; + remoteJid: string; + messageId: string; }; -const targets = new Map(); - -function storeTarget(key: string, binding: { questionId: string; optionValues: string[] }): void { - const existing = targets.get(key); - if (existing) { - clearTimeout(existing.cleanupTimer); - } - const target: WhatsAppQuestionReactionTarget = { - ...binding, - terminal: false, - expiresAtMs: Date.now() + TARGET_TTL_MS, - cleanupTimer: setTimeout(() => { - if (targets.get(key) === target) { - targets.delete(key); - } - }, TARGET_TTL_MS), - }; - target.cleanupTimer.unref?.(); - targets.set(key, target); - questionGatewayRuntime.registerChannelDelivery({ - questionId: binding.questionId, - deliveryId: `whatsapp-reaction:${key}`, - finalize: () => { - target.terminal = true; - }, - }); -} - -function buildKey(accountId: string, remoteJid: string, messageId: string): string | undefined { - const parts = [accountId, remoteJid, messageId].map((part) => part.trim()); +function buildKey(identity: WhatsAppQuestionReactionIdentity): string | undefined { + const parts = [identity.accountId, identity.remoteJid, identity.messageId].map((part) => + part.trim(), + ); return parts.every(Boolean) ? parts.join(":") : undefined; } +const questionReactionTargets = createQuestionReactionTargetStore({ + channel: "whatsapp", + channelDisplayName: "WhatsApp", + buildKey, + registerChannelDelivery: questionGatewayRuntime.registerChannelDelivery, + resolveReaction: questionGatewayRuntime.resolveReaction, +}); + function addCandidate(values: string[], value: string | null | undefined): void { const normalized = value?.trim(); if (normalized && !values.includes(normalized)) { @@ -101,12 +82,8 @@ export function registerWhatsAppQuestionReactionTargetForDeliveredPayload(params }).accountId; let registered = false; for (const identity of listDeliveredIdentities(params.results)) { - const key = buildKey(accountId, identity.remoteJid, identity.messageId); - if (!key) { - continue; - } - storeTarget(key, binding); - registered = true; + registered = + questionReactionTargets.register(binding, { accountId, ...identity }) || registered; } return registered; } @@ -137,57 +114,16 @@ export async function maybeResolveWhatsAppQuestionReaction(params: { addCandidate(candidates, mapped); } } - let matched: { key: string; target: WhatsAppQuestionReactionTarget } | undefined; - for (const remoteJid of candidates) { - const key = buildKey(params.accountId, remoteJid, messageId); - const target = key ? targets.get(key) : undefined; - if (key && target) { - matched = { key, target }; - break; - } - } - if (!matched) { - return false; - } - if (matched.target.expiresAtMs <= Date.now() || matched.target.terminal) { - matched.target.terminal = true; - params.logDebug?.(`whatsapp: stale question reaction ignored id=${matched.target.questionId}`); - return true; - } - const optionValue = matched.target.optionValues[optionIndex]; - if (!optionValue) { - params.logDebug?.( - `whatsapp: out-of-range question reaction ignored id=${matched.target.questionId}`, - ); - return true; - } - try { - const result = await questionGatewayRuntime.resolveReaction({ - cfg: params.cfg, - questionId: matched.target.questionId, - optionValue, - senderId: params.senderId, - gatewayUrl: params.gatewayUrl, - clientDisplayName: `WhatsApp question (${params.senderId})`, - }); - matched.target.terminal = - result?.status === "answered" || result?.status === "already-terminal"; - if (result?.status === "already-terminal") { - params.logDebug?.( - `whatsapp: stale question reaction ignored id=${matched.target.questionId}`, - ); - } - } catch (error) { - params.logDebug?.( - `whatsapp: question reaction failed id=${matched.target.questionId}: ${String(error)}`, - ); - } - return true; -} - -export function clearWhatsAppQuestionReactionTargetsForTest(): void { - for (const target of targets.values()) { - clearTimeout(target.cleanupTimer); - } - targets.clear(); + return await questionReactionTargets.resolve({ + identities: candidates.map((remoteJid) => ({ + accountId: params.accountId, + remoteJid, + messageId, + })), + optionIndex, + cfg: params.cfg, + senderId: params.senderId, + gatewayUrl: params.gatewayUrl, + logDebug: params.logDebug, + }); } diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index 5a340ce158a8..468540227684 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -220,7 +220,8 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { // +3: canonical ready, blocked, and stopped channel lifecycle patch factories. // +1: bounded external-content sanitizer for plugin-owned untrusted projections. // +1: auth-profile preservation decision for native model pickers. - 4831, + // +2: shared channel question-reaction store and preflight-audio factories. + 4833, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( @@ -268,7 +269,8 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { // +3: canonical ready, blocked, and stopped channel lifecycle patch factories. // +1: bounded external-content sanitizer for plugin-owned untrusted projections. // +1: auth-profile preservation decision for native model pickers. - 2908, + // +2: shared channel question-reaction store and preflight-audio factories. + 2910, env, ), publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/media-understanding/echo-transcript.ts b/src/media-understanding/echo-transcript.ts index 56c5355cefef..01b62d76bfdb 100644 --- a/src/media-understanding/echo-transcript.ts +++ b/src/media-understanding/echo-transcript.ts @@ -26,6 +26,8 @@ export async function sendTranscriptEcho(params: { cfg: OpenClawConfig; transcript: string; format?: string; + logSuccess?: boolean; + failureLogPrefix?: string; }): Promise { const { ctx, cfg, transcript } = params; const channel = ctx.Provider ?? ctx.Surface ?? ""; @@ -65,10 +67,11 @@ export async function sendTranscriptEcho(params: { if (send.status === "failed") { throw send.error; } - if (shouldLogVerbose()) { + if ((params.logSuccess ?? true) && shouldLogVerbose()) { logVerbose(`media: echo-transcript sent to ${normalizedChannel}/${to}`); } } catch (err) { - logVerbose(`media: echo-transcript delivery failed: ${String(err)}`); + const prefix = params.failureLogPrefix ?? "media: echo-transcript delivery failed"; + logVerbose(`${prefix}: ${String(err)}`); } } diff --git a/src/plugin-sdk/media-understanding-runtime.test.ts b/src/plugin-sdk/media-understanding-runtime.test.ts new file mode 100644 index 000000000000..c0d68f228c7e --- /dev/null +++ b/src/plugin-sdk/media-understanding-runtime.test.ts @@ -0,0 +1,146 @@ +import { describe, expect, it, vi } from "vitest"; +import { createChannelPreflightAudio } from "./media-understanding-runtime.js"; + +type TranscribeFirstAudio = + typeof import("../media-understanding/audio-preflight.js").transcribeFirstAudio; +type SendTranscriptEcho = + typeof import("../media-understanding/echo-transcript.js").sendTranscriptEcho; + +const audioConfig = { + tools: { + media: { + audio: { + echoTranscript: true, + echoFormat: "heard: {transcript}", + }, + }, + }, +}; + +describe("createChannelPreflightAudio", () => { + it("suppresses speculative echo without mutating the caller config", async () => { + const transcribeFirstAudio = vi.fn().mockResolvedValue("hello"); + const preflight = createChannelPreflightAudio({ + channel: "test", + isAudio: (value: { contentType?: string }) => + value.contentType?.startsWith("audio/") ?? false, + transcribeFirstAudio, + }); + + await expect( + preflight.resolve({ + request: { + ctx: { media: [{ path: "/tmp/voice.ogg", contentType: "audio/ogg" }] }, + cfg: audioConfig, + }, + }), + ).resolves.toBe("hello"); + + expect(transcribeFirstAudio).toHaveBeenCalledWith({ + ctx: { media: [{ path: "/tmp/voice.ogg", contentType: "audio/ogg" }] }, + cfg: { + tools: { media: { audio: { echoTranscript: false, echoFormat: "heard: {transcript}" } } }, + }, + }); + expect(audioConfig.tools.media.audio.echoTranscript).toBe(true); + }); + + it("preserves the original config for channels without deferred echo", async () => { + const transcribeFirstAudio = vi.fn().mockResolvedValue("hello"); + const preflight = createChannelPreflightAudio({ + channel: "test", + isAudio: () => true, + deferTranscriptEcho: false, + transcribeFirstAudio, + }); + + await preflight.resolve({ request: { ctx: { media: [] }, cfg: audioConfig } }); + + expect(transcribeFirstAudio).toHaveBeenCalledWith({ ctx: { media: [] }, cfg: audioConfig }); + }); + + it("checks abort state before and after transcription", async () => { + const before = new AbortController(); + before.abort(); + const transcribeFirstAudio = vi.fn(); + const preflight = createChannelPreflightAudio({ + channel: "test", + isAudio: () => true, + transcribeFirstAudio, + }); + + await expect( + preflight.resolve({ + request: { ctx: { media: [] }, cfg: {} }, + abortSignal: before.signal, + }), + ).resolves.toBeUndefined(); + expect(transcribeFirstAudio).not.toHaveBeenCalled(); + + const after = new AbortController(); + transcribeFirstAudio.mockImplementation(async () => { + after.abort(); + return "too late"; + }); + await expect( + preflight.resolve({ + request: { ctx: { media: [] }, cfg: {} }, + abortSignal: after.signal, + }), + ).resolves.toBeUndefined(); + }); + + it("formats and sends admitted echoes while preserving literal replacement tokens", async () => { + const sendTranscriptEcho = vi.fn().mockResolvedValue(undefined); + const preflight = createChannelPreflightAudio({ + channel: "test", + isAudio: () => true, + sendTranscriptEcho, + }); + + expect(preflight.format("cost is $& and $1", "heard: {transcript}")).toBe( + "heard: cost is $& and $1", + ); + await preflight.send({ + transcript: "cost is $& and $1", + cfg: audioConfig, + accountId: "work", + originatingTo: "channel:C1", + messageThreadId: "thread-1", + }); + + expect(sendTranscriptEcho).toHaveBeenCalledWith({ + ctx: { + Provider: "test", + Surface: "test", + OriginatingChannel: "test", + OriginatingTo: "channel:C1", + AccountId: "work", + MessageThreadId: "thread-1", + }, + cfg: audioConfig, + transcript: "cost is $& and $1", + format: "heard: {transcript}", + logSuccess: false, + failureLogPrefix: "test: audio transcript echo failed", + }); + }); + + it("does not send when echo is disabled", async () => { + const sendTranscriptEcho = vi.fn(); + const preflight = createChannelPreflightAudio({ + channel: "test", + isAudio: () => true, + sendTranscriptEcho, + }); + + await preflight.send({ + transcript: "hello", + cfg: {}, + accountId: "work", + originatingTo: "channel:C1", + }); + + expect(sendTranscriptEcho).not.toHaveBeenCalled(); + }); +}); diff --git a/src/plugin-sdk/media-understanding-runtime.ts b/src/plugin-sdk/media-understanding-runtime.ts index e82a36d9ff2b..f5e785c3f8e9 100644 --- a/src/plugin-sdk/media-understanding-runtime.ts +++ b/src/plugin-sdk/media-understanding-runtime.ts @@ -1,6 +1,117 @@ /** * Runtime SDK subpath for media understanding, image description, and audio transcription. */ +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { logVerbose } from "../globals.js"; +import { sendTranscriptEcho } from "../media-understanding/echo-transcript.js"; +import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; + +type TranscribeFirstAudio = + typeof import("../media-understanding/audio-preflight.js").transcribeFirstAudio; +type SendTranscriptEcho = typeof sendTranscriptEcho; + +const DEFAULT_ECHO_TRANSCRIPT_FORMAT = '📝 "{transcript}"'; +const loadAudioPreflightRuntime = createLazyRuntimeModule( + () => import("../media-understanding/audio-preflight.js"), +); + +/** Creates shared preflight transcription and deferred-echo behavior for a channel. */ +export function createChannelPreflightAudio(params: { + channel: string; + isAudio: (value: TAudio) => boolean; + deferTranscriptEcho?: boolean; + transcribeFirstAudio?: TranscribeFirstAudio; + sendTranscriptEcho?: SendTranscriptEcho; +}) { + const deferTranscriptEcho = params.deferTranscriptEcho ?? true; + + const suppress = (cfg: OpenClawConfig): OpenClawConfig => { + if (!deferTranscriptEcho) { + return cfg; + } + const audio = cfg.tools?.media?.audio; + if (!audio?.echoTranscript) { + return cfg; + } + return { + ...cfg, + tools: { + ...cfg.tools, + media: { + ...cfg.tools?.media, + audio: { + ...audio, + echoTranscript: false, + }, + }, + }, + }; + }; + + const format = (transcript: string, formatTemplate: string): string => { + // Function replacement preserves literal `$` sequences in provider output. + return formatTemplate.replace("{transcript}", () => transcript); + }; + + return { + isAudio: params.isAudio, + suppress, + format, + + async resolve(resolveParams: { + request: Parameters[0]; + abortSignal?: AbortSignal; + }): Promise { + if (resolveParams.abortSignal?.aborted) { + return undefined; + } + try { + const transcribeFirstAudio = + params.transcribeFirstAudio ?? (await loadAudioPreflightRuntime()).transcribeFirstAudio; + if (resolveParams.abortSignal?.aborted) { + return undefined; + } + const transcript = await transcribeFirstAudio({ + ...resolveParams.request, + cfg: suppress(resolveParams.request.cfg), + }); + return resolveParams.abortSignal?.aborted ? undefined : transcript; + } catch (err) { + logVerbose(`${params.channel}: audio preflight transcription failed: ${String(err)}`); + return undefined; + } + }, + + async send(sendParams: { + transcript: string; + cfg: OpenClawConfig; + accountId: string; + originatingTo: string; + messageThreadId?: string; + }): Promise { + const audio = sendParams.cfg.tools?.media?.audio; + if (!audio?.echoTranscript) { + return; + } + await (params.sendTranscriptEcho ?? sendTranscriptEcho)({ + ctx: { + Provider: params.channel, + Surface: params.channel, + OriginatingChannel: params.channel, + OriginatingTo: sendParams.originatingTo, + AccountId: sendParams.accountId, + MessageThreadId: sendParams.messageThreadId, + }, + cfg: sendParams.cfg, + transcript: sendParams.transcript, + format: audio.echoFormat ?? DEFAULT_ECHO_TRANSCRIPT_FORMAT, + logSuccess: false, + failureLogPrefix: `${params.channel}: audio transcript echo failed`, + }); + }, + }; +} + export { describeImageFile, describeImageFileWithModel, diff --git a/src/plugin-sdk/question-gateway-runtime.test.ts b/src/plugin-sdk/question-gateway-runtime.test.ts new file mode 100644 index 000000000000..5a4cbb272c1e --- /dev/null +++ b/src/plugin-sdk/question-gateway-runtime.test.ts @@ -0,0 +1,164 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createQuestionReactionTargetStore } from "./question-gateway-runtime.js"; + +type RegisterChannelDelivery = + typeof import("../infra/question-channel-runtime.js").registerQuestionChannelDelivery; +type ResolveReaction = + typeof import("../infra/question-reaction-runtime.js").resolveQuestionReactionOverGateway; + +const binding = { + questionId: "ask_0123456789abcdef0123456789abcdef", + optionValues: ["One", "Two"], +}; + +function createStore( + overrides: { + ttlMs?: number; + registerChannelDelivery?: RegisterChannelDelivery; + resolveReaction?: ResolveReaction; + } = {}, +) { + return createQuestionReactionTargetStore({ + channel: "test", + channelDisplayName: "Test", + ttlMs: overrides.ttlMs, + buildKey: (identity: { accountId: string; messageId: string }) => { + const accountId = identity.accountId.trim(); + const messageId = identity.messageId.trim(); + return accountId && messageId ? `${accountId}:${messageId}` : null; + }, + registerChannelDelivery: overrides.registerChannelDelivery, + resolveReaction: overrides.resolveReaction, + }); +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("createQuestionReactionTargetStore", () => { + it("replaces target timers without letting the old timer delete the new target", () => { + vi.useFakeTimers(); + const store = createStore({ + ttlMs: 1_000, + registerChannelDelivery: vi.fn(), + }); + const identity = { accountId: "default", messageId: "message-1" }; + + expect(store.register(binding, identity)).toBe(true); + vi.advanceTimersByTime(500); + expect(store.register(binding, identity)).toBe(true); + vi.advanceTimersByTime(500); + expect(store.has([identity])).toBe(true); + vi.advanceTimersByTime(500); + expect(store.has([identity])).toBe(false); + }); + + it("registers the exact delivery id and honors synchronous finalization", async () => { + const registerChannelDelivery = vi.fn((params) => { + void params.finalize("answered"); + }); + const resolveReaction = vi.fn(); + const store = createStore({ registerChannelDelivery, resolveReaction }); + const identity = { accountId: "default", messageId: "message-1" }; + const logDebug = vi.fn(); + + expect(store.register(binding, identity)).toBe(true); + expect(registerChannelDelivery).toHaveBeenCalledWith( + expect.objectContaining({ + questionId: binding.questionId, + deliveryId: "test-reaction:default:message-1", + }), + ); + await expect( + store.resolve({ + identities: [identity], + optionIndex: 0, + cfg: {}, + senderId: "sender-1", + logDebug, + }), + ).resolves.toBe(true); + expect(resolveReaction).not.toHaveBeenCalled(); + expect(logDebug).toHaveBeenCalledWith( + `test: stale question reaction ignored id=${binding.questionId}`, + ); + }); + + it("resolves once, terminalizes the target, and consumes later duplicates", async () => { + const resolveReaction = vi.fn().mockResolvedValue({ + status: "answered", + questionId: "choice", + optionValue: "Two", + }); + const store = createStore({ + registerChannelDelivery: vi.fn(), + resolveReaction, + }); + const identity = { accountId: "default", messageId: "message-1" }; + store.register(binding, identity); + + const resolveParams = { + identities: [identity], + optionIndex: 1, + cfg: {}, + senderId: "sender-1", + gatewayUrl: "ws://127.0.0.1:1234", + logDebug: vi.fn(), + }; + await expect(store.resolve(resolveParams)).resolves.toBe(true); + await expect(store.resolve(resolveParams)).resolves.toBe(true); + + expect(resolveReaction).toHaveBeenCalledOnce(); + expect(resolveReaction).toHaveBeenCalledWith({ + cfg: {}, + questionId: binding.questionId, + optionValue: "Two", + senderId: "sender-1", + gatewayUrl: "ws://127.0.0.1:1234", + clientDisplayName: "Test question (sender-1)", + }); + expect(resolveParams.logDebug).toHaveBeenCalledWith( + `test: stale question reaction ignored id=${binding.questionId}`, + ); + }); + + it("consumes out-of-range choices and resolver failures without losing diagnostics", async () => { + const resolveReaction = vi.fn().mockRejectedValue(new Error("gateway down")); + const store = createStore({ + registerChannelDelivery: vi.fn(), + resolveReaction, + }); + const outOfRange = { accountId: "default", messageId: "out-of-range" }; + const failure = { accountId: "default", messageId: "failure" }; + const logDebug = vi.fn(); + store.register(binding, outOfRange); + store.register(binding, failure); + + await expect( + store.resolve({ + identities: [outOfRange], + optionIndex: 3, + cfg: {}, + senderId: "sender-1", + logDebug, + }), + ).resolves.toBe(true); + await expect( + store.resolve({ + identities: [failure], + optionIndex: 0, + cfg: {}, + senderId: "sender-1", + logDebug, + }), + ).resolves.toBe(true); + + expect(logDebug).toHaveBeenCalledWith( + `test: out-of-range question reaction ignored id=${binding.questionId}`, + ); + expect(logDebug).toHaveBeenCalledWith( + `test: question reaction failed id=${binding.questionId}: Error: gateway down`, + ); + }); +}); diff --git a/src/plugin-sdk/question-gateway-runtime.ts b/src/plugin-sdk/question-gateway-runtime.ts index 392c3bdce358..162db940bc0a 100644 --- a/src/plugin-sdk/question-gateway-runtime.ts +++ b/src/plugin-sdk/question-gateway-runtime.ts @@ -12,6 +12,140 @@ import { resolveQuestionReactionOverGateway, } from "../infra/question-reaction-runtime.js"; +const DEFAULT_QUESTION_REACTION_TARGET_TTL_MS = 24 * 60 * 60 * 1_000; + +/** Creates one channel-owned target store for numbered ask_user reactions. */ +export function createQuestionReactionTargetStore(params: { + channel: string; + channelDisplayName: string; + ttlMs?: number; + buildKey: (identity: TIdentity) => string | null | undefined; + identityMatches?: (stored: TMetadata | undefined, incoming: TMetadata | undefined) => boolean; + registerChannelDelivery?: typeof registerQuestionChannelDelivery; + resolveReaction?: typeof resolveQuestionReactionOverGateway; +}) { + type Target = { + questionId: string; + optionValues: string[]; + metadata?: TMetadata; + terminal: boolean; + expiresAtMs: number; + cleanupTimer: ReturnType; + }; + + const ttlMs = params.ttlMs ?? DEFAULT_QUESTION_REACTION_TARGET_TTL_MS; + const registerChannelDelivery = params.registerChannelDelivery ?? registerQuestionChannelDelivery; + const resolveReaction = params.resolveReaction ?? resolveQuestionReactionOverGateway; + const targets = new Map(); + + const findTarget = (identities: readonly TIdentity[]): Target | undefined => { + for (const identity of identities) { + const key = params.buildKey(identity); + const target = key ? targets.get(key) : undefined; + if (target) { + return target; + } + } + return undefined; + }; + + return { + register( + binding: { questionId: string; optionValues: string[] }, + identity: TIdentity, + metadata?: TMetadata, + ): boolean { + const key = params.buildKey(identity); + if (!key) { + return false; + } + const existing = targets.get(key); + if (existing) { + clearTimeout(existing.cleanupTimer); + } + const target: Target = { + ...binding, + metadata, + terminal: false, + expiresAtMs: Date.now() + ttlMs, + cleanupTimer: setTimeout(() => { + if (targets.get(key) === target) { + targets.delete(key); + } + }, ttlMs), + }; + target.cleanupTimer.unref?.(); + targets.set(key, target); + // Registration can synchronously finalize an already-terminal question. + registerChannelDelivery({ + questionId: binding.questionId, + deliveryId: `${params.channel}-reaction:${key}`, + finalize: () => { + target.terminal = true; + }, + }); + return true; + }, + + has(identities: readonly TIdentity[]): boolean { + return Boolean(findTarget(identities)); + }, + + async resolve(resolveParams: { + identities: readonly TIdentity[]; + optionIndex: number; + cfg: Parameters[0]["cfg"]; + senderId: string; + gatewayUrl?: string; + metadata?: TMetadata; + logDebug?: (message: string) => void; + }): Promise { + const target = findTarget(resolveParams.identities); + if ( + !target || + (params.identityMatches && !params.identityMatches(target.metadata, resolveParams.metadata)) + ) { + return false; + } + if (target.expiresAtMs <= Date.now() || target.terminal) { + target.terminal = true; + resolveParams.logDebug?.( + `${params.channel}: stale question reaction ignored id=${target.questionId}`, + ); + return true; + } + const optionValue = target.optionValues[resolveParams.optionIndex]; + if (!optionValue) { + resolveParams.logDebug?.( + `${params.channel}: out-of-range question reaction ignored id=${target.questionId}`, + ); + return true; + } + try { + const result = await resolveReaction({ + cfg: resolveParams.cfg, + questionId: target.questionId, + optionValue, + senderId: resolveParams.senderId, + gatewayUrl: resolveParams.gatewayUrl, + clientDisplayName: `${params.channelDisplayName} question (${resolveParams.senderId})`, + }); + target.terminal = result?.status === "answered" || result?.status === "already-terminal"; + if (result?.status === "already-terminal") { + resolveParams.logDebug?.( + `${params.channel}: stale question reaction ignored id=${target.questionId}`, + ); + } + } catch (error) { + resolveParams.logDebug?.( + `${params.channel}: question reaction failed id=${target.questionId}: ${String(error)}`, + ); + } + return true; + }, + }; +} + export const questionGatewayRuntime = { resolveOption: resolveQuestionOverGateway, reactionEmojis: QUESTION_REACTION_EMOJIS,