From 3234a2e8a8c72ac98a1a35c8fcc7580aa3715564 Mon Sep 17 00:00:00 2001 From: sadfun Date: Mon, 21 Sep 2026 21:36:07 +0200 Subject: [PATCH] Add messenger reactions through Wirebot MCP --- README.md | 21 ++ bun.lock | 8 + docs/discord.md | 1 + docs/slack.md | 3 + package.json | 2 + scripts/bun-globals.d.ts | 9 + src/channels/discord/channel.ts | 6 + src/channels/discord/reply.ts | 5 + src/channels/draft-stream.ts | 19 +- src/channels/reactions.ts | 107 ++++++++++ src/channels/slack/channel.ts | 5 + src/channels/slack/reply.ts | 5 + src/channels/telegram/channel.ts | 3 + src/codex/rpc.ts | 20 +- src/codex/service.ts | 67 +++++- src/codex/thread-session.ts | 12 +- src/core/bridge.ts | 1 + src/core/channel.ts | 2 + src/index.ts | 15 +- src/mcp/server.ts | 88 ++++++++ test/draft-stream.test.ts | 54 ++++- test/mcp-approvals.test.ts | 198 +++++++++++++++++ test/reactions.test.ts | 350 +++++++++++++++++++++++++++++++ 23 files changed, 981 insertions(+), 20 deletions(-) create mode 100644 src/channels/reactions.ts create mode 100644 src/mcp/server.ts create mode 100644 test/mcp-approvals.test.ts create mode 100644 test/reactions.test.ts diff --git a/README.md b/README.md index a47a29c..d809767 100644 --- a/README.md +++ b/README.md @@ -258,6 +258,27 @@ The runtime card in the Mini App shows the current outcome and offers **Apply ch If the app-server exits on its own — for example the kernel OOM-killer or a stray `kill -9` takes it out — Wirebot relaunches it automatically, retrying five times over roughly 50 seconds while new turns wait. Only when those attempts fail does the runtime stay degraded and ask for a manual `/restart`. +### Message reactions + +Wirebot automatically connects a private localhost MCP server using MCP 2026-07-28. +Its `react_to_last_message(reaction)` tool targets the message the active turn is answering. +Routing uses Codex's thread/turn metadata; no message identifiers are added to prompts. +The default messenger context encourages Codex to react when it fits the conversation. +Empty answers send no reply, so a reaction can stand on its own; Slack and Discord progress +messages are deleted when there is no final text. +Wirebot preapproves all tools from its own MCP server. Other MCP tool approvals use +the messenger's Allow once, Allow for session (when supported), and Deny buttons. +Queued messages cannot retarget a running turn. Scheduled runs and calls after a turn +ends have no reaction target. + +Reactions add the bot's emoji on Slack and Discord, and replace its single reaction on Telegram. +Unsupported emoji and missing permissions return a tool error without substituting another emoji. +Standard shortcodes use the bundled Emoji Mart Unicode 15 names and aliases (for example +`:thumbsup:`, `:heart:`, and `:thumbsup::skin-tone-4:`); Unicode emoji can also be passed directly. +Custom reactions accept Slack workspace `:names:`, Discord server `:names:` or `<:name:id>` +(`` for animated emoji), and Telegram `:custom_emoji_123456789:` when that chat permits it. +Slack requires `reactions:write`; Discord requires Add Reactions and Read Message History. + ## Source development Requirements: [Bun](https://bun.com) 1.4 or newer. diff --git a/bun.lock b/bun.lock index 06b7c03..4d65c4e 100644 --- a/bun.lock +++ b/bun.lock @@ -5,7 +5,9 @@ "": { "name": "wirebot", "dependencies": { + "@emoji-mart/data": "^1.2.1", "@grammyjs/runner": "^2.0.3", + "@modelcontextprotocol/server": "^2.0.0", "@radix-ui/react-switch": "^1.2.6", "@slack/socket-mode": "^3.0.0", "@slack/web-api": "^8.0.0", @@ -67,6 +69,8 @@ "@discordjs/ws": ["@discordjs/ws@1.2.3", "", { "dependencies": { "@discordjs/collection": "^2.1.0", "@discordjs/rest": "^2.5.1", "@discordjs/util": "^1.1.0", "@sapphire/async-queue": "^1.5.2", "@types/ws": "^8.5.10", "@vladfrangu/async_event_emitter": "^2.2.4", "discord-api-types": "^0.38.1", "tslib": "^2.6.2", "ws": "^8.17.0" } }, "sha512-wPlQDxEmlDg5IxhJPuxXr3Vy9AjYq5xCvFWGJyD7w7Np8ZGu+Mc+97LCoEc/+AYCo2IDpKioiH0/c/mj5ZR9Uw=="], + "@emoji-mart/data": ["@emoji-mart/data@1.2.1", "", {}, "sha512-no2pQMWiBy6gpBEiqGeU77/bFejDqUTRY7KX+0+iur13op3bqUsXdnwoZs6Xb1zbv0gAj5VvS1PWoUUckSr5Dw=="], + "@grammyjs/runner": ["@grammyjs/runner@2.0.3", "", { "dependencies": { "abort-controller": "^3.0.0" }, "peerDependencies": { "grammy": "^1.13.1" } }, "sha512-nckmTs1dPWfVQteK9cxqxzE+0m1VRvluLWB8UgFzsjg62w3qthPJt0TYtJBEdG7OedvfQq4vnFAyE6iaMkR42A=="], "@grammyjs/types": ["@grammyjs/types@4.0.0", "", {}, "sha512-Z8lDLTvOlo12e5Vnly/vQh3JC9ppaitS1dGZ3w068gNitOd/y8tTSiib+Xm38aBGbtYGmUZwOc6afYrLs2CSTg=="], @@ -81,6 +85,10 @@ "@jridgewell/trace-mapping": ["@jridgewell/trace-mapping@0.3.31", "", { "dependencies": { "@jridgewell/resolve-uri": "^3.1.0", "@jridgewell/sourcemap-codec": "^1.4.14" } }, "sha512-zzNR+SdQSDJzc8joaeP8QQoCQr8NuYx2dIIytl1QeBEZHJ9uW6hebsrYgbz8hJwUQao3TWCMtmfV8Nu1twOLAw=="], + "@modelcontextprotocol/core": ["@modelcontextprotocol/core@2.0.0", "", { "dependencies": { "zod": "^4.2.0" } }, "sha512-pJCEwGG7Lfr/+PQp9ZTwKXNeO5wzbfKL7H3MYpCorM4oFBoQrdjnBgEoqG+RjhsvS1FKrDbKux+M1HhlnGWqcA=="], + + "@modelcontextprotocol/server": ["@modelcontextprotocol/server@2.0.0", "", { "dependencies": { "@modelcontextprotocol/core": "2.0.0", "zod": "^4.2.0" } }, "sha512-YhHWdHfpFMQfd0prsEnxKeS3Qz3ytIGmsS0sth4KDjnacIT7hxk6hXHkJ9KysxlkvTM+WZAtQbbcUhdoP4Hvtw=="], + "@parcel/watcher": ["@parcel/watcher@2.5.1", "", { "dependencies": { "detect-libc": "^1.0.3", "is-glob": "^4.0.3", "micromatch": "^4.0.5", "node-addon-api": "^7.0.0" }, "optionalDependencies": { "@parcel/watcher-android-arm64": "2.5.1", "@parcel/watcher-darwin-arm64": "2.5.1", "@parcel/watcher-darwin-x64": "2.5.1", "@parcel/watcher-freebsd-x64": "2.5.1", "@parcel/watcher-linux-arm-glibc": "2.5.1", "@parcel/watcher-linux-arm-musl": "2.5.1", "@parcel/watcher-linux-arm64-glibc": "2.5.1", "@parcel/watcher-linux-arm64-musl": "2.5.1", "@parcel/watcher-linux-x64-glibc": "2.5.1", "@parcel/watcher-linux-x64-musl": "2.5.1", "@parcel/watcher-win32-arm64": "2.5.1", "@parcel/watcher-win32-ia32": "2.5.1", "@parcel/watcher-win32-x64": "2.5.1" } }, "sha512-dfUnCxiN9H4ap84DvD2ubjw+3vUNpstxa0TneY/Paat8a3R4uQZDLSvWjmznAY/DoahqTHl9V46HF/Zs3F29pg=="], "@parcel/watcher-android-arm64": ["@parcel/watcher-android-arm64@2.5.1", "", { "os": "android", "cpu": "arm64" }, "sha512-KF8+j9nNbUN8vzOFDpRMsaKBHZ/mcjEjMToVMJOhTozkDonQFFrRcfdLWn6yWKCmJKmdVxSgHiYvTCef4/qcBA=="], diff --git a/docs/discord.md b/docs/discord.md index fb8947a..de0f92a 100644 --- a/docs/discord.md +++ b/docs/discord.md @@ -25,6 +25,7 @@ In **OAuth2 → URL Generator** (or the current **Installation** page), select t - Send Messages in Threads - Create Public Threads - Read Message History +- Add Reactions Open the generated URL and install the app in the server. Wirebot starts a public thread for a new mention in a normal text channel. If it cannot create or send in that thread, it explains the diff --git a/docs/slack.md b/docs/slack.md index fc2cf71..2c3b159 100644 --- a/docs/slack.md +++ b/docs/slack.md @@ -64,6 +64,7 @@ oauth_config: scopes: bot: - chat:write + - reactions:write - im:history - channels:history - groups:history @@ -88,6 +89,8 @@ settings: ## 2. Collect the two tokens +For an existing installation, add `reactions:write` and reinstall the app to enable reactions. + - **App-level token** (`xapp-…`): in the app's **Basic Information** page, scroll to **App-Level Tokens**, click **Generate Token and Scopes**, name it (for example `wirebot-socket`), add the `connections:write` scope, and diff --git a/package.json b/package.json index 0e16d4b..87d0f25 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,9 @@ "codex:update": "bun src/cli/main.ts codex check --apply" }, "dependencies": { + "@emoji-mart/data": "^1.2.1", "@grammyjs/runner": "^2.0.3", + "@modelcontextprotocol/server": "^2.0.0", "@radix-ui/react-switch": "^1.2.6", "@slack/socket-mode": "^3.0.0", "@slack/web-api": "^8.0.0", diff --git a/scripts/bun-globals.d.ts b/scripts/bun-globals.d.ts index 2109f15..7ba83df 100644 --- a/scripts/bun-globals.d.ts +++ b/scripts/bun-globals.d.ts @@ -4,6 +4,15 @@ * skipLibCheck: false, so only what is needed is declared here. */ declare const Bun: { + serve(options: { + hostname: string; + port: number; + maxRequestBodySize: number; + fetch(request: Request): Response | Promise; + }): { + readonly port: number; + stop(closeActiveConnections?: boolean): Promise; + }; build(config: { entrypoints: readonly string[]; outdir: string; diff --git a/src/channels/discord/channel.ts b/src/channels/discord/channel.ts index f894d0a..8790b3a 100644 --- a/src/channels/discord/channel.ts +++ b/src/channels/discord/channel.ts @@ -34,6 +34,7 @@ import { errorMessage } from "../../shared/errors.js"; import type { Logger } from "../../shared/logger.js"; import { PendingChoices } from "../choices.js"; import { decodeCommandAction } from "../command-actions.js"; +import { reactToDiscordMessage } from "../reactions.js"; import { type CodexConfigAccess, DiscordConfigUi, discordConfigActionPrefix } from "./config-ui.js"; import { discordDisplayName, @@ -254,6 +255,7 @@ export class DiscordChannel implements MessagingChannel { deliveryTarget: discordDeliveryTarget(replyRoute.channelId), }, reference: discordMessageReference(message.channelId, message.id), + react: (reaction) => reactToDiscordMessage(message, reaction), ...(message.reference?.messageId === undefined ? {} : { @@ -770,6 +772,10 @@ function discordMessagingApi(client: Client): DiscordMessagingApi { ...(options.components === undefined ? {} : { components: options.components }), }); }, + deleteMessage: async (channelId, messageId) => { + const target = await channel(channelId); + await target.messages.delete(messageId); + }, sendTyping: async (channelId) => { const target = await channel(channelId); await target.sendTyping(); diff --git a/src/channels/discord/reply.ts b/src/channels/discord/reply.ts index 311950d..0ce40d8 100644 --- a/src/channels/discord/reply.ts +++ b/src/channels/discord/reply.ts @@ -42,6 +42,7 @@ export interface DiscordUpdateOptions { export interface DiscordMessagingApi { postMessage(options: DiscordPostOptions): Promise; updateMessage(options: DiscordUpdateOptions): Promise; + deleteMessage(channelId: string, messageId: string): Promise; sendTyping(channelId: string): Promise; } @@ -199,6 +200,10 @@ export class DiscordReplyStream extends DraftReplyStream { await this.#api.updateMessage({ channelId: this.#channelId, messageId, content }); } + protected async remove(messageId: string): Promise { + await this.#api.deleteMessage(this.#channelId, messageId); + } + protected renderProgress(block: string): string { return block; } diff --git a/src/channels/draft-stream.ts b/src/channels/draft-stream.ts index 96f860a..47421e9 100644 --- a/src/channels/draft-stream.ts +++ b/src/channels/draft-stream.ts @@ -9,7 +9,7 @@ import { formatThinkingBlock, splitMessageText } from "./progress.js"; /** * Channel-agnostic streaming reply: one editable message that starts as * progress and is replaced by the final answer. Edits are throttled to one - * per interval. Connectors supply the transport (post/update) and the + * per interval. Connectors supply the transport (post/update/delete) and the * channel-specific text rendering; all draft/lifecycle state lives here. */ export abstract class DraftReplyStream implements OutboundStream { @@ -51,6 +51,9 @@ export abstract class DraftReplyStream implements OutboundStream { /** Edit the progress message in place. */ protected abstract update(messageId: string, content: string): Promise; + /** Delete the progress message when the turn ends without text. */ + protected abstract remove(messageId: string): Promise; + /** Adapt the rendered thinking block for the channel (e.g. entity escaping). */ protected abstract renderProgress(block: string): string; @@ -147,11 +150,17 @@ export abstract class DraftReplyStream implements OutboundStream { const [first, ...remaining] = chunks; let chunksToPost = chunks; if (this.#messageId !== undefined) { - const replacement = - first ?? this.renderProgress(formatThinkingBlock(this.#progress)).slice(0, this.#textLimit); - if (replacement !== this.#lastPublishedText) { + if (first === undefined) { + try { + await this.remove(this.#messageId); + } catch (error) { + this.logger.warn(`${this.#channelLabel} progress message could not be deleted`, { + error: errorMessage(error), + }); + } + } else if (first !== this.#lastPublishedText) { try { - await this.update(this.#messageId, replacement); + await this.update(this.#messageId, first); chunksToPost = remaining; } catch (error) { this.logger.debug(`${this.#channelLabel} final replacement failed`, { diff --git a/src/channels/reactions.ts b/src/channels/reactions.ts new file mode 100644 index 0000000..d3d6109 --- /dev/null +++ b/src/channels/reactions.ts @@ -0,0 +1,107 @@ +import type { EmojiMartData } from "@emoji-mart/data"; +import data from "@emoji-mart/data/sets/15/native.json"; +import type { WebClient } from "@slack/web-api"; +import type { Message } from "discord.js"; +import type { Api } from "grammy"; +import type { ReactionTypeEmoji } from "grammy/types"; + +const emojiData: EmojiMartData = data; +const unicodeByName = new Map(); +const nameByUnicode = new Map(); + +// ponytail: Unicode 15 shortcodes; upgrade the data set when newer names are needed. +for (const emoji of Object.values(emojiData.emojis)) { + for (const [index, skin] of emoji.skins.entries()) { + const name = index === 0 ? emoji.id : `${emoji.id}::skin-tone-${index + 1}`; + unicodeByName.set(name, skin.native); + nameByUnicode.set(withoutPresentation(skin.native), name); + } +} + +/** A standard or provider-custom shortcode, without its surrounding colons. */ +function reactionName(reaction: string): string | undefined { + return /^:([\w+-]+(?:::skin-tone-[2-6])?):$/u.exec(reaction.trim())?.[1]; +} + +/** Resolve a standard shortcode; leave valid Unicode emoji intact. */ +export function unicodeReaction(reaction: string): string { + const value = reaction.trim(); + const name = reactionName(value); + if (name !== undefined) { + const [base = "", tone] = name.split("::"); + const canonical = emojiData.aliases[base] ?? base; + const unicode = unicodeByName.get(tone === undefined ? canonical : `${canonical}::${tone}`); + if (unicode === undefined) throw new Error(`Unknown emoji shortcode: ${value}`); + return unicode; + } + if (nameByUnicode.has(withoutPresentation(value)) || /^\p{RGI_Emoji}$/v.test(value)) return value; + throw new Error("A reaction must be one emoji or an :emoji_shortcode:."); +} + +/** Slack accepts names rather than Unicode, including workspace custom names. */ +export function slackReaction(reaction: string): string { + const name = reactionName(reaction) ?? nameByUnicode.get(withoutPresentation(reaction.trim())); + if (name === undefined) throw new Error("No Slack name for this emoji; use its :shortcode:."); + return name; +} + +/** Presentation selectors do not change the emoji; Telegram expects them omitted. */ +function withoutPresentation(emoji: string): string { + return emoji.replaceAll("\uFE0F", ""); +} + +export async function reactToTelegramMessage( + api: Pick, + chatId: number, + messageId: number, + reaction: string, +): Promise { + const customId = /^:custom_emoji_(\d{1,20}):$/u.exec(reaction.trim())?.[1]; + await api.setMessageReaction(chatId, messageId, [ + customId === undefined + ? { + type: "emoji", + // Telegram validates its supported set; do not duplicate that changing list here. + emoji: withoutPresentation(unicodeReaction(reaction)) as ReactionTypeEmoji["emoji"], + } + : { type: "custom_emoji", custom_emoji_id: customId }, + ]); +} + +export async function reactToSlackMessage( + web: Pick, + channel: string, + timestamp: string, + reaction: string, +): Promise { + try { + await web.reactions.add({ channel, timestamp, name: slackReaction(reaction) }); + } catch (error) { + if ((error as { data?: { error?: string } } | null)?.data?.error !== "already_reacted") { + throw error; + } + } +} + +export async function reactToDiscordMessage( + message: Pick, + reaction: string, +): Promise { + let emoji: string; + if (/^$/u.test(reaction.trim())) { + emoji = reaction.trim(); + } else { + try { + emoji = unicodeReaction(reaction); + } catch (error) { + const name = reactionName(reaction); + if (name === undefined || message.guild === null) throw error; + const emojis = await message.guild.emojis.fetch(); + const matches = emojis.filter((candidate) => candidate.name === name); + const custom = matches.size === 1 ? matches.first() : undefined; + if (custom === undefined) throw new Error(`No unique Discord custom emoji named :${name}:.`); + emoji = custom.id; + } + } + await message.react(emoji); +} diff --git a/src/channels/slack/channel.ts b/src/channels/slack/channel.ts index 6068946..fbe6c6a 100644 --- a/src/channels/slack/channel.ts +++ b/src/channels/slack/channel.ts @@ -22,6 +22,7 @@ import { errorMessage } from "../../shared/errors.js"; import type { Logger } from "../../shared/logger.js"; import { PendingChoices } from "../choices.js"; import { decodeCommandAction } from "../command-actions.js"; +import { reactToSlackMessage } from "../reactions.js"; import { isWorkspaceMember } from "./authorization.js"; import { type CodexConfigAccess, SlackConfigUi, slackConfigActionPrefix } from "./config-ui.js"; import { downloadSlackFile, SlackFileDownloadError } from "./file.js"; @@ -424,6 +425,7 @@ export class SlackChannel implements MessagingChannel { deliveryTarget: slackDeliveryTarget(event.channel, event.channel_type, route.replyThreadTs), }, reference: slackMessageReference(event.channel, event.ts), + react: (reaction) => reactToSlackMessage(this.#web, event.channel, event.ts, reaction), // Slack threads are flat: a reply references the thread root, not the // message being answered. When a scheduled-run notification lives in // this thread, point replyTo at it so its stored context resolves. @@ -819,6 +821,9 @@ function webMessagingApi(web: WebClient): SlackMessagingApi { blocks: options.blocks === undefined ? [] : [...options.blocks], }); }, + async deleteMessage(channel, ts) { + await web.chat.delete({ channel, ts }); + }, async uploadFile(options) { const contents = { file: options.path, filename: options.filename }; if (options.threadTs === undefined) { diff --git a/src/channels/slack/reply.ts b/src/channels/slack/reply.ts index 03d37c3..00c1790 100644 --- a/src/channels/slack/reply.ts +++ b/src/channels/slack/reply.ts @@ -71,6 +71,7 @@ export interface SlackEphemeralOptions { export interface SlackMessagingApi { postMessage(options: SlackPostOptions): Promise; updateMessage(options: SlackUpdateOptions): Promise; + deleteMessage(channel: string, ts: string): Promise; uploadFile(options: SlackUploadOptions): Promise; postEphemeral(options: SlackEphemeralOptions): Promise; fetchThreadReplies( @@ -328,6 +329,10 @@ export class SlackReplyStream extends DraftReplyStream { await this.#api.updateMessage({ channel: this.#channel, ts: messageTs, text: content }); } + protected async remove(messageTs: string): Promise { + await this.#api.deleteMessage(this.#channel, messageTs); + } + // Escaping matters even in previews: `` fragments would vanish and // `` would ping everyone. protected renderProgress(block: string): string { diff --git a/src/channels/telegram/channel.ts b/src/channels/telegram/channel.ts index fdca765..2e8c33a 100644 --- a/src/channels/telegram/channel.ts +++ b/src/channels/telegram/channel.ts @@ -26,6 +26,7 @@ import { errorMessage } from "../../shared/errors.js"; import type { Logger } from "../../shared/logger.js"; import { PendingChoices } from "../choices.js"; import { decodeCommandAction } from "../command-actions.js"; +import { reactToTelegramMessage } from "../reactions.js"; import { downloadTelegramFile, TelegramFileDownloadError } from "./file.js"; import { describeTelegramFile, @@ -392,6 +393,8 @@ export class TelegramChannel implements MessagingChannel { deliveryTarget: telegramDeliveryTarget(message.chat.id, incomingRoute.reply), }, reference: telegramMessageReference(message.chat.id, message.message_id), + react: (reaction) => + reactToTelegramMessage(api, message.chat.id, message.message_id, reaction), ...(content.replyToMessageId === undefined ? {} : { replyTo: telegramMessageReference(message.chat.id, content.replyToMessageId) }), diff --git a/src/codex/rpc.ts b/src/codex/rpc.ts index d6fede9..0e65b6c 100644 --- a/src/codex/rpc.ts +++ b/src/codex/rpc.ts @@ -118,6 +118,7 @@ export class CodexAppServer { readonly #codexHome: string; readonly #clientVersion: string; readonly #logger: Logger; + readonly #wirebotMcp: Readonly<{ url: string; token: string }> | undefined; public constructor( binaryPath: string, @@ -125,12 +126,14 @@ export class CodexAppServer { codexHome: string, clientVersion: string, logger: Logger, + wirebotMcp?: Readonly<{ url: string; token: string }>, ) { this.#binaryPath = binaryPath; this.#workspace = workspace; this.#codexHome = codexHome; this.#clientVersion = clientVersion; this.#logger = logger; + this.#wirebotMcp = wirebotMcp; } public async start(): Promise { @@ -139,10 +142,25 @@ export class CodexAppServer { } this.#stopping = false; - const appServerArgs = ["app-server", "--strict-config", "--listen", "stdio://"] as const; + const appServerArgs = ["app-server", "--strict-config", "--listen", "stdio://"]; + if (this.#wirebotMcp !== undefined) { + appServerArgs.push( + "-c", + `mcp_servers.wirebot.url=${JSON.stringify(this.#wirebotMcp.url)}`, + "-c", + 'mcp_servers.wirebot.bearer_token_env_var="WIREBOT_MCP_TOKEN"', + "-c", + "mcp_servers.wirebot.enabled=true", + "-c", + 'mcp_servers.wirebot.default_tools_approval_mode="approve"', + "-c", + "features.mcp_2026_07_28=true", + ); + } const environment = externalProcessEnvironment({ CODEX_HOME: this.#codexHome, LOG_FORMAT: "json", + ...(this.#wirebotMcp === undefined ? {} : { WIREBOT_MCP_TOKEN: this.#wirebotMcp.token }), }); const launch = await resolveCodexLaunch({ binaryPath: this.#binaryPath, diff --git a/src/codex/service.ts b/src/codex/service.ts index 0be9d9e..19d3b57 100644 --- a/src/codex/service.ts +++ b/src/codex/service.ts @@ -21,6 +21,7 @@ import type { DynamicToolSpec } from "../generated/codex/v2/DynamicToolSpec.js"; import type { FileChangeRequestApprovalResponse } from "../generated/codex/v2/FileChangeRequestApprovalResponse.js"; import type { GetAccountResponse } from "../generated/codex/v2/GetAccountResponse.js"; import type { LoginAccountResponse } from "../generated/codex/v2/LoginAccountResponse.js"; +import type { McpServerElicitationRequestResponse } from "../generated/codex/v2/McpServerElicitationRequestResponse.js"; import type { PermissionsRequestApprovalResponse } from "../generated/codex/v2/PermissionsRequestApprovalResponse.js"; import type { ThreadCompactStartResponse } from "../generated/codex/v2/ThreadCompactStartResponse.js"; import type { ThreadResumeResponse } from "../generated/codex/v2/ThreadResumeResponse.js"; @@ -49,6 +50,7 @@ import { } from "./thread-session.js"; export interface CodexInvocationContext { + readonly reactToMessage?: (reaction: string) => Promise; readonly owner?: ProviderReference; readonly deliveryTarget?: ProviderReference; readonly additionalContext?: ApplicationContext; @@ -249,6 +251,19 @@ export class CodexService { this.#dynamicTools.set(tool.spec.name, tool); } + /** MCP metadata identifies the turn without putting message IDs in model context. */ + public async reactToLastMessage( + threadId: string, + turnId: string, + reaction: string, + ): Promise { + const view = this.#sessions.get(threadId)?.view(turnId); + if (view?.origin !== "user" || view.invocation.reactToMessage === undefined) { + throw new Error("There is no reactable message for this active Wirebot turn."); + } + await view.invocation.reactToMessage(reaction); + } + public tryAcquireBackground(conversationKey: string): BackgroundLeaseDecision { const now = Date.now(); const retryAt = new Date(now + CodexService.#backgroundQuietPeriodMs); @@ -942,6 +957,51 @@ export class CodexService { await this.#rpc.reply(request.id, response); break; } + case "mcpServer/elicitation/request": { + const { threadId, turnId, serverName, message, _meta: metadata } = request.params; + const response: McpServerElicitationRequestResponse = { + action: "decline", + content: null, + _meta: null, + }; + // Only Codex tool approvals fit these buttons; forms need their own input UI. + if ( + request.params.mode === "form" && + Object.keys(request.params.requestedSchema.properties).length === 0 && + metadata !== null && + typeof metadata === "object" && + !Array.isArray(metadata) && + metadata.codex_approval_kind === "mcp_tool_call" + ) { + // Never attach an uncorrelated or completed request to another turn's user. + if (turnId !== null && this.#sessions.get(threadId)?.view(turnId) !== undefined) { + const choice = await this.askApproval( + request.id, + threadId, + turnId, + `MCP server: ${serverName}\n\n${message}${ + typeof metadata.tool_description === "string" + ? `\n\n${metadata.tool_description}` + : "" + }${ + metadata.tool_params === undefined + ? "" + : `\n\nArguments:\n${JSON.stringify(metadata.tool_params, null, 2)}` + }`, + Array.isArray(metadata.persist) && metadata.persist.includes("session"), + ); + if (choice !== "decline") { + response.action = "accept"; + response.content = {}; + response._meta = choice === "session" ? { persist: "session" } : null; + } + } + } else { + this.#logger.warn("MCP form or URL input is not supported", { serverName }); + } + await this.#rpc.reply(request.id, response); + break; + } case "item/tool/requestUserInput": { const session = this.#sessions.get(request.params.threadId); const responder = this.turnResponder(request.params.threadId, request.params.turnId); @@ -1051,6 +1111,7 @@ export class CodexService { threadId: string, turnId: string, prompt: string, + allowSession = true, ): Promise<"once" | "session" | "decline"> { const session = this.#sessions.get(threadId); const responder = this.turnResponder(threadId, turnId); @@ -1061,12 +1122,12 @@ export class CodexService { prompt, [ { id: "once", label: "Allow once" }, - { id: "session", label: "Allow for session" }, + ...(allowSession ? [{ id: "session", label: "Allow for session" }] : []), { id: "decline", label: "Deny" }, ], signal, ); - return answer === "once" || answer === "session" ? answer : "decline"; + return answer === "once" || (allowSession && answer === "session") ? answer : "decline"; } finally { session.endServerRequest(requestId); } @@ -1132,6 +1193,8 @@ function createRemoteClientContext(connector: string): ApplicationContext { kind: "application", value: `This Codex session is operated through Wirebot, a remote messaging bridge. The user reads and replies through ${connectorName} and is not present at the machine where Codex and its commands run. +\`react_to_last_message\` (emoji or :shortcode:): Reactions are part of the conversation. Use them at your discretion when a brief nonverbal response feels appropriate. A reaction can accompany a reply or serve as the entire response. Don’t announce or explain it unless the user asks. It’s also fine not to react. + Host-local UI is not visible or accessible to the user: - Do not open browsers, GUI applications, editors, file managers, or OAuth pages as a way of handing work to the user. - Never ask the user to visit localhost, 127.0.0.1, a file:// URL, or another host-local address. Those addresses refer to the Codex host, not the user's device. diff --git a/src/codex/thread-session.ts b/src/codex/thread-session.ts index 844d06f..dfa2c45 100644 --- a/src/codex/thread-session.ts +++ b/src/codex/thread-session.ts @@ -458,7 +458,7 @@ export class ThreadSession { }; if (view.origin !== "scheduled") { await view.stream.complete( - deliveryText(view.origin, turn, finalText, resolution), + deliveryText(turn, finalText, resolution.unavailable), resolution.attachments, ); await dispose(); @@ -502,21 +502,17 @@ function latestReasoningSummary(reasoning: ReadonlyMap): strin } function deliveryText( - origin: TurnOrigin, turn: Turn, finalText: string, - resolution: Readonly<{ - attachments: readonly OutboundAttachment[]; - unavailable: readonly string[]; - }>, + unavailableAttachments: readonly string[], ): string { if (turn.status === "interrupted" && finalText.length === 0) return "Stopped."; - const text = unavailableAttachmentsWarning(finalText, resolution.unavailable); + const text = unavailableAttachmentsWarning(finalText, unavailableAttachments); if (text.length > 0) return text; if (turn.items.some((item) => item.type === "contextCompaction")) { return "Context compacted."; } - return origin === "user" && resolution.attachments.length === 0 ? "Done." : ""; + return ""; } export const silentStream: OutboundStream = { diff --git a/src/core/bridge.ts b/src/core/bridge.ts index 8eb750b..60843f9 100644 --- a/src/core/bridge.ts +++ b/src/core/bridge.ts @@ -208,6 +208,7 @@ export class CodexBridge { message.attachments, { owner, + ...(message.react === undefined ? {} : { reactToMessage: message.react }), ...(message.address.deliveryTarget === undefined ? {} : { deliveryTarget: message.address.deliveryTarget }), diff --git a/src/core/channel.ts b/src/core/channel.ts index acf554f..b6ba0d6 100644 --- a/src/core/channel.ts +++ b/src/core/channel.ts @@ -136,6 +136,8 @@ export interface InboundMessage { */ readonly isAdmin: boolean; readonly responder: MessageResponder; + /** React to this inbound message; provider credentials and IDs stay in the connector. */ + readonly react?: (reaction: string) => Promise; /** * Release provider-owned temporary resources after the message is fully * handled. The message handler owns this callback, including across a diff --git a/src/index.ts b/src/index.ts index 5a5f681..66690f6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -14,6 +14,7 @@ import { CodexBridge } from "./core/bridge.js"; import type { MessagingChannel } from "./core/channel.js"; import { ConversationStore } from "./core/conversation-store.js"; import { WirebotSettingsStore } from "./core/settings-store.js"; +import { WirebotMcpServer } from "./mcp/server.js"; import { BrowserAuth } from "./miniapp/browser-auth.js"; import { MiniAppServer } from "./miniapp/server.js"; import { QuickTunnel } from "./miniapp/tunnel.js"; @@ -73,12 +74,24 @@ export async function runWirebot(): Promise { ); const binaryPath = await toolchains.ensureVersion(pinnedCodexVersion); + let codex: CodexService | undefined; + const mcp = new WirebotMcpServer( + async (threadId, turnId, reaction) => { + if (codex === undefined) throw new Error("Codex is not ready."); + await codex.reactToLastMessage(threadId, turnId, reaction); + }, + logger.child({ component: "mcp" }), + ); + const mcpConnection = await mcp.start(); + resources.push(mcp); + const rpc = new CodexAppServer( binaryPath, config.workspace, codexHome, wirebotVersion, logger.child({ component: "codex-rpc" }), + mcpConnection, ); resources.push(rpc); await rpc.start(); @@ -109,7 +122,7 @@ export async function runWirebot(): Promise { chatgptAuth === undefined ? undefined : () => Promise.resolve(chatgptAuth), ); let liveRuntime: CodexRuntimeService | undefined; - const codex = new CodexService( + codex = new CodexService( rpc, conversations, config.workspace, diff --git a/src/mcp/server.ts b/src/mcp/server.ts new file mode 100644 index 0000000..ed8a9fc --- /dev/null +++ b/src/mcp/server.ts @@ -0,0 +1,88 @@ +import { randomBytes, timingSafeEqual } from "node:crypto"; +import { + createMcpHandler, + hostHeaderValidationResponse, + type McpHttpHandler, + McpServer, + originValidationResponse, +} from "@modelcontextprotocol/server"; +import { z } from "zod"; +import { errorMessage } from "../shared/errors.js"; +import type { Logger } from "../shared/logger.js"; +import { wirebotVersion } from "../shared/version.js"; + +const turnMetadata = z.object({ thread_id: z.string().min(1), turn_id: z.string().min(1) }); +const maxRequestBytes = 32 * 1_024; + +/** Private MCP endpoint owned by the engine, using the 2026-07-28 stateless protocol. */ +export class WirebotMcpServer { + #server: ReturnType | undefined; + readonly #handler: McpHttpHandler; + readonly #token = randomBytes(32).toString("base64url"); + + public constructor( + react: (threadId: string, turnId: string, reaction: string) => Promise, + logger: Logger, + ) { + this.#handler = createMcpHandler( + () => { + const server = new McpServer( + { name: "wirebot", version: wirebotVersion }, + { supportedProtocolVersions: ["2026-07-28"] }, + ); + server.registerTool( + "react_to_last_message", + { + description: "React to this turn's message with an emoji or :shortcode:.", + inputSchema: z.strictObject({ reaction: z.string().trim().min(1).max(128) }), + annotations: { readOnlyHint: false, destructiveHint: false, idempotentHint: true }, + }, + async ({ reaction }, context) => { + const parsed = turnMetadata.safeParse(context.mcpReq._meta?.["x-codex-turn-metadata"]); + try { + if (!parsed.success) + throw new Error("This tool requires Codex thread and turn metadata."); + await react(parsed.data.thread_id, parsed.data.turn_id, reaction); + return { content: [{ type: "text", text: "Reaction set." }] }; + } catch (error) { + return { isError: true, content: [{ type: "text", text: errorMessage(error) }] }; + } + }, + ); + return server; + }, + { + legacy: "reject", + onerror: (error) => logger.debug("MCP request rejected", { error: errorMessage(error) }), + }, + ); + } + + public start(): { readonly url: string; readonly token: string } { + const authorization = Buffer.from(`Bearer ${this.#token}`); + this.#server ??= Bun.serve({ + hostname: "127.0.0.1", + port: 0, + maxRequestBodySize: maxRequestBytes, + fetch: (request) => { + const rejected = + hostHeaderValidationResponse(request, ["127.0.0.1"]) ?? + originValidationResponse(request, []); + if (rejected !== undefined) return rejected; + if (new URL(request.url).pathname !== "/mcp") return new Response(null, { status: 404 }); + const supplied = Buffer.from(request.headers.get("authorization") ?? ""); + if (supplied.length !== authorization.length || !timingSafeEqual(supplied, authorization)) { + return new Response(null, { status: 401, headers: { "WWW-Authenticate": "Bearer" } }); + } + return this.#handler.fetch(request); + }, + }); + return { url: `http://127.0.0.1:${this.#server.port}/mcp`, token: this.#token }; + } + + public async stop(): Promise { + await this.#handler.close(); + await this.#server?.stop(true); + this.#server = undefined; + } +} diff --git a/test/draft-stream.test.ts b/test/draft-stream.test.ts index dc0b6d2..0a81afc 100644 --- a/test/draft-stream.test.ts +++ b/test/draft-stream.test.ts @@ -1,11 +1,17 @@ import { describe, expect, it } from "bun:test"; import { DraftReplyStream } from "../src/channels/draft-stream.js"; +import type { OutboundAttachment } from "../src/core/channel.js"; +import { deferred } from "../src/shared/async.js"; import { Logger } from "../src/shared/logger.js"; class RecordingDraftStream extends DraftReplyStream { public readonly posts: string[] = []; public readonly updates: string[] = []; + public readonly deletions: string[] = []; + public readonly attachments: OutboundAttachment[] = []; + public initialPost = Promise.resolve("message-1"); public failUpdates = false; + public failDeletion = false; public constructor(limit = 100) { super(new Logger("error"), "Test", limit); @@ -13,7 +19,7 @@ class RecordingDraftStream extends DraftReplyStream { protected async postInitial(content: string): Promise { this.posts.push(content); - return "message-1"; + return await this.initialPost; } protected async post(content: string): Promise { @@ -25,6 +31,15 @@ class RecordingDraftStream extends DraftReplyStream { this.updates.push(content); } + protected async remove(messageId: string): Promise { + this.deletions.push(messageId); + if (this.failDeletion) throw new Error("delete failed"); + } + + protected override async finishExtras(attachments: readonly OutboundAttachment[]): Promise { + this.attachments.push(...attachments); + } + protected renderProgress(block: string): string { return block; } @@ -78,13 +93,46 @@ describe("DraftReplyStream", () => { expect(stream.posts).toEqual(["▌ Thinking…", "Finished"]); }); - it("publishes the latest progress when a turn ends without an answer", async () => { + it("deletes progress when a turn ends without an answer", async () => { const stream = new RecordingDraftStream(); await stream.start({ summary: "Thinking…", actions: [], plan: [] }); stream.setProgress({ summary: "Waiting for input", actions: [], plan: [] }); await stream.complete(""); + await stream.complete(""); + stream.setProgress({ summary: "Late progress", actions: [], plan: [] }); - expect(stream.updates).toEqual(["▌ Waiting for input"]); + expect(stream.deletions).toEqual(["message-1"]); + expect(stream.updates).toEqual([]); + expect(stream.posts).toEqual(["▌ Thinking…"]); + }); + + it("waits for the initial post before deleting and still delivers attachments on deletion failure", async () => { + const stream = new RecordingDraftStream(); + const initialPost = deferred(); + stream.initialPost = initialPost.promise; + stream.failDeletion = true; + const attachments = [{ path: "/tmp/report.pdf", filename: "report.pdf" }]; + + const starting = stream.start(); + const completing = stream.complete("", attachments); + expect(stream.deletions).toEqual([]); + initialPost.resolve("late-message"); + await Promise.all([starting, completing]); + + expect(stream.deletions).toEqual(["late-message"]); + expect(stream.attachments).toEqual(attachments); + expect(stream.updates).toEqual([]); + expect(stream.posts).toEqual(["▌ Thinking…"]); + }); + + it("sends nothing for an empty answer without a progress message", async () => { + const stream = new RecordingDraftStream(); + + await stream.complete(""); + + expect(stream.deletions).toEqual([]); + expect(stream.updates).toEqual([]); + expect(stream.posts).toEqual([]); }); }); diff --git a/test/mcp-approvals.test.ts b/test/mcp-approvals.test.ts new file mode 100644 index 0000000..bf3b6dd --- /dev/null +++ b/test/mcp-approvals.test.ts @@ -0,0 +1,198 @@ +import { expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { + CodexAppServer, + NotificationListener, + ServerRequestHandler, +} from "../src/codex/rpc.js"; +import { CodexService } from "../src/codex/service.js"; +import { silentStream } from "../src/codex/thread-session.js"; +import type { ChoiceOption, MessageResponder } from "../src/core/channel.js"; +import { ConversationStore } from "../src/core/conversation-store.js"; +import type { RequestId } from "../src/generated/codex/RequestId.js"; +import type { McpServerElicitationRequestParams } from "../src/generated/codex/v2/McpServerElicitationRequestParams.js"; +import type { McpServerElicitationRequestResponse } from "../src/generated/codex/v2/McpServerElicitationRequestResponse.js"; +import type { Turn } from "../src/generated/codex/v2/Turn.js"; +import { deferred } from "../src/shared/async.js"; +import { Logger } from "../src/shared/logger.js"; + +test("MCP approvals use the active user's choices and preserve Codex's approval scope", async () => { + const directory = await mkdtemp(join(tmpdir(), "wirebot-approvals-")); + const logger = new Logger("error"); + const listeners: NotificationListener[] = []; + let handleRequest: ServerRequestHandler = async () => { + throw new Error("Server request handler not installed"); + }; + const replies = new Map(); + let started = deferred(); + let nextTurn = 0; + const rpc = { + onNotification: (listener: NotificationListener) => { + listeners.push(listener); + return () => {}; + }, + onExit: () => () => {}, + setServerRequestHandler: (handler: ServerRequestHandler) => { + handleRequest = handler; + }, + reply: async (id: RequestId, response: McpServerElicitationRequestResponse) => { + replies.set(id, response); + }, + request: async ({ method }: { method: string }) => { + if (method === "account/read") return { account: null, requiresOpenaiAuth: false }; + if (method === "thread/start") return { thread: { id: "thread" } }; + if (method !== "turn/start") throw new Error(`Unexpected RPC: ${method}`); + const turn: Turn = { + id: `turn-${++nextTurn}`, + items: [], + itemsView: "full", + status: "inProgress", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; + for (const listener of listeners) + listener({ method: "turn/started", params: { threadId: "thread", turn } }); + started.resolve(turn); + return { turn }; + }, + } as unknown as CodexAppServer; + const service = new CodexService( + rpc, + new ConversationStore(join(directory, "conversations.json"), logger), + directory, + directory, + directory, + logger, + ); + let decision = "once"; + const prompts: { text: string; options: readonly ChoiceOption[] }[] = []; + const waiting = deferred(); + const responder: MessageResponder = { + createStream: () => silentStream, + sendText: async () => {}, + askChoice: async (text, options, signal) => { + prompts.push({ text, options }); + if (decision !== "pending") return decision; + if (signal === undefined) throw new Error("Approval cancellation signal is missing"); + return await new Promise((resolve) => { + signal.addEventListener("abort", () => resolve("decline"), { once: true }); + waiting.resolve(); + }); + }, + }; + const run = service.runTurn("chat", "telegram", "Hello", responder, false, [], {}); + const turn = await started.promise; + const metadata = { + codex_approval_kind: "mcp_tool_call", + persist: ["session", "always"], + tool_description: "Update an issue.", + tool_params: { issue: 42 }, + }; + const params: McpServerElicitationRequestParams = { + threadId: "thread", + turnId: turn.id, + serverName: "external", + mode: "form", + message: 'Allow tool "update_issue"?', + _meta: metadata, + requestedSchema: { type: "object", properties: {} }, + }; + let requestId = 0; + const request = async (overrides: Partial = {}) => { + const id = ++requestId; + await handleRequest({ + id, + method: "mcpServer/elicitation/request", + params: { ...params, ...overrides } as McpServerElicitationRequestParams, + }); + return replies.get(id); + }; + const declined = { action: "decline", content: null, _meta: null }; + const complete = (current: Turn) => { + for (const listener of listeners) + listener({ + method: "turn/completed", + params: { + threadId: "thread", + turn: { ...current, status: "completed" }, + }, + }); + }; + try { + expect(await request()).toEqual({ action: "accept", content: {}, _meta: null }); + expect(prompts[0]?.text).toContain("MCP server: external"); + expect(prompts[0]?.text).toContain("Update an issue."); + expect(prompts[0]?.text).toContain('"issue": 42'); + expect(prompts[0]?.options.map(({ id }) => id)).toEqual(["once", "session", "decline"]); + decision = "session"; + expect(await request()).toEqual({ + action: "accept", + content: {}, + _meta: { persist: "session" }, + }); + decision = "decline"; + expect(await request()).toEqual(declined); + decision = "once"; + expect(await request({ _meta: { ...metadata, persist: [] } })).toEqual({ + action: "accept", + content: {}, + _meta: null, + }); + expect(prompts.at(-1)?.options.map(({ id }) => id)).toEqual(["once", "decline"]); + decision = "session"; + expect(await request({ _meta: { ...metadata, persist: [] } })).toEqual(declined); + + const promptCount = prompts.length; + for (const overrides of [ + { turnId: null }, + { turnId: "stale" }, + { threadId: "other" }, + { _meta: null }, + { _meta: [] }, + { _meta: { codex_approval_kind: "other" } }, + { + requestedSchema: { + type: "object" as const, + properties: { name: { type: "string" as const } }, + }, + }, + { mode: "url" as const, url: "https://example.com/login", elicitationId: "login" }, + ]) + expect(await request(overrides)).toEqual(declined); + expect(prompts).toHaveLength(promptCount); + + decision = "pending"; + const pending = request(); + await waiting.promise; + for (const listener of listeners) + listener({ method: "serverRequest/resolved", params: { threadId: "thread", requestId } }); + expect(await pending).toEqual(declined); + + complete(turn); + await run; + const afterCompletion = prompts.length; + expect(await request()).toEqual(declined); + expect(prompts).toHaveLength(afterCompletion); + started = deferred(); + const scheduled = service.runScheduledTurn({ + conversationKey: "chat", + connector: "telegram", + prompt: "Background work", + thread: { mode: "existing", threadId: "thread" }, + invocation: {}, + }); + const unattended = await started.promise; + expect(await request({ turnId: unattended.id })).toEqual(declined); + expect(prompts).toHaveLength(afterCompletion); + complete(unattended); + await (await scheduled).dispose(); + } finally { + complete(turn); + await run; + await rm(directory, { recursive: true, force: true }); + } +}); diff --git a/test/reactions.test.ts b/test/reactions.test.ts new file mode 100644 index 0000000..3adac29 --- /dev/null +++ b/test/reactions.test.ts @@ -0,0 +1,350 @@ +import { describe, expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { CallToolResult, DiscoverResult, ListToolsResult } from "@modelcontextprotocol/server"; +import { Collection } from "discord.js"; +import { + reactToDiscordMessage, + reactToSlackMessage, + reactToTelegramMessage, + slackReaction, + unicodeReaction, +} from "../src/channels/reactions.js"; +import type { CodexAppServer, NotificationListener } from "../src/codex/rpc.js"; +import { CodexService } from "../src/codex/service.js"; +import { silentStream } from "../src/codex/thread-session.js"; +import { CodexBridge } from "../src/core/bridge.js"; +import type { InboundMessage } from "../src/core/channel.js"; +import { ConversationStore } from "../src/core/conversation-store.js"; +import type { Turn } from "../src/generated/codex/v2/Turn.js"; +import { WirebotMcpServer } from "../src/mcp/server.js"; +import { deferred } from "../src/shared/async.js"; +import { Logger } from "../src/shared/logger.js"; + +const logger = new Logger("error"); +const version = "2026-07-28"; + +describe("Message reactions", () => { + test("converts shortcodes, presentation variants, and skin tones without changing the emoji", () => { + for (const name of [":thumbsup:", ":+1:"]) expect(unicodeReaction(name)).toBe("👍"); + expect(unicodeReaction(":heart:")).toBe("❤️"); + expect(unicodeReaction(":thumbsup::skin-tone-4:")).toBe("👍🏽"); + expect(slackReaction("👍🏽")).toBe("+1::skin-tone-4"); + expect(slackReaction("❤️")).toBe("heart"); + expect(slackReaction("❤")).toBe("heart"); + expect(slackReaction(":party_blob:")).toBe("party_blob"); + expect(unicodeReaction("👨‍💻")).toBe("👨‍💻"); + for (const invalid of ["", "hello", ":unknown_reaction:", "👍👍", "hi 👍", "::"]) { + expect(() => unicodeReaction(invalid)).toThrow(); + } + }); + + test("uses provider APIs, handles duplicates, and preserves provider failures", async () => { + const telegram: unknown[][] = []; + const api = { + setMessageReaction: async (...args: unknown[]) => { + telegram.push(args); + return true; + }, + }; + await reactToTelegramMessage(api, -123, 42, ":heart:"); + expect(telegram[0]).toEqual([-123, 42, [{ type: "emoji", emoji: "❤" }]]); + await reactToTelegramMessage(api, -123, 43, ":custom_emoji_123456789:"); + expect(telegram[1]).toEqual([ + -123, + 43, + [{ type: "custom_emoji", custom_emoji_id: "123456789" }], + ]); + await reactToTelegramMessage(api, -123, 44, ":100:"); + expect(telegram[2]).toEqual([-123, 44, [{ type: "emoji", emoji: "💯" }]]); + + const slack: unknown[] = []; + const web = { + reactions: { + add: async (args: unknown) => { + slack.push(args); + return {}; + }, + }, + }; + await reactToSlackMessage(web, "C123", "100.123456", "👍🏽"); + expect(slack[0]).toEqual({ channel: "C123", timestamp: "100.123456", name: "+1::skin-tone-4" }); + web.reactions.add = async () => { + throw { data: { error: "already_reacted" } }; + }; + await reactToSlackMessage(web, "C123", "100.123456", ":thumbsup:"); + const denied = new Error("missing_scope"); + web.reactions.add = async () => { + throw denied; + }; + await expect(reactToSlackMessage(web, "C123", "100.123456", "👍")).rejects.toBe(denied); + api.setMessageReaction = async () => { + throw new Error("REACTION_INVALID"); + }; + await expect(reactToTelegramMessage(api, -123, 42, "👍")).rejects.toThrow("REACTION_INVALID"); + + const discord: string[] = []; + const message = { + react: async (emoji: string) => { + discord.push(emoji); + }, + guild: { + emojis: { + fetch: async () => + new Collection([ + ["123456789012345678", { name: "party_blob", id: "123456789012345678" }], + ]), + }, + }, + }; + await reactToDiscordMessage(message, ":thumbsup:"); + await reactToDiscordMessage(message, ":party_blob:"); + await reactToDiscordMessage(message, ""); + expect(discord).toEqual(["👍", "123456789012345678", ""]); + await expect(reactToDiscordMessage(message, ":missing:")).rejects.toThrow("No unique Discord"); + }); + + test("MCP routes concurrent and queued turns, rejects stale calls, and adds no prompt context", async () => { + const directory = await mkdtemp(join(tmpdir(), "wirebot-reactions-")); + const listeners: NotificationListener[] = []; + const starts = new Map( + ["first", "second", "other", "silent", "scheduled"].map((text) => [ + text, + deferred<{ threadId: string; turn: Turn; params: Record }>(), + ]), + ); + const startOf = (text: string) => { + const started = starts.get(text); + if (started === undefined) throw new Error(`Unexpected turn input: ${text}`); + return started; + }; + let nextThread = 0; + let nextTurn = 0; + const rpc = { + onNotification: (listener: NotificationListener) => { + listeners.push(listener); + return () => {}; + }, + onExit: () => () => {}, + setServerRequestHandler: () => {}, + request: async ({ method, params }: { method: string; params: Record }) => { + if (method === "account/read") return { account: null, requiresOpenaiAuth: false }; + if (method === "thread/start") return { thread: { id: `thread-${++nextThread}` } }; + if (method === "turn/start") { + const turn: Turn = { + id: `turn-${++nextTurn}`, + items: [], + itemsView: "full", + status: "inProgress", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; + const threadId = String(params.threadId); + for (const listener of listeners) + listener({ method: "turn/started", params: { threadId, turn } }); + const input = params.input as [{ text: string }]; + startOf(input[0].text).resolve({ threadId, turn, params }); + return { turn }; + } + throw new Error(`Unexpected RPC: ${method}`); + }, + } as unknown as CodexAppServer; + const service = new CodexService( + rpc, + new ConversationStore(join(directory, "conversations.json"), logger), + directory, + directory, + directory, + logger, + undefined, + () => false, + ); + const bridge = new CodexBridge( + service, + undefined, + logger, + {} as never, + { contextForReply: async () => undefined } as never, + ); + const server = new WirebotMcpServer(service.reactToLastMessage.bind(service), logger); + const connection = await server.start(); + const authorizations = { Authorization: `Bearer ${connection.token}` }; + const request = async ( + method: string, + params: Record = {}, + headers: Record = {}, + ) => { + const response = await fetch(connection.url, { + method: "POST", + headers: { + ...authorizations, + "Content-Type": "application/json", + Accept: "application/json, text/event-stream", + "MCP-Protocol-Version": version, + "Mcp-Method": method, + ...(method === "tools/call" ? { "Mcp-Name": String(params.name) } : {}), + ...headers, + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method, + params: { + ...params, + _meta: { + "io.modelcontextprotocol/protocolVersion": version, + "io.modelcontextprotocol/clientCapabilities": {}, + ...(params._meta as object), + }, + }, + }), + }); + return { response, body: (await response.json()) as { result: Result } }; + }; + const react = (threadId: string, turnId: string, reaction = "👍") => + request("tools/call", { + name: "react_to_last_message", + arguments: { reaction }, + _meta: { "x-codex-turn-metadata": { thread_id: threadId, turn_id: turnId } }, + }); + const reactions: string[] = []; + const replies = new Map(); + const message = (text: string, key: string): InboundMessage => ({ + id: text, + address: { channel: "telegram", key, isPrivate: true, isGuest: false }, + sender: { id: "42", displayName: "User" }, + text, + attachments: [], + isAdmin: true, + react: async (reaction) => { + reactions.push(`${text}:${reaction}`); + }, + responder: { + createStream: () => ({ + ...silentStream, + complete: async (reply) => { + replies.set(text, reply); + }, + }), + sendText: async () => {}, + askChoice: async () => "decline", + }, + }); + const complete = (started: { threadId: string; turn: Turn }) => { + for (const listener of listeners) + listener({ + method: "turn/completed", + params: { threadId: started.threadId, turn: { ...started.turn, status: "completed" } }, + }); + }; + try { + const discovery = await request("server/discover"); + expect(discovery.response.status).toBe(200); + expect(discovery.body.result.supportedVersions).toEqual([version]); + expect(discovery.body.result._meta?.["io.modelcontextprotocol/serverInfo"]).toMatchObject({ + name: "wirebot", + }); + expect(discovery.response.headers.get("Mcp-Session-Id")).toBeNull(); + const tools = await request("tools/list"); + expect(tools.body.result.tools).toHaveLength(1); + expect(tools.body.result.tools[0].name).toBe("react_to_last_message"); + expect(Object.keys(tools.body.result.tools[0].inputSchema.properties)).toEqual(["reaction"]); + + expect((await fetch(connection.url)).status).toBe(401); + const oversized = await fetch(connection.url, { + method: "POST", + headers: authorizations, + body: " ".repeat(40_000), + }); + expect(oversized.status).toBe(413); + const legacy = await fetch(connection.url, { + method: "POST", + headers: { + ...authorizations, + "Content-Type": "application/json", + Accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-11-25", + capabilities: {}, + clientInfo: { name: "legacy", version: "1" }, + }, + }), + }); + expect(legacy.status).toBe(400); + expect( + ( + await fetch(connection.url, { + headers: { ...authorizations, Origin: "https://evil.example" }, + }) + ).status, + ).toBe(403); + expect( + (await fetch(connection.url, { headers: { ...authorizations, Host: "evil.example" } })) + .status, + ).toBe(403); + expect( + (await request("tools/list", {}, { "Mcp-Method": "tools/call" })).response.status, + ).toBe(400); + const missingMetadata = await request("tools/call", { + name: "react_to_last_message", + arguments: { reaction: "👍" }, + }); + expect(missingMetadata.body.result.isError).toBe(true); + + const firstRun = bridge.handleMessage(message("first", "chat-a")); + const first = await startOf("first").promise; + const secondRun = bridge.handleMessage(message("second", "chat-a")); + const otherRun = bridge.handleMessage(message("other", "chat-b")); + const other = await startOf("other").promise; + expect(first.params.input).toEqual([{ type: "text", text: "first", text_elements: [] }]); + expect(first.params.additionalContext).toBeUndefined(); + const results = await Promise.all([ + react(first.threadId, first.turn.id), + react(other.threadId, other.turn.id, ":heart:"), + ]); + expect(results.every(({ body }) => !body.result.isError)).toBe(true); + expect(reactions.sort()).toEqual(["first:👍", "other::heart:"].sort()); + expect((await react(first.threadId, other.turn.id)).body.result.isError).toBe(true); + expect((await react(first.threadId, first.turn.id, "")).body.result.isError).toBe(true); + + complete(first); + await firstRun; + expect(replies.get("first")).toBe(""); + const second = await startOf("second").promise; + expect((await react(first.threadId, first.turn.id)).body.result.isError).toBe(true); + expect((await react(second.threadId, second.turn.id)).body.result.isError).not.toBe(true); + expect(reactions).toContain("second:👍"); + complete(second); + complete(other); + await Promise.all([secondRun, otherRun]); + + const silentRun = bridge.handleMessage(message("silent", "chat-a")); + complete(await startOf("silent").promise); + await silentRun; + expect(replies.get("silent")).toBe(""); + + const scheduledRun = service.runScheduledTurn({ + conversationKey: "chat-a", + connector: "telegram", + prompt: "scheduled", + thread: { mode: "existing", threadId: first.threadId }, + invocation: {}, + }); + const scheduled = await startOf("scheduled").promise; + expect((await react(scheduled.threadId, scheduled.turn.id)).body.result.isError).toBe(true); + complete(scheduled); + await (await scheduledRun).dispose(); + } finally { + await server.stop(); + await rm(directory, { recursive: true, force: true }); + } + }); +});