fix(openclaw): split reply listener state and startup flow
This commit is contained in:
GeonWoo Jeon (Jay)
@@ -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<number>()
|
||||
const daemonPids = new Set<number>()
|
||||
|
||||
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<string, unknown>
|
||||
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<string, unknown>
|
||||
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<string, unknown>
|
||||
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<string, unknown>
|
||||
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<string, unknown>
|
||||
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()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -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 })
|
||||
}
|
||||
}
|
||||
@@ -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<void>
|
||||
}
|
||||
|
||||
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<ReplyListenerDaemonState | null> {
|
||||
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
|
||||
}
|
||||
@@ -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<ReplyListenerDaemonState>
|
||||
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<ReplyListenerDaemonState>): 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
|
||||
}
|
||||
+232
-620
@@ -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<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms))
|
||||
}
|
||||
|
||||
export const DAEMON_IDENTITY_MARKER = "--openclaw-reply-listener-daemon"
|
||||
async function terminateReplyListenerProcess(pid: number): Promise<void> {
|
||||
if (!isReplyListenerProcessRunning(pid)) return
|
||||
if (!(await isReplyListenerDaemonProcess(pid))) return
|
||||
|
||||
function createMinimalDaemonEnv(): Record<string, string> {
|
||||
const env: Record<string, string> = {}
|
||||
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<OpenClawConfig, "replyListener"> | null): string {
|
||||
return JSON.stringify(config?.replyListener ?? null)
|
||||
}
|
||||
|
||||
async function waitForDaemonToStop(timeoutMs: number): Promise<boolean> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
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<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 && 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<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) {
|
||||
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<void> {
|
||||
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<void> {
|
||||
|
||||
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 }
|
||||
|
||||
Reference in New Issue
Block a user