fix(openclaw): stabilize reply polling and tmux injection

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")
+  })
+})

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<void> {
+  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}`)
+  }
+}

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<boolean> {
+  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
+}

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<void> {
+  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}`)
+  }
+}