diff --git a/src/openclaw/__tests__/reply-listener.test.ts b/src/openclaw/__tests__/reply-listener.test.ts new file mode 100644 index 000000000..59fe7082d --- /dev/null +++ b/src/openclaw/__tests__/reply-listener.test.ts @@ -0,0 +1,413 @@ +import { afterAll, afterEach, beforeAll, describe, expect, mock, spyOn, test } from "bun:test" +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "fs" +import { tmpdir } from "os" +import { join } from "path" +import type { OpenClawConfig } from "../types" + +interface MockSpawnProcess { + pid: number + unref(): void +} + +type SpawnImplementation = (...args: unknown[]) => MockSpawnProcess + +const originalHome = process.env.HOME +const originalUserProfile = process.env.USERPROFILE +const originalStartupTimeout = process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS + +const tempHome = mkdtempSync(join(tmpdir(), "openclaw-reply-listener-")) +const stateDir = join(tempHome, ".omx", "state") +const configFilePath = join(stateDir, "reply-listener-config.json") +const stateFilePath = join(stateDir, "reply-listener-state.json") +const pidFilePath = join(stateDir, "reply-listener.pid") + +const livePids = new Set() +const daemonPids = new Set() + +let spawnImplementation: SpawnImplementation = () => ({ + pid: 0, + unref() { + }, +}) + +let replyListenerModule: typeof import("../reply-listener") + +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 getReplyListenerConfigSignature(config: OpenClawConfig): string { + return JSON.stringify(config.replyListener ?? null) +} + +function resetStateDir(): void { + rmSync(stateDir, { recursive: true, force: true }) + mkdirSync(stateDir, { recursive: true }) + livePids.clear() + daemonPids.clear() +} + +beforeAll(async () => { + process.env.HOME = tempHome + process.env.USERPROFILE = tempHome + + mock.module("../reply-listener-spawn", () => ({ + spawnReplyListenerDaemon: (...args: unknown[]) => spawnImplementation(...args), + })) + + mock.module("../reply-listener-process", () => ({ + isReplyListenerProcessRunning: (pid: number) => livePids.has(pid), + isReplyListenerDaemonProcess: async (pid: number) => daemonPids.has(pid), + })) + + mock.module("../tmux", () => ({ + isTmuxAvailable: async () => true, + captureTmuxPane: async () => "", + analyzePaneContent: () => ({ confidence: 1 }), + sendToPane: async () => true, + })) + + replyListenerModule = await import("../reply-listener") +}) + +afterEach(() => { + resetStateDir() + process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS = "25" +}) + +afterAll(() => { + if (originalHome === undefined) delete process.env.HOME + else process.env.HOME = originalHome + + if (originalUserProfile === undefined) delete process.env.USERPROFILE + else process.env.USERPROFILE = originalUserProfile + + if (originalStartupTimeout === undefined) { + delete process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS + } else { + process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS = originalStartupTimeout + } + + rmSync(tempHome, { recursive: true, force: true }) + mock.restore() +}) + +describe("startReplyListener", () => { + test("returns the child's ready state only after detached startup reaches the poll loop", async () => { + const killSpy = spyOn(process, "kill").mockImplementation((pid: number | string) => { + if (pid === 4321) { + return true + } + return true + }) + + spawnImplementation = () => { + const markReady = (): void => { + if (!existsSync(stateFilePath)) { + setTimeout(markReady, 5) + return + } + + const pendingState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record + writeFileSync( + stateFilePath, + JSON.stringify( + { + ...pendingState, + isRunning: true, + pid: 4321, + lastPollAt: "2026-04-07T00:00:00.000Z", + discordLastMessageId: "discord-99", + messagesSeen: 4, + }, + null, + 2, + ), + ) + } + + setTimeout(markReady, 5) + + return { + pid: 4321, + unref() { + }, + } + } + + const result = await replyListenerModule.startReplyListener(createConfig()) + + try { + expect(result.success).toBe(true) + expect(result.state).toMatchObject({ + isRunning: true, + pid: 4321, + lastPollAt: "2026-04-07T00:00:00.000Z", + discordLastMessageId: "discord-99", + lastDiscordMessageId: "discord-99", + messagesSeen: 4, + }) + + const persistedState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record + expect(persistedState.messagesSeen).toBe(4) + expect(persistedState.discordLastMessageId).toBe("discord-99") + expect(persistedState.lastDiscordMessageId).toBe("discord-99") + } finally { + killSpy.mockRestore() + } + }) + + test("does not report success or leave stale running state when detached child never becomes ready", async () => { + spawnImplementation = () => ({ + pid: 9876, + unref() { + }, + }) + + const result = await replyListenerModule.startReplyListener(createConfig()) + + expect(result.success).toBe(false) + expect(result.message).toContain("ready") + expect(existsSync(pidFilePath)).toBe(false) + + if (existsSync(stateFilePath)) { + const persistedState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record + expect(persistedState.isRunning).toBe(false) + expect(persistedState.pid).toBeNull() + } + }) + + test("does not restart an already running daemon when persisted config already matches", async () => { + const existingPid = 3210 + livePids.add(existingPid) + daemonPids.add(existingPid) + writeFileSync(pidFilePath, `${existingPid}`) + writeFileSync( + stateFilePath, + JSON.stringify({ isRunning: true, pid: existingPid, startupToken: "existing", errors: 0 }, null, 2), + ) + writeFileSync(configFilePath, JSON.stringify({ ...createConfig(), replyListener: { ...createConfig().replyListener, pollIntervalMs: 500 } }, null, 2)) + + let spawnCalls = 0 + spawnImplementation = () => { + spawnCalls += 1 + return { + pid: 9999, + unref() { + }, + } + } + + const killSpy = spyOn(process, "kill").mockImplementation(() => true) + + try { + const result = await replyListenerModule.startReplyListener(createConfig()) + + expect(result.success).toBe(true) + expect(result.message).toContain("already running") + expect(spawnCalls).toBe(0) + expect(killSpy).not.toHaveBeenCalled() + } finally { + killSpy.mockRestore() + } + }) + + test("restarts an already running daemon when persisted reply-listener config is stale", async () => { + const existingPid = 3210 + livePids.add(existingPid) + daemonPids.add(existingPid) + writeFileSync(pidFilePath, `${existingPid}`) + writeFileSync( + stateFilePath, + JSON.stringify({ isRunning: true, pid: existingPid, startupToken: "existing", errors: 0 }, null, 2), + ) + writeFileSync( + configFilePath, + JSON.stringify({ + ...createConfig(), + replyListener: { + ...createConfig().replyListener, + discordChannelId: "stale-channel", + authorizedDiscordUserIds: ["stale-user"], + pollIntervalMs: 500, + }, + }, null, 2), + ) + + const killSpy = spyOn(process, "kill").mockImplementation((pid: number | string) => { + if (typeof pid === "number") { + livePids.delete(pid) + daemonPids.delete(pid) + } + return true + }) + + let spawnCalls = 0 + spawnImplementation = () => { + spawnCalls += 1 + const nextPid = 4321 + livePids.add(nextPid) + daemonPids.add(nextPid) + + const markReady = (): void => { + if (!existsSync(stateFilePath)) { + setTimeout(markReady, 5) + return + } + + const pendingState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record + writeFileSync( + stateFilePath, + JSON.stringify( + { + ...pendingState, + isRunning: true, + pid: nextPid, + lastPollAt: "2026-04-07T00:00:00.000Z", + messagesSeen: 2, + }, + null, + 2, + ), + ) + } + + setTimeout(markReady, 5) + + return { + pid: nextPid, + unref() { + }, + } + } + + try { + const result = await replyListenerModule.startReplyListener(createConfig()) + + expect(result.success).toBe(true) + expect(spawnCalls).toBe(1) + expect(killSpy).toHaveBeenCalledWith(existingPid, "SIGTERM") + + const persistedConfig = JSON.parse(readFileSync(configFilePath, "utf-8")) as OpenClawConfig + expect(persistedConfig.replyListener?.discordChannelId).toBe("channel-1") + expect(persistedConfig.replyListener?.authorizedDiscordUserIds).toEqual(["user-1"]) + expect(persistedConfig.replyListener?.pollIntervalMs).toBe(500) + } finally { + killSpy.mockRestore() + } + }) + + test("restarts an already running daemon when runtime state config signature is stale even if persisted config matches", async () => { + const existingPid = 3210 + const matchingConfig: OpenClawConfig = { + ...createConfig(), + replyListener: { + ...createConfig().replyListener!, + pollIntervalMs: 500, + }, + } + const baseConfig = matchingConfig + const staleConfig: OpenClawConfig = { + ...baseConfig, + replyListener: { + ...baseConfig.replyListener!, + discordBotToken: "stale-token", + }, + } + + livePids.add(existingPid) + daemonPids.add(existingPid) + writeFileSync(pidFilePath, `${existingPid}`) + writeFileSync( + stateFilePath, + JSON.stringify( + { + isRunning: true, + pid: existingPid, + startupToken: "existing", + errors: 0, + configSignature: getReplyListenerConfigSignature(staleConfig), + }, + null, + 2, + ), + ) + writeFileSync(configFilePath, JSON.stringify(matchingConfig, null, 2)) + + const killSpy = spyOn(process, "kill").mockImplementation((pid: number | string) => { + if (typeof pid === "number") { + livePids.delete(pid) + daemonPids.delete(pid) + } + return true + }) + + let spawnCalls = 0 + spawnImplementation = () => { + spawnCalls += 1 + const nextPid = 4321 + livePids.add(nextPid) + daemonPids.add(nextPid) + + const markReady = (): void => { + if (!existsSync(stateFilePath)) { + setTimeout(markReady, 5) + return + } + + const pendingState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record + writeFileSync( + stateFilePath, + JSON.stringify( + { + ...pendingState, + isRunning: true, + pid: nextPid, + lastPollAt: "2026-04-07T00:00:00.000Z", + messagesSeen: 1, + }, + null, + 2, + ), + ) + } + + setTimeout(markReady, 5) + + return { + pid: nextPid, + unref() { + }, + } + } + + try { + const result = await replyListenerModule.startReplyListener(createConfig()) + + expect(result.success).toBe(true) + expect(spawnCalls).toBe(1) + expect(killSpy).toHaveBeenCalledWith(existingPid, "SIGTERM") + } finally { + killSpy.mockRestore() + } + }) +}) diff --git a/src/openclaw/reply-listener-paths.ts b/src/openclaw/reply-listener-paths.ts new file mode 100644 index 000000000..fc83b4fa7 --- /dev/null +++ b/src/openclaw/reply-listener-paths.ts @@ -0,0 +1,36 @@ +import { existsSync, mkdirSync } from "fs" +import { homedir } from "os" +import { join } from "path" + +export const REPLY_LISTENER_SECURE_FILE_MODE = 0o600 + +function resolveReplyListenerHomeDir(): string { + return process.env.HOME ?? process.env.USERPROFILE ?? homedir() +} + +export function getReplyListenerStateDir(): string { + return join(resolveReplyListenerHomeDir(), ".omx", "state") +} + +export function getReplyListenerPidFilePath(): string { + return join(getReplyListenerStateDir(), "reply-listener.pid") +} + +export function getReplyListenerStateFilePath(): string { + return join(getReplyListenerStateDir(), "reply-listener-state.json") +} + +export function getReplyListenerConfigFilePath(): string { + return join(getReplyListenerStateDir(), "reply-listener-config.json") +} + +export function getReplyListenerLogFilePath(): string { + return join(getReplyListenerStateDir(), "reply-listener.log") +} + +export function ensureReplyListenerStateDir(): void { + const stateDir = getReplyListenerStateDir() + if (!existsSync(stateDir)) { + mkdirSync(stateDir, { recursive: true, mode: 0o700 }) + } +} diff --git a/src/openclaw/reply-listener-startup.ts b/src/openclaw/reply-listener-startup.ts new file mode 100644 index 000000000..2628368e6 --- /dev/null +++ b/src/openclaw/reply-listener-startup.ts @@ -0,0 +1,60 @@ +import { randomUUID } from "crypto" +import type { ReplyListenerDaemonState } from "./reply-listener-state" + +const DEFAULT_REPLY_LISTENER_STARTUP_TIMEOUT_MS = 500 +const REPLY_LISTENER_READY_POLL_INTERVAL_MS = 10 + +interface WaitForReplyListenerReadyOptions { + pid: number + startupToken: string + timeoutMs: number + readState: () => ReplyListenerDaemonState | null + sleep: (ms: number) => Promise +} + +function isPositiveInteger(value: number): boolean { + return Number.isInteger(value) && value > 0 +} + +export function createReplyListenerStartupToken(): string { + return randomUUID() +} + +export function getReplyListenerStartupTimeoutMs(): number { + const raw = process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS + if (!raw) return DEFAULT_REPLY_LISTENER_STARTUP_TIMEOUT_MS + + const parsed = Number.parseInt(raw, 10) + return isPositiveInteger(parsed) ? parsed : DEFAULT_REPLY_LISTENER_STARTUP_TIMEOUT_MS +} + +function isReadyState( + state: ReplyListenerDaemonState | null, + pid: number, + startupToken: string, +): state is ReplyListenerDaemonState { + return Boolean( + state + && state.isRunning + && state.pid === pid + && state.startupToken === startupToken + && state.lastPollAt !== null, + ) +} + +export async function waitForReplyListenerReady( + options: WaitForReplyListenerReadyOptions, +): Promise { + const deadline = Date.now() + options.timeoutMs + + while (Date.now() <= deadline) { + const state = options.readState() + if (isReadyState(state, options.pid, options.startupToken)) { + return state + } + + await options.sleep(REPLY_LISTENER_READY_POLL_INTERVAL_MS) + } + + return null +} diff --git a/src/openclaw/reply-listener-state.ts b/src/openclaw/reply-listener-state.ts new file mode 100644 index 000000000..dcc061f94 --- /dev/null +++ b/src/openclaw/reply-listener-state.ts @@ -0,0 +1,187 @@ +import { existsSync, readFileSync, unlinkSync } from "fs" +import type { OpenClawConfig } from "./types" +import { writeSecureReplyListenerFile } from "./reply-listener-log" +import { + getReplyListenerConfigFilePath, + getReplyListenerPidFilePath, + getReplyListenerStateFilePath, +} from "./reply-listener-paths" + +export const REPLY_LISTENER_STARTUP_TOKEN_ENV = "OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TOKEN" + +export interface ReplyListenerDaemonState { + isRunning: boolean + pid: number | null + startedAt: string + startupToken: string | null + configSignature: string | null + lastPollAt: string | null + telegramLastUpdateId: number | null + discordLastMessageId: string | null + lastDiscordMessageId: string | null + messagesSeen: number + messagesInjected: number + errors: number + lastError?: string +} + +function createDefaultReplyListenerState(): ReplyListenerDaemonState { + return { + isRunning: false, + pid: null, + startedAt: new Date().toISOString(), + startupToken: null, + configSignature: null, + lastPollAt: null, + telegramLastUpdateId: null, + discordLastMessageId: null, + lastDiscordMessageId: null, + messagesSeen: 0, + messagesInjected: 0, + errors: 0, + } +} + +function isNumber(value: unknown): value is number { + return typeof value === "number" && Number.isFinite(value) +} + +function normalizeReplyListenerState(raw: unknown): ReplyListenerDaemonState { + const defaults = createDefaultReplyListenerState() + + if (typeof raw !== "object" || raw === null) { + return defaults + } + + const state = raw as Partial + return { + isRunning: state.isRunning === true, + pid: isNumber(state.pid) ? state.pid : null, + startedAt: typeof state.startedAt === "string" ? state.startedAt : defaults.startedAt, + startupToken: typeof state.startupToken === "string" ? state.startupToken : null, + configSignature: typeof state.configSignature === "string" ? state.configSignature : null, + lastPollAt: typeof state.lastPollAt === "string" ? state.lastPollAt : null, + telegramLastUpdateId: isNumber(state.telegramLastUpdateId) ? state.telegramLastUpdateId : null, + discordLastMessageId: getDiscordMessageId(state), + lastDiscordMessageId: getDiscordMessageId(state), + messagesSeen: isNumber(state.messagesSeen) ? state.messagesSeen : 0, + messagesInjected: isNumber(state.messagesInjected) ? state.messagesInjected : 0, + errors: isNumber(state.errors) ? state.errors : 0, + ...(typeof state.lastError === "string" ? { lastError: state.lastError } : {}), + } +} + +function getDiscordMessageId(state: Partial): string | null { + if (typeof state.lastDiscordMessageId === "string") { + return state.lastDiscordMessageId + } + + if (typeof state.discordLastMessageId === "string") { + return state.discordLastMessageId + } + + return null +} + +export function createPendingReplyListenerState(startupToken: string): ReplyListenerDaemonState { + return { + ...createDefaultReplyListenerState(), + startedAt: new Date().toISOString(), + startupToken, + } +} + +export function readReplyListenerDaemonState(): ReplyListenerDaemonState | null { + try { + const stateFilePath = getReplyListenerStateFilePath() + if (!existsSync(stateFilePath)) return null + return normalizeReplyListenerState(JSON.parse(readFileSync(stateFilePath, "utf-8"))) + } catch { + return null + } +} + +export function writeReplyListenerDaemonState(state: ReplyListenerDaemonState): void { + writeSecureReplyListenerFile( + getReplyListenerStateFilePath(), + JSON.stringify( + { + ...state, + lastDiscordMessageId: state.lastDiscordMessageId ?? state.discordLastMessageId, + discordLastMessageId: state.discordLastMessageId ?? state.lastDiscordMessageId, + }, + null, + 2, + ), + ) +} + +export function readReplyListenerDaemonConfig(): OpenClawConfig | null { + try { + const configFilePath = getReplyListenerConfigFilePath() + if (!existsSync(configFilePath)) return null + return JSON.parse(readFileSync(configFilePath, "utf-8")) as OpenClawConfig + } catch { + return null + } +} + +export function writeReplyListenerDaemonConfig(config: OpenClawConfig): void { + writeSecureReplyListenerFile(getReplyListenerConfigFilePath(), JSON.stringify(config, null, 2)) +} + +export function readReplyListenerPid(): number | null { + try { + const pidFilePath = getReplyListenerPidFilePath() + if (!existsSync(pidFilePath)) return null + const pid = Number.parseInt(readFileSync(pidFilePath, "utf-8").trim(), 10) + return Number.isNaN(pid) ? null : pid + } catch { + return null + } +} + +export function writeReplyListenerPid(pid: number): void { + writeSecureReplyListenerFile(getReplyListenerPidFilePath(), String(pid)) +} + +export function removeReplyListenerPid(): void { + const pidFilePath = getReplyListenerPidFilePath() + if (existsSync(pidFilePath)) { + unlinkSync(pidFilePath) + } +} + +export function getReplyListenerStartupTokenFromEnv(): string | null { + const token = process.env[REPLY_LISTENER_STARTUP_TOKEN_ENV] + return token && token.length > 0 ? token : null +} + +export function recordReplyListenerPoll(state: ReplyListenerDaemonState, pid: number): void { + state.isRunning = true + state.pid = pid + state.lastPollAt = new Date().toISOString() +} + +export function recordSeenDiscordMessage( + state: ReplyListenerDaemonState, + messageId: string, +): void { + state.discordLastMessageId = messageId + state.lastDiscordMessageId = messageId + state.messagesSeen += 1 +} + +export function markReplyListenerStopped( + state: ReplyListenerDaemonState | null, + error?: string, +): ReplyListenerDaemonState { + const nextState = state ?? createDefaultReplyListenerState() + nextState.isRunning = false + nextState.pid = null + nextState.startupToken = null + if (error) { + nextState.lastError = error + } + return nextState +} diff --git a/src/openclaw/reply-listener.ts b/src/openclaw/reply-listener.ts index 4c1f10008..77fe12453 100644 --- a/src/openclaw/reply-listener.ts +++ b/src/openclaw/reply-listener.ts @@ -1,562 +1,118 @@ -import { - existsSync, - mkdirSync, - readFileSync, - writeFileSync, - unlinkSync, - chmodSync, - statSync, - appendFileSync, - renameSync, -} from "fs" -import { join, dirname } from "path" -import { homedir } from "os" -import { spawn } from "bun" // Use bun spawn -import { captureTmuxPane, analyzePaneContent, sendToPane, isTmuxAvailable } from "./tmux" -import { lookupByMessageId, removeMessagesByPane, pruneStale } from "./session-registry" -import type { OpenClawConfig } from "./types" +import { dirname, join } from "path" import { normalizeReplyListenerConfig } from "./config" +import { pollDiscordReplies } from "./reply-listener-discord" +import { ReplyListenerRateLimiter } from "./reply-listener-injection" +import { logReplyListenerMessage } from "./reply-listener-log" +import { + isReplyListenerDaemonProcess, + isReplyListenerProcessRunning, +} from "./reply-listener-process" +import { spawnReplyListenerDaemon } from "./reply-listener-spawn" +import { ensureReplyListenerStateDir } from "./reply-listener-paths" +import { + createPendingReplyListenerState, + getReplyListenerStartupTokenFromEnv, + markReplyListenerStopped, + readReplyListenerDaemonConfig, + readReplyListenerDaemonState, + readReplyListenerPid, + recordReplyListenerPoll, + removeReplyListenerPid, + type ReplyListenerDaemonState, + writeReplyListenerDaemonConfig, + writeReplyListenerDaemonState, + writeReplyListenerPid, +} from "./reply-listener-state" +import { + createReplyListenerStartupToken, + getReplyListenerStartupTimeoutMs, + waitForReplyListenerReady, +} from "./reply-listener-startup" +import { pollTelegramReplies } from "./reply-listener-telegram" +import { pruneStale } from "./session-registry" +import { isTmuxAvailable } from "./tmux" +import type { OpenClawConfig } from "./types" -const SECURE_FILE_MODE = 0o600 -const MAX_LOG_SIZE_BYTES = 1 * 1024 * 1024 -const DAEMON_ENV_ALLOWLIST = [ - "PATH", - "HOME", - "USERPROFILE", - "USER", - "USERNAME", - "LOGNAME", - "LANG", - "LC_ALL", - "LC_CTYPE", - "TERM", - "TMUX", - "TMUX_PANE", - "TMPDIR", - "TMP", - "TEMP", - "XDG_RUNTIME_DIR", - "XDG_DATA_HOME", - "XDG_CONFIG_HOME", - "SHELL", - "NODE_ENV", - "HTTP_PROXY", - "HTTPS_PROXY", - "http_proxy", - "https_proxy", - "NO_PROXY", - "no_proxy", - "SystemRoot", - "SYSTEMROOT", - "windir", - "COMSPEC", -] +const PRUNE_INTERVAL_MS = 60 * 60 * 1000 +const REPLY_LISTENER_STOP_TIMEOUT_MS = 1_000 -const DEFAULT_STATE_DIR = join(homedir(), ".omx", "state") -const PID_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener.pid") -const STATE_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener-state.json") -const CONFIG_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener-config.json") -const LOG_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener.log") +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)) +} -export const DAEMON_IDENTITY_MARKER = "--openclaw-reply-listener-daemon" +async function terminateReplyListenerProcess(pid: number): Promise { + if (!isReplyListenerProcessRunning(pid)) return + if (!(await isReplyListenerDaemonProcess(pid))) return -function createMinimalDaemonEnv(): Record { - const env: Record = {} - for (const key of DAEMON_ENV_ALLOWLIST) { - if (process.env[key] !== undefined) { - env[key] = process.env[key] as string + try { + process.kill(pid, "SIGTERM") + } catch { + } +} + +function hasReplyListenerCredentials(config: OpenClawConfig): boolean { + return Boolean(config.replyListener?.discordBotToken || config.replyListener?.telegramBotToken) +} + +function getNormalizedReplyListenerConfig(config: OpenClawConfig): OpenClawConfig { + return normalizeReplyListenerConfig(config) +} + +function getReplyListenerRuntimeSignature(config: Pick | null): string { + return JSON.stringify(config?.replyListener ?? null) +} + +async function waitForDaemonToStop(timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs + + while (Date.now() <= deadline) { + if (!(await isDaemonRunning())) { + return true } - } - return env -} -function ensureStateDir(): void { - if (!existsSync(DEFAULT_STATE_DIR)) { - mkdirSync(DEFAULT_STATE_DIR, { recursive: true, mode: 0o700 }) - } -} - -function writeSecureFile(filePath: string, content: string): void { - ensureStateDir() - writeFileSync(filePath, content, { mode: SECURE_FILE_MODE }) - try { - chmodSync(filePath, SECURE_FILE_MODE) - } catch { - } -} - -function rotateLogIfNeeded(logPath: string): void { - try { - if (!existsSync(logPath)) return - const stats = statSync(logPath) - if (stats.size > MAX_LOG_SIZE_BYTES) { - const backupPath = `${logPath}.old` - if (existsSync(backupPath)) { - unlinkSync(backupPath) - } - renameSync(logPath, backupPath) - } - } catch { - } -} - -function log(message: string): void { - try { - ensureStateDir() - rotateLogIfNeeded(LOG_FILE_PATH) - const timestamp = new Date().toISOString() - const logLine = `[${timestamp}] ${message}\n` - appendFileSync(LOG_FILE_PATH, logLine, { mode: SECURE_FILE_MODE }) - } catch { - } -} - -export function logReplyListenerMessage(message: string): void { - log(message) -} - -interface DaemonState { - isRunning: boolean - pid: number | null - startedAt: string - lastPollAt: string | null - telegramLastUpdateId: number | null - discordLastMessageId: string | null - messagesInjected: number - errors: number - lastError?: string -} - -interface TelegramMessage { - message_id?: number - chat?: { id?: number | string } - text?: string - reply_to_message?: { message_id?: number } -} - -interface TelegramUpdate { - update_id?: number - message?: TelegramMessage -} - -interface TelegramUpdatesResponse { - result?: TelegramUpdate[] -} - -function parseTelegramUpdatesResponse(body: unknown): TelegramUpdate[] { - if (typeof body !== "object" || body === null) { - return [] + await sleep(10) } - const result = (body as TelegramUpdatesResponse).result - return Array.isArray(result) ? result : [] -} - -function readDaemonState(): DaemonState | null { - try { - if (!existsSync(STATE_FILE_PATH)) return null - const content = readFileSync(STATE_FILE_PATH, "utf-8") - return JSON.parse(content) - } catch { - return null - } -} - -function writeDaemonState(state: DaemonState): void { - writeSecureFile(STATE_FILE_PATH, JSON.stringify(state, null, 2)) -} - -function readDaemonConfig(): OpenClawConfig | null { - try { - if (!existsSync(CONFIG_FILE_PATH)) return null - const content = readFileSync(CONFIG_FILE_PATH, "utf-8") - return JSON.parse(content) - } catch { - return null - } -} - -function writeDaemonConfig(config: OpenClawConfig): void { - writeSecureFile(CONFIG_FILE_PATH, JSON.stringify(config, null, 2)) -} - -function readPidFile(): number | null { - try { - if (!existsSync(PID_FILE_PATH)) return null - const content = readFileSync(PID_FILE_PATH, "utf-8") - const pid = parseInt(content.trim(), 10) - if (Number.isNaN(pid)) return null - return pid - } catch { - return null - } -} - -function writePidFile(pid: number): void { - writeSecureFile(PID_FILE_PATH, String(pid)) -} - -function removePidFile(): void { - if (existsSync(PID_FILE_PATH)) { - unlinkSync(PID_FILE_PATH) - } -} - -function isProcessRunning(pid: number): boolean { - try { - process.kill(pid, 0) - return true - } catch { - return false - } -} - -export async function isReplyListenerProcess(pid: number): Promise { - try { - if (process.platform === "linux") { - const cmdline = readFileSync(`/proc/${pid}/cmdline`, "utf-8") - return cmdline.includes(DAEMON_IDENTITY_MARKER) - } - const proc = spawn(["ps", "-p", String(pid), "-o", "args="], { - stdout: "pipe", - stderr: "ignore", - }) - const stdout = await new Response(proc.stdout).text() - if (proc.exitCode !== 0) return false - return stdout.includes(DAEMON_IDENTITY_MARKER) - } catch { - return false - } + return !(await isDaemonRunning()) } export async function isDaemonRunning(): Promise { - const pid = readPidFile() + const pid = readReplyListenerPid() if (pid === null) return false - if (!isProcessRunning(pid)) { - removePidFile() + if (!isReplyListenerProcessRunning(pid)) { + removeReplyListenerPid() return false } - if (!(await isReplyListenerProcess(pid))) { - removePidFile() + if (!(await isReplyListenerDaemonProcess(pid))) { + removeReplyListenerPid() return false } return true } -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() -} - -class RateLimiter { - maxPerMinute: number - timestamps: number[] = [] - windowMs = 60 * 1000 - - constructor(maxPerMinute: number) { - this.maxPerMinute = maxPerMinute - } - - canProceed(): boolean { - const now = Date.now() - this.timestamps = this.timestamps.filter((t) => now - t < this.windowMs) - if (this.timestamps.length >= this.maxPerMinute) return false - this.timestamps.push(now) - return true - } -} - -async function injectReply( - 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) { // Lower threshold for simple check - log( - `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) { - log( - `Injected reply from ${platform} into pane ${paneId}: "${truncated.slice(0, 50)}${truncated.length > 50 ? "..." : ""}"`, - ) - } else { - log(`ERROR: Failed to inject reply into pane ${paneId}`) - } - return success -} - -let discordBackoffUntil = 0 - -async function pollDiscord( - config: OpenClawConfig, - state: DaemonState, - rateLimiter: RateLimiter, -): 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 && parseInt(remaining, 10) < 2) { - const parsed = reset ? parseFloat(reset) : Number.NaN - const resetTime = Number.isFinite(parsed) ? parsed * 1000 : Date.now() + 10000 - discordBackoffUntil = resetTime - log( - `WARN: Discord rate limit low (remaining: ${remaining}), backing off until ${new Date(resetTime).toISOString()}`, - ) - } - - if (!response.ok) { - log(`Discord API error: HTTP ${response.status}`) - return - } - - const messages = await response.json() - if (!Array.isArray(messages) || messages.length === 0) return - - const sorted = [...messages].reverse() - - for (const msg of sorted) { - if (!msg.message_reference?.message_id) { - state.discordLastMessageId = msg.id - writeDaemonState(state) - continue - } - - if (!replyListener.authorizedDiscordUserIds.includes(msg.author.id)) { - state.discordLastMessageId = msg.id - writeDaemonState(state) - continue - } - - const mapping = lookupByMessageId("discord-bot", msg.message_reference.message_id) - if (!mapping) { - state.discordLastMessageId = msg.id - writeDaemonState(state) - continue - } - - if (!rateLimiter.canProceed()) { - log(`WARN: Rate limit exceeded, dropping Discord message ${msg.id}`) - state.discordLastMessageId = msg.id - writeDaemonState(state) - state.errors++ - continue - } - - state.discordLastMessageId = msg.id - writeDaemonState(state) - - const success = await injectReply(mapping.tmuxPaneId, msg.content, "discord", config) - - if (success) { - state.messagesInjected++ - // Add reaction - try { - await fetch( - `https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages/${msg.id}/reactions/%E2%9C%85/@me`, - { - method: "PUT", - headers: { Authorization: `Bot ${replyListener.discordBotToken}` }, - }, - ) - } catch { - } - } else { - state.errors++ - } - } - } catch (error) { - state.errors++ - state.lastError = error instanceof Error ? error.message : String(error) - log(`Discord polling error: ${state.lastError}`) - } -} - -async function pollTelegram( - config: OpenClawConfig, - state: DaemonState, - rateLimiter: RateLimiter, -): 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) { - log(`Telegram API error: HTTP ${response.status}`) - return - } - - const body = await response.json() - const updates = parseTelegramUpdatesResponse(body) - - for (const update of updates) { - const msg = update.message - if (!msg) { - state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId - writeDaemonState(state) - continue - } - - if (msg.reply_to_message?.message_id === undefined) { - state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId - writeDaemonState(state) - continue - } - - if (String(msg.chat?.id) !== replyListener.telegramChatId) { - state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId - writeDaemonState(state) - continue - } - - const mapping = lookupByMessageId("telegram", String(msg.reply_to_message.message_id)) - if (!mapping) { - state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId - writeDaemonState(state) - continue - } - - const text = msg.text || "" - if (!text) { - state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId - writeDaemonState(state) - continue - } - - if (!rateLimiter.canProceed()) { - log(`WARN: Rate limit exceeded, dropping Telegram message ${msg.message_id}`) - state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId - writeDaemonState(state) - state.errors++ - continue - } - - state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId - writeDaemonState(state) - - const success = await injectReply(mapping.tmuxPaneId, text, "telegram", config) - - if (success) { - state.messagesInjected++ - 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: msg.message_id, - }), - }, - ) - } catch { - // Ignore - } - } else { - state.errors++ - } - } - } catch (error) { - state.errors++ - state.lastError = error instanceof Error ? error.message : String(error) - log(`Telegram polling error: ${state.lastError}`) - } -} - -const PRUNE_INTERVAL_MS = 60 * 60 * 1000 - export async function pollLoop(): Promise { - log("Reply listener daemon starting poll loop") - const config = readDaemonConfig() + logReplyListenerMessage("Reply listener daemon starting poll loop") + + const config = readReplyListenerDaemonConfig() if (!config) { - log("ERROR: No daemon config found, exiting") + logReplyListenerMessage("ERROR: No daemon config found, exiting") process.exit(1) } - const state = readDaemonState() || { - isRunning: true, - pid: process.pid, - startedAt: new Date().toISOString(), - lastPollAt: null, - telegramLastUpdateId: null, - discordLastMessageId: null, - messagesInjected: 0, - errors: 0, + const startupToken = getReplyListenerStartupTokenFromEnv() + const state = readReplyListenerDaemonState() ?? createPendingReplyListenerState(startupToken ?? "") + state.configSignature = getReplyListenerRuntimeSignature(config) + if (startupToken) { + state.startupToken = startupToken } - state.isRunning = true - state.pid = process.pid - - const rateLimiter = new RateLimiter(config.replyListener?.rateLimitPerMinute || 10) + const rateLimiter = new ReplyListenerRateLimiter(config.replyListener?.rateLimitPerMinute || 10) let lastPruneAt = Date.now() const shutdown = (): void => { - log("Shutdown signal received") - state.isRunning = false - writeDaemonState(state) - removePidFile() + logReplyListenerMessage("Shutdown signal received") + writeReplyListenerDaemonState(markReplyListenerStopped(state)) + removeReplyListenerPid() process.exit(0) } @@ -565,51 +121,96 @@ export async function pollLoop(): Promise { try { pruneStale() - log("Pruned stale registry entries") - } catch (e) { - log(`WARN: Failed to prune stale entries: ${e}`) + logReplyListenerMessage("Pruned stale registry entries") + } catch (error) { + logReplyListenerMessage( + `WARN: Failed to prune stale entries: ${error instanceof Error ? error.message : String(error)}`, + ) } - - while (state.isRunning) { + + while (state.isRunning || state.pid === null) { try { - state.lastPollAt = new Date().toISOString() - await pollDiscord(config, state, rateLimiter) - await pollTelegram(config, state, rateLimiter) - + recordReplyListenerPoll(state, process.pid) + writeReplyListenerDaemonState(state) + + await pollDiscordReplies(config, state, rateLimiter) + await pollTelegramReplies(config, state, rateLimiter) + if (Date.now() - lastPruneAt > PRUNE_INTERVAL_MS) { try { pruneStale() lastPruneAt = Date.now() - log("Pruned stale registry entries") - } catch (e) { - log(`WARN: Prune failed: ${e instanceof Error ? e.message : String(e)}`) + logReplyListenerMessage("Pruned stale registry entries") + } catch (error) { + logReplyListenerMessage( + `WARN: Prune failed: ${error instanceof Error ? error.message : String(error)}`, + ) } } - writeDaemonState(state) - await new Promise((resolve) => - setTimeout(resolve, config.replyListener?.pollIntervalMs || 3000), - ) + await sleep(config.replyListener?.pollIntervalMs || 3000) } catch (error) { - state.errors++ + state.errors += 1 state.lastError = error instanceof Error ? error.message : String(error) - log(`Poll error: ${state.lastError}`) - writeDaemonState(state) - await new Promise((resolve) => - setTimeout(resolve, (config.replyListener?.pollIntervalMs || 3000) * 2), - ) + logReplyListenerMessage(`Poll error: ${state.lastError}`) + writeReplyListenerDaemonState(state) + await sleep((config.replyListener?.pollIntervalMs || 3000) * 2) } } - log("Poll loop ended") + + logReplyListenerMessage("Poll loop ended") } -export async function startReplyListener(config: OpenClawConfig): Promise<{ success: boolean; message: string; state?: DaemonState; error?: string }> { - if (await isDaemonRunning()) { - const state = readDaemonState() +function createStartFailureResult( + message: string, + state: ReplyListenerDaemonState, +): { success: false; message: string; state: ReplyListenerDaemonState } { + return { + success: false, + message, + state, + } +} + +export async function startReplyListener( + config: OpenClawConfig, +): Promise<{ success: boolean; message: string; state?: ReplyListenerDaemonState; error?: string }> { + const normalizedConfig = getNormalizedReplyListenerConfig(config) + const replyListener = normalizedConfig.replyListener + if (!replyListener?.discordBotToken && !replyListener?.telegramBotToken) { return { - success: true, - message: "Reply listener daemon is already running", - state: state || undefined, + success: false, + message: "No enabled reply listener platforms configured (missing bot tokens/channels)", + } + } + + if (await isDaemonRunning()) { + const state = readReplyListenerDaemonState() + const runtimeSignature = state?.configSignature ?? getReplyListenerRuntimeSignature(readReplyListenerDaemonConfig()) + if (runtimeSignature === getReplyListenerRuntimeSignature(normalizedConfig)) { + return { + success: true, + message: "Reply listener daemon is already running", + state: state || undefined, + } + } + + const stopResult = await stopReplyListener() + if (!stopResult.success) { + return { + success: false, + message: "Failed to restart reply listener daemon", + state: stopResult.state, + error: stopResult.error ?? stopResult.message, + } + } + + if (!(await waitForDaemonToStop(REPLY_LISTENER_STOP_TIMEOUT_MS))) { + return { + success: false, + message: "Timed out waiting for reply listener daemon to stop before restart", + state: readReplyListenerDaemonState() || undefined, + } } } @@ -620,108 +221,117 @@ export async function startReplyListener(config: OpenClawConfig): Promise<{ succ } } - const normalizedConfig = normalizeReplyListenerConfig(config) - const replyListener = normalizedConfig.replyListener - if (!replyListener?.discordBotToken && !replyListener?.telegramBotToken) { - return { - success: false, - message: "No enabled reply listener platforms configured (missing bot tokens/channels)", - } - } + ensureReplyListenerStateDir() + writeReplyListenerDaemonConfig(normalizedConfig) - writeDaemonConfig(normalizedConfig) - ensureStateDir() + const startupToken = createReplyListenerStartupToken() + const pendingState = createPendingReplyListenerState(startupToken) + pendingState.configSignature = getReplyListenerRuntimeSignature(normalizedConfig) + writeReplyListenerDaemonState(pendingState) const currentFile = import.meta.url - const isTs = currentFile.endsWith(".ts") - const daemonScript = isTs + const daemonScript = currentFile.endsWith(".ts") ? join(dirname(new URL(currentFile).pathname), "daemon.ts") : join(dirname(new URL(currentFile).pathname), "daemon.js") try { - const proc = spawn(["bun", "run", daemonScript, DAEMON_IDENTITY_MARKER], { - detached: true, - stdio: ["ignore", "ignore", "ignore"], - cwd: process.cwd(), - env: createMinimalDaemonEnv(), - }) - - proc.unref() - const pid = proc.pid - - if (pid) { - writePidFile(pid) - const state: DaemonState = { - isRunning: true, - pid, - startedAt: new Date().toISOString(), - lastPollAt: null, - telegramLastUpdateId: null, - discordLastMessageId: null, - messagesInjected: 0, - errors: 0, - } - writeDaemonState(state) - log(`Reply listener daemon started with PID ${pid}`) - return { - success: true, - message: `Reply listener daemon started with PID ${pid}`, - state, - } + const processInfo = spawnReplyListenerDaemon(daemonScript, startupToken) + + processInfo.unref() + + if (!processInfo.pid) { + const stoppedState = markReplyListenerStopped(pendingState, "Failed to start daemon process") + writeReplyListenerDaemonState(stoppedState) + return createStartFailureResult("Failed to start daemon process", stoppedState) } - + + writeReplyListenerPid(processInfo.pid) + + const readyState = await waitForReplyListenerReady({ + pid: processInfo.pid, + startupToken, + timeoutMs: getReplyListenerStartupTimeoutMs(), + readState: readReplyListenerDaemonState, + sleep, + }) + + if (!readyState) { + await terminateReplyListenerProcess(processInfo.pid) + removeReplyListenerPid() + const stoppedState = markReplyListenerStopped( + readReplyListenerDaemonState() ?? pendingState, + `Reply listener daemon did not become ready within ${getReplyListenerStartupTimeoutMs()}ms`, + ) + writeReplyListenerDaemonState(stoppedState) + return createStartFailureResult( + `Reply listener daemon did not become ready within ${getReplyListenerStartupTimeoutMs()}ms`, + stoppedState, + ) + } + + writeReplyListenerDaemonState(readyState) + logReplyListenerMessage(`Reply listener daemon started with PID ${processInfo.pid}`) return { - success: false, - message: "Failed to start daemon process", + success: true, + message: `Reply listener daemon started with PID ${processInfo.pid}`, + state: readyState, } } catch (error) { + const stoppedState = markReplyListenerStopped( + readReplyListenerDaemonState() ?? pendingState, + error instanceof Error ? error.message : String(error), + ) + writeReplyListenerDaemonState(stoppedState) + removeReplyListenerPid() return { success: false, message: "Failed to start daemon", + state: stoppedState, error: error instanceof Error ? error.message : String(error), } } } -export async function stopReplyListener(): Promise<{ success: boolean; message: string; state?: DaemonState; error?: string }> { - const pid = readPidFile() +export async function stopReplyListener(): Promise<{ + success: boolean + message: string + state?: ReplyListenerDaemonState + error?: string +}> { + const pid = readReplyListenerPid() if (pid === null) { return { success: true, message: "Reply listener daemon is not running", } } - - if (!isProcessRunning(pid)) { - removePidFile() + + if (!isReplyListenerProcessRunning(pid)) { + removeReplyListenerPid() return { success: true, message: "Reply listener daemon was not running (cleaned up stale PID file)", } } - - if (!(await isReplyListenerProcess(pid))) { - removePidFile() + + if (!(await isReplyListenerDaemonProcess(pid))) { + removeReplyListenerPid() return { success: false, message: `Refusing to kill PID ${pid}: process identity does not match the reply listener daemon (stale or reused PID - removed PID file)`, } } - + try { process.kill(pid, "SIGTERM") - removePidFile() - const state = readDaemonState() - if (state) { - state.isRunning = false - state.pid = null - writeDaemonState(state) - } - log(`Reply listener daemon stopped (PID ${pid})`) + removeReplyListenerPid() + const state = markReplyListenerStopped(readReplyListenerDaemonState()) + writeReplyListenerDaemonState(state) + logReplyListenerMessage(`Reply listener daemon stopped (PID ${pid})`) return { success: true, message: `Reply listener daemon stopped (PID ${pid})`, - state: state || undefined, + state, } } catch (error) { return { @@ -731,3 +341,5 @@ export async function stopReplyListener(): Promise<{ success: boolean; message: } } } + +export { logReplyListenerMessage }