From ed4617ec8440dab43c41e131ff424326e5f865e0 Mon Sep 17 00:00:00 2001 From: "GeonWoo Jeon (Jay)" Date: Tue, 7 Apr 2026 22:49:37 +0900 Subject: [PATCH] fix(openclaw): stabilize reply polling and tmux injection --- .../__tests__/reply-listener-discord.test.ts | 131 ++++++++++++++++++ src/openclaw/reply-listener-discord.ts | 110 +++++++++++++++ src/openclaw/reply-listener-injection.ts | 74 ++++++++++ src/openclaw/reply-listener-telegram.ts | 92 ++++++++++++ 4 files changed, 407 insertions(+) create mode 100644 src/openclaw/__tests__/reply-listener-discord.test.ts create mode 100644 src/openclaw/reply-listener-discord.ts create mode 100644 src/openclaw/reply-listener-injection.ts create mode 100644 src/openclaw/reply-listener-telegram.ts diff --git a/src/openclaw/__tests__/reply-listener-discord.test.ts b/src/openclaw/__tests__/reply-listener-discord.test.ts new file mode 100644 index 000000000..357c8fd89 --- /dev/null +++ b/src/openclaw/__tests__/reply-listener-discord.test.ts @@ -0,0 +1,131 @@ +import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from "bun:test" +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync } from "fs" +import { tmpdir } from "os" +import { join } from "path" +import { ReplyListenerRateLimiter } from "../reply-listener-injection" +import { pollDiscordReplies } from "../reply-listener-discord" +import * as injectionModule from "../reply-listener-injection" +import * as sessionRegistryModule from "../session-registry" +import type { ReplyListenerDaemonState } from "../reply-listener-state" +import type { OpenClawConfig } from "../types" + +const originalHome = process.env.HOME +const originalUserProfile = process.env.USERPROFILE + +const tempHome = mkdtempSync(join(tmpdir(), "openclaw-reply-listener-discord-")) +const stateDir = join(tempHome, ".omx", "state") +const stateFilePath = join(stateDir, "reply-listener-state.json") + +function createConfig(): OpenClawConfig { + return { + enabled: true, + gateways: { + gateway: { + type: "http", + url: "https://example.com", + method: "POST", + }, + }, + hooks: {}, + replyListener: { + discordBotToken: "discord-token", + discordChannelId: "channel-1", + authorizedDiscordUserIds: ["user-1"], + pollIntervalMs: 10, + rateLimitPerMinute: 10, + maxMessageLength: 500, + includePrefix: true, + }, + } +} + +function createState(): ReplyListenerDaemonState { + return { + isRunning: true, + pid: 1234, + startedAt: "2026-04-07T00:00:00.000Z", + startupToken: "startup-token", + configSignature: null, + lastPollAt: "2026-04-07T00:00:01.000Z", + telegramLastUpdateId: null, + discordLastMessageId: null, + lastDiscordMessageId: null, + messagesSeen: 0, + messagesInjected: 0, + errors: 0, + } +} + +describe("pollDiscordReplies", () => { + beforeEach(() => { + process.env.HOME = tempHome + process.env.USERPROFILE = tempHome + rmSync(stateDir, { recursive: true, force: true }) + mkdirSync(stateDir, { recursive: true }) + }) + + afterEach(() => { + mock.restore() + }) + + test("records HTTP failures in daemon state when Discord returns non-ok", async () => { + const fetchSpy = spyOn(globalThis, "fetch").mockResolvedValue( + new Response("unauthorized", { + status: 401, + }), + ) + + const state = createState() + + await pollDiscordReplies(createConfig(), state, new ReplyListenerRateLimiter(10)) + + expect(fetchSpy).toHaveBeenCalledTimes(1) + expect(state.errors).toBe(1) + expect(state.lastError).toBe("Discord API error: HTTP 401") + expect(existsSync(stateFilePath)).toBe(true) + + const persistedState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as ReplyListenerDaemonState + expect(persistedState.errors).toBe(1) + expect(persistedState.lastError).toBe("Discord API error: HTTP 401") + expect(persistedState.messagesSeen).toBe(0) + }) + + test("increments messagesInjected when a Discord reply matches a registered message", async () => { + const fetchSpy = spyOn(globalThis, "fetch") + .mockResolvedValueOnce( + new Response( + JSON.stringify([ + { + id: "incoming-1", + content: "Ship it", + author: { id: "user-1" }, + message_reference: { message_id: "outbound-1" }, + }, + ]), + { status: 200 }, + ), + ) + .mockResolvedValueOnce(new Response(null, { status: 204 })) + const lookupSpy = spyOn(sessionRegistryModule, "lookupByMessageId").mockReturnValue({ + sessionId: "ses-1", + tmuxSession: "session-1", + tmuxPaneId: "%7", + projectPath: "/tmp/project", + platform: "discord-bot", + messageId: "outbound-1", + createdAt: "2026-04-07T00:00:00.000Z", + }) + const injectSpy = spyOn(injectionModule, "injectReplyIntoPane").mockResolvedValue(true) + + const state = createState() + + await pollDiscordReplies(createConfig(), state, new ReplyListenerRateLimiter(10)) + + expect(lookupSpy).toHaveBeenCalledWith("discord-bot", "outbound-1") + expect(injectSpy).toHaveBeenCalledWith("%7", "Ship it", "discord", createConfig()) + expect(fetchSpy).toHaveBeenCalledTimes(2) + expect(state.messagesSeen).toBe(1) + expect(state.messagesInjected).toBe(1) + expect(state.lastDiscordMessageId).toBe("incoming-1") + }) +}) diff --git a/src/openclaw/reply-listener-discord.ts b/src/openclaw/reply-listener-discord.ts new file mode 100644 index 000000000..fbc504e46 --- /dev/null +++ b/src/openclaw/reply-listener-discord.ts @@ -0,0 +1,110 @@ +import { lookupByMessageId } from "./session-registry" +import { injectReplyIntoPane, ReplyListenerRateLimiter } from "./reply-listener-injection" +import { logReplyListenerMessage } from "./reply-listener-log" +import { + recordSeenDiscordMessage, + writeReplyListenerDaemonState, + type ReplyListenerDaemonState, +} from "./reply-listener-state" +import type { OpenClawConfig } from "./types" + +interface DiscordMessage { + id: string + content: string + author: { id: string } + message_reference?: { message_id?: string } +} + +let discordBackoffUntil = 0 + +export async function pollDiscordReplies( + config: OpenClawConfig, + state: ReplyListenerDaemonState, + rateLimiter: ReplyListenerRateLimiter, +): Promise { + const replyListener = config.replyListener + if (!replyListener?.discordBotToken || !replyListener.discordChannelId) return + if (!replyListener.authorizedDiscordUserIds || replyListener.authorizedDiscordUserIds.length === 0) { + return + } + if (Date.now() < discordBackoffUntil) return + + try { + const after = state.discordLastMessageId + ? `?after=${state.discordLastMessageId}&limit=10` + : "?limit=10" + const url = `https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages${after}` + + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 10000) + const response = await fetch(url, { + method: "GET", + headers: { Authorization: `Bot ${replyListener.discordBotToken}` }, + signal: controller.signal, + }) + clearTimeout(timeout) + + const remaining = response.headers.get("x-ratelimit-remaining") + const reset = response.headers.get("x-ratelimit-reset") + if (remaining !== null && Number.parseInt(remaining, 10) < 2) { + const parsedReset = reset ? Number.parseFloat(reset) : Number.NaN + const resetTime = Number.isFinite(parsedReset) ? parsedReset * 1000 : Date.now() + 10000 + discordBackoffUntil = resetTime + logReplyListenerMessage( + `WARN: Discord rate limit low (remaining: ${remaining}), backing off until ${new Date(resetTime).toISOString()}`, + ) + } + + if (!response.ok) { + state.errors += 1 + state.lastError = `Discord API error: HTTP ${response.status}` + logReplyListenerMessage(state.lastError) + writeReplyListenerDaemonState(state) + return + } + + const messages = await response.json() + if (!Array.isArray(messages) || messages.length === 0) return + + for (const message of [...messages as DiscordMessage[]].reverse()) { + recordSeenDiscordMessage(state, message.id) + writeReplyListenerDaemonState(state) + + const replyToMessageId = message.message_reference?.message_id + if (!replyToMessageId) continue + if (!replyListener.authorizedDiscordUserIds.includes(message.author.id)) continue + + const mapping = lookupByMessageId("discord-bot", replyToMessageId) + if (!mapping) continue + + if (!rateLimiter.canProceed()) { + logReplyListenerMessage(`WARN: Rate limit exceeded, dropping Discord message ${message.id}`) + state.errors += 1 + continue + } + + const success = await injectReplyIntoPane(mapping.tmuxPaneId, message.content, "discord", config) + if (success) { + state.messagesInjected += 1 + try { + await fetch( + `https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages/${message.id}/reactions/%E2%9C%85/@me`, + { + method: "PUT", + headers: { Authorization: `Bot ${replyListener.discordBotToken}` }, + }, + ) + } catch { + } + } else { + state.errors += 1 + } + + writeReplyListenerDaemonState(state) + } + } catch (error) { + state.errors += 1 + state.lastError = error instanceof Error ? error.message : String(error) + logReplyListenerMessage(`Discord polling error: ${state.lastError}`) + } +} diff --git a/src/openclaw/reply-listener-injection.ts b/src/openclaw/reply-listener-injection.ts new file mode 100644 index 000000000..97669e9ba --- /dev/null +++ b/src/openclaw/reply-listener-injection.ts @@ -0,0 +1,74 @@ +import { removeMessagesByPane } from "./session-registry" +import { analyzePaneContent, captureTmuxPane, sendToPane } from "./tmux" +import { logReplyListenerMessage } from "./reply-listener-log" +import type { OpenClawConfig } from "./types" + +export function sanitizeReplyInput(text: string): string { + return text + .replace(/[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]/g, "") + .replace(/[\u200e\u200f\u202a-\u202e\u2066-\u2069]/g, "") + .replace(/\r?\n/g, " ") + .replace(/\\/g, "\\\\") + .replace(/`/g, "\\`") + .replace(/\$\(/g, "\\$(") + .replace(/\$\{/g, "\\${") + .trim() +} + +export class ReplyListenerRateLimiter { + private readonly maxPerMinute: number + private readonly timestamps: number[] = [] + private readonly windowMs = 60 * 1000 + + constructor(maxPerMinute: number) { + this.maxPerMinute = maxPerMinute + } + + canProceed(): boolean { + const now = Date.now() + const recent = this.timestamps.filter((timestamp) => now - timestamp < this.windowMs) + this.timestamps.length = 0 + this.timestamps.push(...recent) + + if (this.timestamps.length >= this.maxPerMinute) { + return false + } + + this.timestamps.push(now) + return true + } +} + +export async function injectReplyIntoPane( + paneId: string, + text: string, + platform: string, + config: OpenClawConfig, +): Promise { + const replyListener = config.replyListener + const content = await captureTmuxPane(paneId, 15) + const analysis = analyzePaneContent(content) + + if (analysis.confidence < 0.3) { + logReplyListenerMessage( + `WARN: Pane ${paneId} does not appear to be running OpenCode CLI (confidence: ${analysis.confidence}). Skipping injection, removing stale mapping.`, + ) + removeMessagesByPane(paneId) + return false + } + + const prefix = replyListener?.includePrefix === false ? "" : `[reply:${platform}] ` + const sanitized = sanitizeReplyInput(prefix + text) + const truncated = sanitized.slice(0, replyListener?.maxMessageLength ?? 500) + const success = await sendToPane(paneId, truncated, true) + + if (success) { + logReplyListenerMessage( + `Injected reply from ${platform} into pane ${paneId}: "${truncated.slice(0, 50)}${truncated.length > 50 ? "..." : ""}"`, + ) + } else { + logReplyListenerMessage(`ERROR: Failed to inject reply into pane ${paneId}`) + } + + return success +} diff --git a/src/openclaw/reply-listener-telegram.ts b/src/openclaw/reply-listener-telegram.ts new file mode 100644 index 000000000..e64e563a0 --- /dev/null +++ b/src/openclaw/reply-listener-telegram.ts @@ -0,0 +1,92 @@ +import { lookupByMessageId } from "./session-registry" +import { injectReplyIntoPane, ReplyListenerRateLimiter } from "./reply-listener-injection" +import { logReplyListenerMessage } from "./reply-listener-log" +import { writeReplyListenerDaemonState, type ReplyListenerDaemonState } from "./reply-listener-state" +import type { OpenClawConfig } from "./types" + +interface TelegramMessage { + message_id?: number + chat?: { id?: number | string } + text?: string + reply_to_message?: { message_id?: number } +} + +interface TelegramUpdate { + update_id?: number + message?: TelegramMessage +} + +function parseTelegramUpdatesResponse(body: unknown): TelegramUpdate[] { + if (typeof body !== "object" || body === null) return [] + const result = (body as { result?: TelegramUpdate[] }).result + return Array.isArray(result) ? result : [] +} + +export async function pollTelegramReplies( + config: OpenClawConfig, + state: ReplyListenerDaemonState, + rateLimiter: ReplyListenerRateLimiter, +): Promise { + const replyListener = config.replyListener + if (!replyListener?.telegramBotToken || !replyListener.telegramChatId) return + + try { + const offset = state.telegramLastUpdateId ? state.telegramLastUpdateId + 1 : 0 + const url = `https://api.telegram.org/bot${replyListener.telegramBotToken}/getUpdates?offset=${offset}&timeout=0` + + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 10000) + const response = await fetch(url, { method: "GET", signal: controller.signal }) + clearTimeout(timeout) + + if (!response.ok) { + logReplyListenerMessage(`Telegram API error: HTTP ${response.status}`) + return + } + + const updates = parseTelegramUpdatesResponse(await response.json()) + for (const update of updates) { + const message = update.message + state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId + writeReplyListenerDaemonState(state) + + if (!message?.reply_to_message?.message_id) continue + if (String(message.chat?.id) !== replyListener.telegramChatId) continue + if (!message.text) continue + + const mapping = lookupByMessageId("telegram", String(message.reply_to_message.message_id)) + if (!mapping) continue + + if (!rateLimiter.canProceed()) { + logReplyListenerMessage(`WARN: Rate limit exceeded, dropping Telegram message ${message.message_id}`) + state.errors += 1 + continue + } + + const success = await injectReplyIntoPane(mapping.tmuxPaneId, message.text, "telegram", config) + if (success) { + state.messagesInjected += 1 + try { + await fetch(`https://api.telegram.org/bot${replyListener.telegramBotToken}/sendMessage`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + chat_id: replyListener.telegramChatId, + text: "Injected into Codex CLI session.", + reply_to_message_id: message.message_id, + }), + }) + } catch { + } + } else { + state.errors += 1 + } + + writeReplyListenerDaemonState(state) + } + } catch (error) { + state.errors += 1 + state.lastError = error instanceof Error ? error.message : String(error) + logReplyListenerMessage(`Telegram polling error: ${state.lastError}`) + } +}