diff --git a/script/run-ci-tests.ts b/script/run-ci-tests.ts index 10cf80b21..116d5e4ff 100644 --- a/script/run-ci-tests.ts +++ b/script/run-ci-tests.ts @@ -8,6 +8,7 @@ type CiTestPlan = { const TEST_ROOTS = ["bin", "script", "src"] as const const MODULE_MOCK_PATTERN = "mock.module(" +const ALWAYS_ISOLATED_TEST_FILES = ["src/openclaw/__tests__/reply-listener-discord.test.ts"] as const async function collectTestFiles(rootDirectory: string): Promise { const testFiles: string[] = [] @@ -54,8 +55,11 @@ export async function createCiTestPlan(rootDirectory: string = process.cwd()): P } } + const isolatedTestFiles = Array.from( + new Set([...isolatedModuleMockFiles, ...ALWAYS_ISOLATED_TEST_FILES.filter((testFile) => allTestFiles.includes(testFile))]), + ) const isolatedTestTargets = collapseNestedTargets( - Array.from(new Set(isolatedModuleMockFiles.map((testFile) => toIsolatedTarget(testFile)))).sort((left, right) => + isolatedTestFiles.map((testFile) => toIsolatedTarget(testFile)).sort((left, right) => left.localeCompare(right), ), ) diff --git a/src/create-managers.test.ts b/src/create-managers.test.ts index e246e1dcb..94a18460d 100644 --- a/src/create-managers.test.ts +++ b/src/create-managers.test.ts @@ -1,23 +1,66 @@ /// -import { beforeEach, describe, expect, it, mock } from "bun:test" +import { afterAll, beforeEach, describe, expect, it, mock, spyOn } from "bun:test" -import { createManagers } from "./create-managers" +import * as openclawRuntimeDispatch from "./openclaw/runtime-dispatch" -class MockBackgroundManager { - constructor(..._args: unknown[]) {} -} +const markServerRunningInProcess = mock(() => {}) +let backgroundManagerOptions: { + onSubagentSessionCreated?: (event: { sessionID: string; parentID: string; title: string }) => Promise +} | null = null +const trackedPaneBySession = new Map() -class MockSkillMcpManager { - constructor(..._args: unknown[]) {} -} +mock.module("./features/background-agent", () => ({ + BackgroundManager: class BackgroundManager { + constructor(_ctx: unknown, _config: unknown, options: typeof backgroundManagerOptions) { + backgroundManagerOptions = options + } + }, +})) -class MockTmuxSessionManager { - constructor(..._args: unknown[]) {} +mock.module("./features/skill-mcp-manager", () => ({ + SkillMcpManager: class SkillMcpManager { + constructor(..._args: unknown[]) {} + }, +})) - async cleanup(): Promise {} - async onSessionCreated(..._args: unknown[]): Promise {} -} +mock.module("./features/task-toast-manager", () => ({ + initTaskToastManager: mock(() => {}), +})) + +mock.module("./features/tmux-subagent", () => ({ + TmuxSessionManager: class TmuxSessionManager { + constructor(..._args: unknown[]) {} + + async cleanup(): Promise {} + async onSessionCreated(event: { properties?: { info?: { id?: string } } }): Promise { + const sessionID = event.properties?.info?.id + if (sessionID) { + trackedPaneBySession.set(sessionID, `%pane-${sessionID}`) + } + } + + getTrackedPaneId(sessionID: string): string | undefined { + return trackedPaneBySession.get(sessionID) + } + }, +})) + +mock.module("./features/background-agent/process-cleanup", () => ({ + registerManagerForCleanup: mock(() => {}), +})) + +mock.module("./plugin-handlers", () => ({ + createConfigHandler: mock(() => ({ kind: "config-handler" })), +})) + +mock.module("./shared/tmux/tmux-utils/server-health", () => ({ + isServerRunning: mock(async () => true), + markServerRunningInProcess, + resetServerCheck: mock(() => {}), +})) + +const { createManagers } = await import("./create-managers") function createTmuxConfig(enabled: boolean) { return { @@ -31,63 +74,79 @@ function createTmuxConfig(enabled: boolean) { } describe("createManagers", () => { - const markServerRunningInProcess = mock(() => {}) - const initTaskToastManager = mock(() => ({}) as never) - const registerManagerForCleanup = mock(() => {}) - const createConfigHandler = mock(() => (async () => {}) as never) - - function createMockArgs(enabled: boolean): Parameters[0] { - return { - ctx: { - directory: "/tmp", - client: {} as never, - project: {} as never, - worktree: "/tmp", - serverUrl: new URL("https://example.com"), - $: Bun.$, - }, - pluginConfig: {} as never, - tmuxConfig: createTmuxConfig(enabled), - modelCacheState: {} as never, - backgroundNotificationHookEnabled: false, - deps: { - BackgroundManagerClass: MockBackgroundManager as never, - SkillMcpManagerClass: MockSkillMcpManager as never, - TmuxSessionManagerClass: MockTmuxSessionManager as never, - initTaskToastManagerFn: initTaskToastManager, - registerManagerForCleanupFn: registerManagerForCleanup, - createConfigHandlerFn: createConfigHandler, - markServerRunningInProcessFn: markServerRunningInProcess, - }, - } - } + const dispatchOpenClawEvent = spyOn(openclawRuntimeDispatch, "dispatchOpenClawEvent") beforeEach(() => { markServerRunningInProcess.mockClear() - initTaskToastManager.mockClear() - registerManagerForCleanup.mockClear() - createConfigHandler.mockClear() + dispatchOpenClawEvent.mockReset() + backgroundManagerOptions = null + trackedPaneBySession.clear() + }) + + afterAll(() => { + mock.restore() }) it("#given tmux integration is disabled #when managers are created #then it does not mark the tmux server as running", () => { - // #given - const args = createMockArgs(false) + const args = { + ctx: { directory: "/tmp", client: {} }, + pluginConfig: {}, + tmuxConfig: createTmuxConfig(false), + modelCacheState: {}, + backgroundNotificationHookEnabled: false, + } as Parameters[0] - // #when createManagers(args) - // #then expect(markServerRunningInProcess).not.toHaveBeenCalled() }) it("#given tmux integration is enabled #when managers are created #then it marks the tmux server as running", () => { - // #given - const args = createMockArgs(true) + const args = { + ctx: { directory: "/tmp", client: {} }, + pluginConfig: {}, + tmuxConfig: createTmuxConfig(true), + modelCacheState: {}, + backgroundNotificationHookEnabled: false, + } as Parameters[0] - // #when createManagers(args) - // #then expect(markServerRunningInProcess).toHaveBeenCalledTimes(1) }) + + it("#given openclaw is enabled #when the background session-created callback runs #then it dispatches openclaw with the tracked pane id", async () => { + const args = { + ctx: { directory: "/tmp/project", client: {} }, + pluginConfig: { + openclaw: { + enabled: true, + gateways: {}, + hooks: {}, + }, + }, + tmuxConfig: createTmuxConfig(true), + modelCacheState: {}, + backgroundNotificationHookEnabled: false, + } as Parameters[0] + + createManagers(args) + + await backgroundManagerOptions?.onSubagentSessionCreated?.({ + sessionID: "ses-bg-1", + parentID: "ses-parent", + title: "child task", + }) + + expect(dispatchOpenClawEvent).toHaveBeenCalledTimes(1) + expect(dispatchOpenClawEvent).toHaveBeenCalledWith({ + config: args.pluginConfig.openclaw, + rawEvent: "session.created", + context: { + sessionId: "ses-bg-1", + projectPath: "/tmp/project", + tmuxPaneId: "%pane-ses-bg-1", + }, + }) + }) }) diff --git a/src/create-managers.ts b/src/create-managers.ts index 51b93270e..dc2a99a98 100644 --- a/src/create-managers.ts +++ b/src/create-managers.ts @@ -7,6 +7,7 @@ import { BackgroundManager } from "./features/background-agent" import { SkillMcpManager } from "./features/skill-mcp-manager" import { initTaskToastManager } from "./features/task-toast-manager" import { TmuxSessionManager } from "./features/tmux-subagent" +import * as openclawRuntimeDispatch from "./openclaw/runtime-dispatch" import { registerManagerForCleanup } from "./features/background-agent/process-cleanup" import { createConfigHandler } from "./plugin-handlers" import { log } from "./shared" @@ -86,6 +87,18 @@ export function createManagers(args: { }, }) + if (pluginConfig.openclaw) { + await openclawRuntimeDispatch.dispatchOpenClawEvent({ + config: pluginConfig.openclaw, + rawEvent: "session.created", + context: { + sessionId: event.sessionID, + projectPath: ctx.directory, + tmuxPaneId: tmuxSessionManager.getTrackedPaneId?.(event.sessionID) ?? process.env.TMUX_PANE, + }, + }) + } + log("[index] onSubagentSessionCreated callback completed") }, onShutdown: async () => { diff --git a/src/features/tmux-subagent/manager.ts b/src/features/tmux-subagent/manager.ts index 7bab2d215..a31f668bf 100644 --- a/src/features/tmux-subagent/manager.ts +++ b/src/features/tmux-subagent/manager.ts @@ -182,6 +182,10 @@ export class TmuxSessionManager { })) } + getTrackedPaneId(sessionId: string): string | undefined { + return this.sessions.get(sessionId)?.paneId + } + private removeTrackedSession(sessionId: string): void { this.sessions.delete(sessionId) diff --git a/src/features/tmux-subagent/zombie-pane.test.ts b/src/features/tmux-subagent/zombie-pane.test.ts index 267c03cb3..d908e0b9c 100644 --- a/src/features/tmux-subagent/zombie-pane.test.ts +++ b/src/features/tmux-subagent/zombie-pane.test.ts @@ -1,3 +1,4 @@ +/// import { beforeEach, describe, expect, mock, test, afterAll } from "bun:test" import type { TmuxConfig } from "../../config/schema" import type { ActionResult, ExecuteContext, ExecuteActionsResult } from "./action-executor" @@ -25,6 +26,9 @@ const mockExecuteActions = mock<( results: [], })) +const mockSpawnTmuxWindow = mock(async () => ({ success: true, paneId: "%window" })) +const mockSpawnTmuxSession = mock(async () => ({ success: true, paneId: "%session" })) + const mockIsInsideTmux = mock<() => boolean>(() => true) const mockGetCurrentPaneId = mock<() => string | undefined>(() => "%0") @@ -47,14 +51,14 @@ mock.module("../../shared/tmux", () => ({ spawnTmuxPane: mock(async () => ({ success: true, paneId: "%1" })), closeTmuxPane: mock(async () => ({ success: true })), replaceTmuxPane: mock(async () => ({ success: true, paneId: "%1" })), - spawnTmuxWindow: mock(async () => ({ success: true, windowId: "@1" })), - spawnTmuxSession: mock(async () => ({ success: true, sessionId: "mock" })), applyLayout: mock(async () => ({ success: true })), enforceMainPaneWidth: mock(async () => ({ success: true })), POLL_INTERVAL_BACKGROUND_MS: 10, SESSION_READY_POLL_INTERVAL_MS: 10, SESSION_READY_TIMEOUT_MS: 50, SESSION_MISSING_GRACE_MS: 1_000, + spawnTmuxWindow: mockSpawnTmuxWindow, + spawnTmuxSession: mockSpawnTmuxSession, SESSION_TIMEOUT_MS: 600_000, })) @@ -68,6 +72,7 @@ const mockTmuxDeps: TmuxUtilDeps = { function createConfig(): TmuxConfig { return { enabled: true, + isolation: "inline", layout: "main-vertical", main_pane_size: 60, main_pane_min_width: 80, @@ -168,6 +173,8 @@ describe("TmuxSessionManager zombie pane handling", () => { mockQueryWindowState.mockClear() mockExecuteAction.mockClear() mockExecuteActions.mockClear() + mockSpawnTmuxWindow.mockClear() + mockSpawnTmuxSession.mockClear() mockIsInsideTmux.mockClear() mockGetCurrentPaneId.mockClear() @@ -183,6 +190,8 @@ describe("TmuxSessionManager zombie pane handling", () => { spawnedPaneId: "%1", results: [], })) + mockSpawnTmuxWindow.mockImplementation(async () => ({ success: true, paneId: "%window" })) + mockSpawnTmuxSession.mockImplementation(async () => ({ success: true, paneId: "%session" })) mockIsInsideTmux.mockReturnValue(true) mockGetCurrentPaneId.mockReturnValue("%0") }) @@ -271,9 +280,15 @@ describe("TmuxSessionManager zombie pane handling", () => { "ses_pending", createTrackedSession({ closePending: true, closeRetryCount: 0 }), ) - mockExecuteAction.mockImplementationOnce(async () => { - sessions.delete("ses_pending") - return { success: false } + let shouldFailClose = true + mockExecuteAction.mockImplementation(async () => { + if (shouldFailClose) { + shouldFailClose = false + sessions.delete("ses_pending") + return { success: false } + } + + return { success: true } }) // when diff --git a/src/index.test.ts b/src/index.test.ts index 0082d1e2c..e9d4baee3 100644 --- a/src/index.test.ts +++ b/src/index.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, mock } from "bun:test" +import { afterAll, beforeEach, describe, expect, it, mock } from "bun:test" describe("experimental.session.compacting handler", () => { function createCompactingHandler(hooks: { @@ -217,3 +217,181 @@ describe("look_at tool conditional registration", () => { }) }) }) + +const mockInitConfigContext = mock(() => {}) +const mockDetectExternalSkillPlugin = mock(() => ({ detected: false, pluginName: null })) +const mockGetSkillPluginConflictWarning = mock(() => "") +const mockInjectServerAuthIntoClient = mock(() => {}) +const mockLogLegacyPluginStartupWarning = mock(() => {}) +const mockLoadPluginConfig = mock(() => ({})) +const mockIsTmuxIntegrationEnabled = mock( + (pluginConfig: { tmux?: { enabled?: boolean } | undefined }) => pluginConfig.tmux?.enabled ?? false, +) +const mockIsInteractiveBashEnabled = mock(() => false) +const mockCreateRuntimeTmuxConfig = mock(() => ({ + enabled: false, + layout: "tiled" as const, + main_pane_size: 60, + main_pane_min_width: 80, + agent_pane_min_width: 40, + isolation: "inline" as const, +})) +const mockCreateManagers = mock(() => ({ + backgroundManager: { shutdown: async () => {} }, + skillMcpManager: { disconnectAll: async () => {} }, + configHandler: async () => {}, +})) +const mockCreateTools = mock(async () => ({ + mergedSkills: [], + availableSkills: [], + filteredTools: {}, +})) +const mockCreateHooks = mock(() => ({ + disposeHooks: () => {}, + compactionContextInjector: undefined, + compactionTodoPreserver: undefined, + claudeCodeHooks: undefined, +})) +const mockCreatePluginDispose = mock(() => async () => {}) +const mockCreatePluginInterface = mock(() => ({})) +const mockInitializeOpenClaw = mock(async () => {}) +const mockStartTmuxCheck = mock(() => {}) + +mock.module("./cli/config-manager/config-context", () => ({ + initConfigContext: mockInitConfigContext, +})) + +mock.module("./shared/external-plugin-detector", () => ({ + detectExternalSkillPlugin: mockDetectExternalSkillPlugin, + getSkillPluginConflictWarning: mockGetSkillPluginConflictWarning, +})) + +mock.module("./shared", () => ({ + injectServerAuthIntoClient: mockInjectServerAuthIntoClient, + log: mock(() => {}), + logLegacyPluginStartupWarning: mockLogLegacyPluginStartupWarning, +})) + +mock.module("./plugin-config", () => ({ + loadPluginConfig: mockLoadPluginConfig, +})) + +mock.module("./create-runtime-tmux-config", () => ({ + createRuntimeTmuxConfig: mockCreateRuntimeTmuxConfig, + isTmuxIntegrationEnabled: mockIsTmuxIntegrationEnabled, + isInteractiveBashEnabled: mockIsInteractiveBashEnabled, +})) + +mock.module("./create-managers", () => ({ + createManagers: mockCreateManagers, +})) + +mock.module("./create-tools", () => ({ + createTools: mockCreateTools, +})) + +mock.module("./create-hooks", () => ({ + createHooks: mockCreateHooks, +})) + +mock.module("./plugin-dispose", () => ({ + createPluginDispose: mockCreatePluginDispose, +})) + +mock.module("./plugin-interface", () => ({ + createPluginInterface: mockCreatePluginInterface, +})) + +mock.module("./plugin-state", () => ({ + createModelCacheState: mock(() => ({})), +})) + +mock.module("./shared/first-message-variant", () => ({ + createFirstMessageVariantGate: mock(() => ({ + shouldOverride: () => false, + markApplied: () => {}, + markSessionCreated: () => {}, + clear: () => {}, + })), +})) + +mock.module("./openclaw", () => ({ + initializeOpenClaw: mockInitializeOpenClaw, +})) + +mock.module("./tools/interactive-bash", () => ({ + interactive_bash: {}, + startBackgroundCheck: mockStartTmuxCheck, +})) + +mock.module("./tools/lsp/client", () => ({ + lspManager: { + cleanupTempDirectoryClients: async () => {}, + }, +})) + +const { default: OhMyOpenCodePlugin } = await import("./index") + +describe("OhMyOpenCodePlugin", () => { + beforeEach(() => { + mockInitConfigContext.mockClear() + mockDetectExternalSkillPlugin.mockClear() + mockGetSkillPluginConflictWarning.mockClear() + mockInjectServerAuthIntoClient.mockClear() + mockLogLegacyPluginStartupWarning.mockClear() + mockLoadPluginConfig.mockClear() + mockIsTmuxIntegrationEnabled.mockClear() + mockIsInteractiveBashEnabled.mockClear() + mockCreateRuntimeTmuxConfig.mockClear() + mockCreateManagers.mockClear() + mockCreateTools.mockClear() + mockCreateHooks.mockClear() + mockCreatePluginDispose.mockClear() + mockCreatePluginInterface.mockClear() + mockInitializeOpenClaw.mockClear() + mockStartTmuxCheck.mockClear() + }) + + afterAll(() => { + mock.restore() + }) + + it("starts openclaw during plugin bootstrap when openclaw config exists", async () => { + // given + const openclawConfig = { + enabled: true, + gateways: {}, + hooks: {}, + replyListener: { + discordBotToken: "discord-token", + }, + } + mockLoadPluginConfig.mockReturnValue({ + openclaw: openclawConfig, + }) + + // when + await OhMyOpenCodePlugin({ + directory: "/tmp/project", + client: {}, + } as Parameters[0]) + + // then + expect(mockInitializeOpenClaw).toHaveBeenCalledTimes(1) + expect(mockInitializeOpenClaw).toHaveBeenCalledWith(openclawConfig) + }) + + it("does not start openclaw when openclaw config is absent", async () => { + // given + mockLoadPluginConfig.mockReturnValue({}) + + // when + await OhMyOpenCodePlugin({ + directory: "/tmp/project", + client: {}, + } as Parameters[0]) + + // then + expect(mockInitializeOpenClaw).not.toHaveBeenCalled() + }) +}) diff --git a/src/index.ts b/src/index.ts index 4ad88a870..f13faf1ec 100644 --- a/src/index.ts +++ b/src/index.ts @@ -7,6 +7,7 @@ import { createHooks } from "./create-hooks" import { createManagers } from "./create-managers" import { createRuntimeTmuxConfig, isTmuxIntegrationEnabled } from "./create-runtime-tmux-config" import { createTools } from "./create-tools" +import { initializeOpenClaw } from "./openclaw" import { createPluginInterface } from "./plugin-interface" import { createPluginDispose, type PluginDispose } from "./plugin-dispose" @@ -15,8 +16,8 @@ import { createModelCacheState } from "./plugin-state" import { createFirstMessageVariantGate } from "./shared/first-message-variant" import { injectServerAuthIntoClient, log, logLegacyPluginStartupWarning } from "./shared" import { detectExternalSkillPlugin, getSkillPluginConflictWarning } from "./shared/external-plugin-detector" +import { startBackgroundCheck as startTmuxCheck } from "./tools/interactive-bash" import { lspManager } from "./tools/lsp/client" -import { startTmuxCheck } from "./tools" let activePluginDispose: PluginDispose | null = null @@ -36,6 +37,9 @@ const OhMyOpenCodePlugin: Plugin = async (ctx) => { await activePluginDispose?.() const pluginConfig = loadPluginConfig(ctx.directory, ctx) + if (pluginConfig.openclaw) { + await initializeOpenClaw(pluginConfig.openclaw) + } const tmuxIntegrationEnabled = isTmuxIntegrationEnabled(pluginConfig) if (tmuxIntegrationEnabled) { startTmuxCheck() diff --git a/src/openclaw/__tests__/dispatcher.test.ts b/src/openclaw/__tests__/dispatcher.test.ts index 62a467abb..96bed7335 100644 --- a/src/openclaw/__tests__/dispatcher.test.ts +++ b/src/openclaw/__tests__/dispatcher.test.ts @@ -54,6 +54,75 @@ describe("OpenClaw Dispatcher", () => { } }) + test("wakeGateway returns correlation metadata from JSON response", async () => { + const fetchSpy = spyOn(global, "fetch").mockResolvedValue( + new Response( + JSON.stringify({ + data: { + messageId: "msg-123", + platform: "discord", + channelId: "chan-1", + threadId: "thread-9", + }, + }), + { status: 200 }, + ), + ) + + try { + const result = await wakeGateway( + "test", + { url: "https://example.com", method: "POST", timeout: 1000, type: "http" }, + { foo: "bar" }, + ) + + expect(result).toMatchObject({ + success: true, + messageId: "msg-123", + platform: "discord", + channelId: "chan-1", + threadId: "thread-9", + }) + } finally { + fetchSpy.mockRestore() + } + }) + + test("wakeGateway prefers nested message metadata over wrapper ids", async () => { + const fetchSpy = spyOn(global, "fetch").mockResolvedValue( + new Response( + JSON.stringify({ + id: "job-42", + data: { + messageId: "msg-123", + platform: "discord", + channelId: "chan-1", + threadId: "thread-9", + }, + }), + { status: 200 }, + ), + ) + + try { + const result = await wakeGateway( + "test", + { url: "https://example.com", method: "POST", timeout: 1000, type: "http" }, + { foo: "bar" }, + ) + + expect(result).toMatchObject({ + success: true, + messageId: "msg-123", + platform: "discord", + channelId: "chan-1", + threadId: "thread-9", + }) + } finally { + fetchSpy.mockRestore() + } + }) + test("wakeGateway fails on invalid URL", async () => { const result = await wakeGateway("test", { url: "http://example.com", method: "POST", timeout: 1000, type: "http" }, {}) expect(result.success).toBe(false) @@ -108,4 +177,43 @@ describe("OpenClaw Dispatcher", () => { killSpy.mockRestore() } }) + + test("wakeCommandGateway returns correlation metadata from stdout JSON", async () => { + const result = await wakeCommandGateway( + "command", + { + type: "command", + method: "POST", + command: "printf '%s' '{\"messageId\":\"55\",\"platform\":\"telegram\",\"threadId\":\"thr\"}'", + timeout: 1000, + }, + {}, + ) + + expect(result).toMatchObject({ + success: true, + messageId: "55", + platform: "telegram", + threadId: "thr", + }) + }) + + test("wakeCommandGateway returns correlation metadata from OpenClaw CLI stdout", async () => { + const result = await wakeCommandGateway( + "command", + { + type: "command", + method: "POST", + command: "printf '%s' '✅ Sent via Discord. Message ID: 55'", + timeout: 1000, + }, + {}, + ) + + expect(result).toMatchObject({ + success: true, + messageId: "55", + platform: "discord", + }) + }) }) diff --git a/src/openclaw/__tests__/reply-listener-discord.test.ts b/src/openclaw/__tests__/reply-listener-discord.test.ts new file mode 100644 index 000000000..d1bdba01e --- /dev/null +++ b/src/openclaw/__tests__/reply-listener-discord.test.ts @@ -0,0 +1,134 @@ +import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from "bun:test" +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync } from "fs" +import { tmpdir } from "os" +import { join } from "path" +import { ReplyListenerRateLimiter } from "../reply-listener-injection" +import { pollDiscordReplies } from "../reply-listener-discord" +import * as injectionModule from "../reply-listener-injection" +import * as sessionRegistryModule from "../session-registry" +import type { ReplyListenerDaemonState } from "../reply-listener-state" +import type { OpenClawConfig } from "../types" + +const originalHome = process.env.HOME +const originalUserProfile = process.env.USERPROFILE +const originalFetch = globalThis.fetch + +const tempHome = mkdtempSync(join(tmpdir(), "openclaw-reply-listener-discord-")) +const stateDir = join(tempHome, ".omx", "state") +const stateFilePath = join(stateDir, "reply-listener-state.json") + +function createConfig(): OpenClawConfig { + return { + enabled: true, + gateways: { + gateway: { + type: "http", + url: "https://example.com", + method: "POST", + }, + }, + hooks: {}, + replyListener: { + discordBotToken: "discord-token", + discordChannelId: "channel-1", + authorizedDiscordUserIds: ["user-1"], + pollIntervalMs: 10, + rateLimitPerMinute: 10, + maxMessageLength: 500, + includePrefix: true, + }, + } +} + +function createState(): ReplyListenerDaemonState { + return { + isRunning: true, + pid: 1234, + startedAt: "2026-04-07T00:00:00.000Z", + startupToken: "startup-token", + configSignature: null, + lastPollAt: "2026-04-07T00:00:01.000Z", + telegramLastUpdateId: null, + discordLastMessageId: null, + lastDiscordMessageId: null, + messagesSeen: 0, + messagesInjected: 0, + errors: 0, + } +} + +describe("pollDiscordReplies", () => { + beforeEach(() => { + process.env.HOME = tempHome + process.env.USERPROFILE = tempHome + globalThis.fetch = originalFetch + rmSync(stateDir, { recursive: true, force: true }) + mkdirSync(stateDir, { recursive: true }) + }) + + afterEach(() => { + mock.restore() + globalThis.fetch = originalFetch + }) + + test("records HTTP failures in daemon state when Discord returns non-ok", async () => { + const fetchSpy = spyOn(globalThis, "fetch").mockResolvedValue( + new Response("unauthorized", { + status: 401, + }), + ) + + const state = createState() + + await pollDiscordReplies(createConfig(), state, new ReplyListenerRateLimiter(10)) + + expect(fetchSpy).toHaveBeenCalledTimes(1) + expect(state.errors).toBe(1) + expect(state.lastError).toBe("Discord API error: HTTP 401") + expect(existsSync(stateFilePath)).toBe(true) + + const persistedState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as ReplyListenerDaemonState + expect(persistedState.errors).toBe(1) + expect(persistedState.lastError).toBe("Discord API error: HTTP 401") + expect(persistedState.messagesSeen).toBe(0) + }) + + test("increments messagesInjected when a Discord reply matches a registered message", async () => { + const fetchSpy = spyOn(globalThis, "fetch") + .mockResolvedValueOnce( + new Response( + JSON.stringify([ + { + id: "incoming-1", + content: "Ship it", + author: { id: "user-1" }, + message_reference: { message_id: "outbound-1" }, + }, + ]), + { status: 200 }, + ), + ) + .mockResolvedValueOnce(new Response(null, { status: 204 })) + const lookupSpy = spyOn(sessionRegistryModule, "lookupByMessageId").mockReturnValue({ + sessionId: "ses-1", + tmuxSession: "session-1", + tmuxPaneId: "%7", + projectPath: "/tmp/project", + platform: "discord-bot", + messageId: "outbound-1", + createdAt: "2026-04-07T00:00:00.000Z", + }) + const injectSpy = spyOn(injectionModule, "injectReplyIntoPane").mockResolvedValue(true) + + const state = createState() + + await pollDiscordReplies(createConfig(), state, new ReplyListenerRateLimiter(10)) + + expect(lookupSpy).toHaveBeenCalledWith("discord-bot", "outbound-1") + expect(injectSpy).toHaveBeenCalledWith("%7", "Ship it", "discord", createConfig()) + expect(fetchSpy).toHaveBeenCalledTimes(2) + expect(state.messagesSeen).toBe(1) + expect(state.messagesInjected).toBe(1) + expect(state.lastDiscordMessageId).toBe("incoming-1") + }) +}) diff --git a/src/openclaw/__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/__tests__/runtime-dispatch.test.ts b/src/openclaw/__tests__/runtime-dispatch.test.ts new file mode 100644 index 000000000..941253520 --- /dev/null +++ b/src/openclaw/__tests__/runtime-dispatch.test.ts @@ -0,0 +1,91 @@ +import { afterEach, describe, expect, mock, spyOn, test } from "bun:test" +import * as openclawModule from "../index" +import * as sessionRegistryModule from "../session-registry" +import { dispatchOpenClawEvent } from "../runtime-dispatch" +import type { OpenClawConfig } from "../types" + +function createConfig(hooks: OpenClawConfig["hooks"]): OpenClawConfig { + return { + enabled: true, + gateways: { + gateway: { + type: "http", + url: "https://example.com", + method: "POST", + }, + }, + hooks, + } +} + +afterEach(() => { + mock.restore() +}) + +describe("dispatchOpenClawEvent", () => { + test("falls back from raw session.created to canonical session-start", async () => { + const wakeSpy = spyOn(openclawModule, "wakeOpenClaw") + .mockResolvedValueOnce(null) + .mockResolvedValueOnce({ gateway: "gateway", success: true }) + + await dispatchOpenClawEvent({ + config: createConfig({ + "session-start": { enabled: true, gateway: "gateway", instruction: "hi" }, + }), + rawEvent: "session.created", + context: { sessionId: "ses-1", projectPath: "/tmp/project", tmuxPaneId: "%1", tmuxSession: "main" }, + }) + + expect(wakeSpy.mock.calls.map((call) => call[1])).toEqual(["session.created", "session-start"]) + }) + + test("registers reply correlation when wake returns outbound metadata", async () => { + spyOn(openclawModule, "wakeOpenClaw").mockResolvedValue({ + gateway: "gateway", + success: true, + messageId: "msg-1", + platform: "discord", + channelId: "chan-1", + threadId: "thread-1", + }) + const registerSpy = spyOn(sessionRegistryModule, "registerMessage").mockReturnValue(true) + + await dispatchOpenClawEvent({ + config: createConfig({ + "session.created": { enabled: true, gateway: "gateway", instruction: "hi" }, + }), + rawEvent: "session.created", + context: { + sessionId: "ses-1", + projectPath: "/tmp/project", + tmuxPaneId: "%7", + tmuxSession: "session-1", + }, + }) + + const [mapping] = registerSpy.mock.calls[0] ?? [] + expect(mapping).toMatchObject({ + sessionId: "ses-1", + tmuxPaneId: "%7", + tmuxSession: "session-1", + projectPath: "/tmp/project", + platform: "discord-bot", + messageId: "msg-1", + channelId: "chan-1", + threadId: "thread-1", + }) + }) + + test("cleans up session mappings on session.deleted", async () => { + spyOn(openclawModule, "wakeOpenClaw").mockResolvedValue(null) + const removeSpy = spyOn(sessionRegistryModule, "removeSession").mockImplementation(() => {}) + + await dispatchOpenClawEvent({ + config: createConfig({}), + rawEvent: "session.deleted", + context: { sessionId: "ses-2", projectPath: "/tmp/project" }, + }) + + expect(removeSpy).toHaveBeenCalledWith("ses-2") + }) +}) diff --git a/src/openclaw/dispatcher.ts b/src/openclaw/dispatcher.ts index 75cab8c23..173819c93 100644 --- a/src/openclaw/dispatcher.ts +++ b/src/openclaw/dispatcher.ts @@ -1,5 +1,5 @@ import { spawn } from "bun" -import type { OpenClawGateway } from "./types" +import type { OpenClawGateway, WakeResult } from "./types" const DEFAULT_HTTP_TIMEOUT_MS = 10_000 const DEFAULT_COMMAND_TIMEOUT_MS = 5_000 @@ -66,11 +66,70 @@ export function resolveCommandTimeoutMs( ) } +function asRecord(value: unknown): Record | null { + return typeof value === "object" && value !== null ? (value as Record) : null +} + +function firstStringValue(record: Record, keys: string[]): string | undefined { + for (const key of keys) { + const value = record[key] + if (typeof value === "string" && value.trim().length > 0) return value + if (typeof value === "number" && Number.isFinite(value)) return String(value) + } + return undefined +} + +function extractWakeMetadata(payload: unknown): Pick { + const record = asRecord(payload) + if (!record) return {} + + const nestedCandidates = [record, asRecord(record.data), asRecord(record.result), asRecord(record.message)] + .filter((candidate): candidate is Record => candidate !== null) + + let bestMatch: Pick = {} + let bestScore = -1 + + for (const candidate of nestedCandidates) { + const messageId = firstStringValue(candidate, ["messageId", "message_id", "id"]) + const platform = firstStringValue(candidate, ["platform", "source"]) + const channelId = firstStringValue(candidate, ["channelId", "channel_id", "channel"]) + const threadId = firstStringValue(candidate, ["threadId", "thread_id", "thread"]) + + const score = + (messageId ? 4 : 0) + + (platform ? 3 : 0) + + (channelId ? 2 : 0) + + (threadId ? 1 : 0) + + if (score > bestScore) { + bestMatch = { messageId, platform, channelId, threadId } + bestScore = score + } + } + + return bestScore > 0 ? bestMatch : {} +} + +function parseWakeMetadata(raw: string): Pick { + const trimmed = raw.trim() + if (!trimmed) return {} + try { + return extractWakeMetadata(JSON.parse(trimmed)) + } catch { + const messageId = trimmed.match(/message\s+id:\s*([^\s]+)/i)?.[1] + const platform = trimmed.match(/sent\s+via\s+([a-z0-9_-]+)/i)?.[1]?.toLowerCase() + return { + ...(messageId ? { messageId } : {}), + ...(platform ? { platform } : {}), + } + } +} + export async function wakeGateway( gatewayName: string, gatewayConfig: OpenClawGateway, payload: unknown, -): Promise<{ gateway: string; success: boolean; error?: string; statusCode?: number }> { +): Promise { if (!gatewayConfig.url || !validateGatewayUrl(gatewayConfig.url)) { return { gateway: gatewayName, @@ -107,8 +166,10 @@ export async function wakeGateway( statusCode: response.status, } } - - return { gateway: gatewayName, success: true, statusCode: response.status } + + const metadata = parseWakeMetadata(await response.text()) + + return { gateway: gatewayName, success: true, statusCode: response.status, ...metadata } } catch (error) { return { gateway: gatewayName, @@ -122,7 +183,7 @@ export async function wakeCommandGateway( gatewayName: string, gatewayConfig: OpenClawGateway, variables: Record, -): Promise<{ gateway: string; success: boolean; error?: string }> { +): Promise { if (!gatewayConfig.command) { return { gateway: gatewayName, @@ -142,10 +203,11 @@ export async function wakeCommandGateway( const proc = spawn(["sh", "-c", interpolated], { env: { ...process.env }, - stdout: "ignore", + stdout: "pipe", stderr: "ignore", detached: process.platform !== "win32", }) + const stdoutPromise = new Response(proc.stdout).text() let timeoutId: ReturnType | undefined const timeoutPromise = new Promise((_, reject) => { @@ -167,7 +229,9 @@ export async function wakeCommandGateway( throw new Error(`Command exited with code ${proc.exitCode}`) } - return { gateway: gatewayName, success: true } + const metadata = parseWakeMetadata(await stdoutPromise) + + return { gateway: gatewayName, success: true, ...metadata } } catch (error) { return { gateway: gatewayName, diff --git a/src/openclaw/index.ts b/src/openclaw/index.ts index 5cbbe3362..cf352ab21 100644 --- a/src/openclaw/index.ts +++ b/src/openclaw/index.ts @@ -132,10 +132,16 @@ export async function wakeOpenClaw( } export async function initializeOpenClaw(config: OpenClawConfig): Promise { - const replyListener = config.replyListener - if (config.enabled && (replyListener?.discordBotToken || replyListener?.telegramBotToken)) { + const hasReplyListenerCredentials = Boolean( + config.replyListener?.discordBotToken || config.replyListener?.telegramBotToken, + ) + + if (config.enabled && hasReplyListenerCredentials) { await startReplyListener(config) + return } + + await stopReplyListener() } export { startReplyListener, stopReplyListener } diff --git a/src/openclaw/reply-listener-discord.ts b/src/openclaw/reply-listener-discord.ts new file mode 100644 index 000000000..fbc504e46 --- /dev/null +++ b/src/openclaw/reply-listener-discord.ts @@ -0,0 +1,110 @@ +import { lookupByMessageId } from "./session-registry" +import { injectReplyIntoPane, ReplyListenerRateLimiter } from "./reply-listener-injection" +import { logReplyListenerMessage } from "./reply-listener-log" +import { + recordSeenDiscordMessage, + writeReplyListenerDaemonState, + type ReplyListenerDaemonState, +} from "./reply-listener-state" +import type { OpenClawConfig } from "./types" + +interface DiscordMessage { + id: string + content: string + author: { id: string } + message_reference?: { message_id?: string } +} + +let discordBackoffUntil = 0 + +export async function pollDiscordReplies( + config: OpenClawConfig, + state: ReplyListenerDaemonState, + rateLimiter: ReplyListenerRateLimiter, +): Promise { + const replyListener = config.replyListener + if (!replyListener?.discordBotToken || !replyListener.discordChannelId) return + if (!replyListener.authorizedDiscordUserIds || replyListener.authorizedDiscordUserIds.length === 0) { + return + } + if (Date.now() < discordBackoffUntil) return + + try { + const after = state.discordLastMessageId + ? `?after=${state.discordLastMessageId}&limit=10` + : "?limit=10" + const url = `https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages${after}` + + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 10000) + const response = await fetch(url, { + method: "GET", + headers: { Authorization: `Bot ${replyListener.discordBotToken}` }, + signal: controller.signal, + }) + clearTimeout(timeout) + + const remaining = response.headers.get("x-ratelimit-remaining") + const reset = response.headers.get("x-ratelimit-reset") + if (remaining !== null && Number.parseInt(remaining, 10) < 2) { + const parsedReset = reset ? Number.parseFloat(reset) : Number.NaN + const resetTime = Number.isFinite(parsedReset) ? parsedReset * 1000 : Date.now() + 10000 + discordBackoffUntil = resetTime + logReplyListenerMessage( + `WARN: Discord rate limit low (remaining: ${remaining}), backing off until ${new Date(resetTime).toISOString()}`, + ) + } + + if (!response.ok) { + state.errors += 1 + state.lastError = `Discord API error: HTTP ${response.status}` + logReplyListenerMessage(state.lastError) + writeReplyListenerDaemonState(state) + return + } + + const messages = await response.json() + if (!Array.isArray(messages) || messages.length === 0) return + + for (const message of [...messages as DiscordMessage[]].reverse()) { + recordSeenDiscordMessage(state, message.id) + writeReplyListenerDaemonState(state) + + const replyToMessageId = message.message_reference?.message_id + if (!replyToMessageId) continue + if (!replyListener.authorizedDiscordUserIds.includes(message.author.id)) continue + + const mapping = lookupByMessageId("discord-bot", replyToMessageId) + if (!mapping) continue + + if (!rateLimiter.canProceed()) { + logReplyListenerMessage(`WARN: Rate limit exceeded, dropping Discord message ${message.id}`) + state.errors += 1 + continue + } + + const success = await injectReplyIntoPane(mapping.tmuxPaneId, message.content, "discord", config) + if (success) { + state.messagesInjected += 1 + try { + await fetch( + `https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages/${message.id}/reactions/%E2%9C%85/@me`, + { + method: "PUT", + headers: { Authorization: `Bot ${replyListener.discordBotToken}` }, + }, + ) + } catch { + } + } else { + state.errors += 1 + } + + writeReplyListenerDaemonState(state) + } + } catch (error) { + state.errors += 1 + state.lastError = error instanceof Error ? error.message : String(error) + logReplyListenerMessage(`Discord polling error: ${state.lastError}`) + } +} diff --git a/src/openclaw/reply-listener-injection.ts b/src/openclaw/reply-listener-injection.ts new file mode 100644 index 000000000..97669e9ba --- /dev/null +++ b/src/openclaw/reply-listener-injection.ts @@ -0,0 +1,74 @@ +import { removeMessagesByPane } from "./session-registry" +import { analyzePaneContent, captureTmuxPane, sendToPane } from "./tmux" +import { logReplyListenerMessage } from "./reply-listener-log" +import type { OpenClawConfig } from "./types" + +export function sanitizeReplyInput(text: string): string { + return text + .replace(/[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]/g, "") + .replace(/[\u200e\u200f\u202a-\u202e\u2066-\u2069]/g, "") + .replace(/\r?\n/g, " ") + .replace(/\\/g, "\\\\") + .replace(/`/g, "\\`") + .replace(/\$\(/g, "\\$(") + .replace(/\$\{/g, "\\${") + .trim() +} + +export class ReplyListenerRateLimiter { + private readonly maxPerMinute: number + private readonly timestamps: number[] = [] + private readonly windowMs = 60 * 1000 + + constructor(maxPerMinute: number) { + this.maxPerMinute = maxPerMinute + } + + canProceed(): boolean { + const now = Date.now() + const recent = this.timestamps.filter((timestamp) => now - timestamp < this.windowMs) + this.timestamps.length = 0 + this.timestamps.push(...recent) + + if (this.timestamps.length >= this.maxPerMinute) { + return false + } + + this.timestamps.push(now) + return true + } +} + +export async function injectReplyIntoPane( + paneId: string, + text: string, + platform: string, + config: OpenClawConfig, +): Promise { + const replyListener = config.replyListener + const content = await captureTmuxPane(paneId, 15) + const analysis = analyzePaneContent(content) + + if (analysis.confidence < 0.3) { + logReplyListenerMessage( + `WARN: Pane ${paneId} does not appear to be running OpenCode CLI (confidence: ${analysis.confidence}). Skipping injection, removing stale mapping.`, + ) + removeMessagesByPane(paneId) + return false + } + + const prefix = replyListener?.includePrefix === false ? "" : `[reply:${platform}] ` + const sanitized = sanitizeReplyInput(prefix + text) + const truncated = sanitized.slice(0, replyListener?.maxMessageLength ?? 500) + const success = await sendToPane(paneId, truncated, true) + + if (success) { + logReplyListenerMessage( + `Injected reply from ${platform} into pane ${paneId}: "${truncated.slice(0, 50)}${truncated.length > 50 ? "..." : ""}"`, + ) + } else { + logReplyListenerMessage(`ERROR: Failed to inject reply into pane ${paneId}`) + } + + return success +} diff --git a/src/openclaw/reply-listener-log.ts b/src/openclaw/reply-listener-log.ts new file mode 100644 index 000000000..58a536d6c --- /dev/null +++ b/src/openclaw/reply-listener-log.ts @@ -0,0 +1,55 @@ +import { + appendFileSync, + chmodSync, + existsSync, + renameSync, + statSync, + unlinkSync, + writeFileSync, +} from "fs" +import { + ensureReplyListenerStateDir, + REPLY_LISTENER_SECURE_FILE_MODE, + getReplyListenerLogFilePath, +} from "./reply-listener-paths" + +const MAX_REPLY_LISTENER_LOG_SIZE_BYTES = 1024 * 1024 + +export function writeSecureReplyListenerFile(filePath: string, content: string): void { + ensureReplyListenerStateDir() + writeFileSync(filePath, content, { mode: REPLY_LISTENER_SECURE_FILE_MODE }) + + try { + chmodSync(filePath, REPLY_LISTENER_SECURE_FILE_MODE) + } catch { + } +} + +function rotateReplyListenerLogIfNeeded(logPath: string): void { + try { + if (!existsSync(logPath)) return + + const stats = statSync(logPath) + if (stats.size <= MAX_REPLY_LISTENER_LOG_SIZE_BYTES) return + + const backupPath = `${logPath}.old` + if (existsSync(backupPath)) { + unlinkSync(backupPath) + } + renameSync(logPath, backupPath) + } catch { + } +} + +export function logReplyListenerMessage(message: string): void { + try { + ensureReplyListenerStateDir() + const logFilePath = getReplyListenerLogFilePath() + rotateReplyListenerLogIfNeeded(logFilePath) + const timestamp = new Date().toISOString() + appendFileSync(logFilePath, `[${timestamp}] ${message}\n`, { + mode: REPLY_LISTENER_SECURE_FILE_MODE, + }) + } catch { + } +} 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-process.ts b/src/openclaw/reply-listener-process.ts new file mode 100644 index 000000000..f6309f168 --- /dev/null +++ b/src/openclaw/reply-listener-process.ts @@ -0,0 +1,78 @@ +import { readFileSync } from "fs" +import { spawn } from "bun" + +export const REPLY_LISTENER_DAEMON_IDENTITY_MARKER = "--openclaw-reply-listener-daemon" + +const REPLY_LISTENER_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", +] as const + +export function createReplyListenerDaemonEnv(extraEnv: Record): Record { + const env: Record = {} + + for (const key of REPLY_LISTENER_DAEMON_ENV_ALLOWLIST) { + const value = process.env[key] + if (value !== undefined) { + env[key] = value + } + } + + return { ...env, ...extraEnv } +} + +export function isReplyListenerProcessRunning(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch { + return false + } +} + +export async function isReplyListenerDaemonProcess(pid: number): Promise { + try { + if (process.platform === "linux") { + const cmdline = readFileSync(`/proc/${pid}/cmdline`, "utf-8") + return cmdline.includes(REPLY_LISTENER_DAEMON_IDENTITY_MARKER) + } + + const processInfo = spawn(["ps", "-p", String(pid), "-o", "args="], { + stdout: "pipe", + stderr: "ignore", + }) + const stdout = await new Response(processInfo.stdout).text() + if (processInfo.exitCode !== 0) return false + return stdout.includes(REPLY_LISTENER_DAEMON_IDENTITY_MARKER) + } catch { + return false + } +} diff --git a/src/openclaw/reply-listener-spawn.ts b/src/openclaw/reply-listener-spawn.ts new file mode 100644 index 000000000..1cd0a1818 --- /dev/null +++ b/src/openclaw/reply-listener-spawn.ts @@ -0,0 +1,25 @@ +import { spawn } from "bun" +import { + createReplyListenerDaemonEnv, + REPLY_LISTENER_DAEMON_IDENTITY_MARKER, +} from "./reply-listener-process" +import { REPLY_LISTENER_STARTUP_TOKEN_ENV } from "./reply-listener-state" + +export interface ReplyListenerSpawnProcess { + pid: number | undefined + unref(): void +} + +export function spawnReplyListenerDaemon( + daemonScript: string, + startupToken: string, +): ReplyListenerSpawnProcess { + return spawn(["bun", "run", daemonScript, REPLY_LISTENER_DAEMON_IDENTITY_MARKER], { + detached: true, + stdio: ["ignore", "ignore", "ignore"], + cwd: process.cwd(), + env: createReplyListenerDaemonEnv({ + [REPLY_LISTENER_STARTUP_TOKEN_ENV]: startupToken, + }), + }) +} 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-telegram.ts b/src/openclaw/reply-listener-telegram.ts new file mode 100644 index 000000000..e64e563a0 --- /dev/null +++ b/src/openclaw/reply-listener-telegram.ts @@ -0,0 +1,92 @@ +import { lookupByMessageId } from "./session-registry" +import { injectReplyIntoPane, ReplyListenerRateLimiter } from "./reply-listener-injection" +import { logReplyListenerMessage } from "./reply-listener-log" +import { writeReplyListenerDaemonState, type ReplyListenerDaemonState } from "./reply-listener-state" +import type { OpenClawConfig } from "./types" + +interface TelegramMessage { + message_id?: number + chat?: { id?: number | string } + text?: string + reply_to_message?: { message_id?: number } +} + +interface TelegramUpdate { + update_id?: number + message?: TelegramMessage +} + +function parseTelegramUpdatesResponse(body: unknown): TelegramUpdate[] { + if (typeof body !== "object" || body === null) return [] + const result = (body as { result?: TelegramUpdate[] }).result + return Array.isArray(result) ? result : [] +} + +export async function pollTelegramReplies( + config: OpenClawConfig, + state: ReplyListenerDaemonState, + rateLimiter: ReplyListenerRateLimiter, +): Promise { + const replyListener = config.replyListener + if (!replyListener?.telegramBotToken || !replyListener.telegramChatId) return + + try { + const offset = state.telegramLastUpdateId ? state.telegramLastUpdateId + 1 : 0 + const url = `https://api.telegram.org/bot${replyListener.telegramBotToken}/getUpdates?offset=${offset}&timeout=0` + + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 10000) + const response = await fetch(url, { method: "GET", signal: controller.signal }) + clearTimeout(timeout) + + if (!response.ok) { + logReplyListenerMessage(`Telegram API error: HTTP ${response.status}`) + return + } + + const updates = parseTelegramUpdatesResponse(await response.json()) + for (const update of updates) { + const message = update.message + state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId + writeReplyListenerDaemonState(state) + + if (!message?.reply_to_message?.message_id) continue + if (String(message.chat?.id) !== replyListener.telegramChatId) continue + if (!message.text) continue + + const mapping = lookupByMessageId("telegram", String(message.reply_to_message.message_id)) + if (!mapping) continue + + if (!rateLimiter.canProceed()) { + logReplyListenerMessage(`WARN: Rate limit exceeded, dropping Telegram message ${message.message_id}`) + state.errors += 1 + continue + } + + const success = await injectReplyIntoPane(mapping.tmuxPaneId, message.text, "telegram", config) + if (success) { + state.messagesInjected += 1 + try { + await fetch(`https://api.telegram.org/bot${replyListener.telegramBotToken}/sendMessage`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + chat_id: replyListener.telegramChatId, + text: "Injected into Codex CLI session.", + reply_to_message_id: message.message_id, + }), + }) + } catch { + } + } else { + state.errors += 1 + } + + writeReplyListenerDaemonState(state) + } + } catch (error) { + state.errors += 1 + state.lastError = error instanceof Error ? error.message : String(error) + logReplyListenerMessage(`Telegram polling error: ${state.lastError}`) + } +} 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 } diff --git a/src/openclaw/runtime-dispatch.ts b/src/openclaw/runtime-dispatch.ts new file mode 100644 index 000000000..79500d343 --- /dev/null +++ b/src/openclaw/runtime-dispatch.ts @@ -0,0 +1,89 @@ +import * as openclaw from "./index" +import { registerMessage, removeSession } from "./session-registry" +import { getCurrentTmuxSession } from "./tmux" +import type { OpenClawConfig, WakeResult } from "./types" + +interface DispatchOpenClawContext { + sessionId?: string + projectPath?: string + tmuxPaneId?: string + tmuxSession?: string + replyChannel?: string + replyTarget?: string + replyThread?: string +} + +interface DispatchOpenClawEventParams { + config: OpenClawConfig + rawEvent: string + context: DispatchOpenClawContext +} + +function mapRawEventToOpenClawEvents(rawEvent: string): string[] { + const aliases: Record = { + "session.created": "session-start", + "session.deleted": "session-end", + "session.idle": "stop", + } + + const mapped = aliases[rawEvent] + return Array.from(new Set([rawEvent, mapped].filter((value): value is string => Boolean(value)))) +} + +function normalizePlatform(platform?: string): string | undefined { + if (!platform) return undefined + if (platform === "discord") return "discord-bot" + return platform +} + +function shouldRegisterReplyCorrelation(result: WakeResult, params: DispatchOpenClawEventParams): boolean { + if (params.rawEvent === "session.deleted") return false + if (!result.success) return false + if (!result.messageId || !result.platform) return false + if (!params.context.sessionId || !params.context.projectPath || !params.context.tmuxPaneId) return false + return true +} + +export async function dispatchOpenClawEvent( + params: DispatchOpenClawEventParams, +): Promise { + let result: WakeResult | null = null + + if (params.config.enabled) { + for (const event of mapRawEventToOpenClawEvents(params.rawEvent)) { + result = await openclaw.wakeOpenClaw(params.config, event, { + sessionId: params.context.sessionId, + projectPath: params.context.projectPath, + tmuxSession: params.context.tmuxSession, + replyChannel: params.context.replyChannel, + replyTarget: params.context.replyTarget, + replyThread: params.context.replyThread, + }) + if (result !== null) break + } + } + + if (shouldRegisterReplyCorrelation(result ?? { gateway: "", success: false }, params)) { + const tmuxSession = params.context.tmuxSession ?? getCurrentTmuxSession() + const platform = normalizePlatform(result?.platform) + if (tmuxSession && platform && params.context.sessionId && params.context.projectPath && params.context.tmuxPaneId) { + registerMessage({ + sessionId: params.context.sessionId, + tmuxSession, + tmuxPaneId: params.context.tmuxPaneId, + projectPath: params.context.projectPath, + platform, + messageId: result!.messageId!, + channelId: result?.channelId, + threadId: result?.threadId, + createdAt: new Date().toISOString(), + }) + } + } + + if (params.rawEvent === "session.deleted" && params.context.sessionId) { + removeSession(params.context.sessionId) + } + + return result +} diff --git a/src/openclaw/types.ts b/src/openclaw/types.ts index b05325da2..e29a5f201 100644 --- a/src/openclaw/types.ts +++ b/src/openclaw/types.ts @@ -49,4 +49,8 @@ export interface WakeResult { success: boolean error?: string statusCode?: number + messageId?: string + platform?: string + channelId?: string + threadId?: string } diff --git a/src/plugin/event.test.ts b/src/plugin/event.test.ts index 2fc453dc2..1d39b1ed5 100644 --- a/src/plugin/event.test.ts +++ b/src/plugin/event.test.ts @@ -1,7 +1,8 @@ -import { describe, it, expect, afterEach } from "bun:test" +import { describe, it, expect, afterEach, mock, spyOn } from "bun:test" import { createEventHandler } from "./event" import { createChatMessageHandler } from "./chat-message" +import * as openclawRuntimeDispatch from "../openclaw/runtime-dispatch" import { _resetForTesting, setMainSession } from "../features/claude-code-session-state" import { clearPendingModelFallback, createModelFallbackHook } from "../hooks/model-fallback/hook" import { getSessionPromptParams, setSessionPromptParams } from "../shared/session-prompt-params-state" @@ -63,6 +64,7 @@ function createChatMessageHandlerHooks( } afterEach(() => { + mock.restore() _resetForTesting() }) @@ -588,6 +590,54 @@ describe("createEventHandler - event forwarding", () => { expect(createdSessions).toHaveLength(0) }) + it("dispatches OpenClaw after session.created using tracked pane metadata", async () => { + const openClawSpy = spyOn(openclawRuntimeDispatch, "dispatchOpenClawEvent").mockResolvedValue(null) + const eventHandler = createEventHandler({ + ctx: asEventHandlerContext({ directory: "/tmp/project-created" }), + pluginConfig: asPluginConfig({ + openclaw: { enabled: true, gateways: {}, hooks: {} }, + tmux: { + enabled: true, + layout: "main-vertical", + main_pane_size: 60, + main_pane_min_width: 120, + agent_pane_min_width: 40, + isolation: "inline", + }, + }), + firstMessageVariantGate: { + markSessionCreated: () => {}, + clear: () => {}, + }, + managers: createEventHandlerManagers({ + skillMcpManager: { disconnectSession: async () => {} }, + tmuxSessionManager: { + onSessionCreated: async () => {}, + onSessionDeleted: async () => {}, + getTrackedPaneId: (sessionID: string) => (sessionID === "ses_openclaw_created" ? "%9" : undefined), + }, + }), + hooks: createEventHandlerHooks({}), + }) + + await eventHandler(asEventHandlerInput({ + event: { + type: "session.created", + properties: { info: { id: "ses_openclaw_created", parentID: "ses_parent" } }, + }, + })) + + const [call] = openClawSpy.mock.calls[0] ?? [] + expect(call).toMatchObject({ + rawEvent: "session.created", + context: { + sessionId: "ses_openclaw_created", + projectPath: "/tmp/project-created", + tmuxPaneId: "%9", + }, + }) + }) + it("forwards session.deleted to write-existing-file-guard hook", async () => { //#given const forwardedEvents: EventInput[] = [] @@ -647,6 +697,44 @@ describe("createEventHandler - event forwarding", () => { expect(deletedSessions).toEqual([sessionID]) }) + it("dispatches OpenClaw for synthetic session.idle events", async () => { + const openClawSpy = spyOn(openclawRuntimeDispatch, "dispatchOpenClawEvent").mockResolvedValue(null) + const eventHandler = createEventHandler({ + ctx: asEventHandlerContext({ directory: "/tmp/project-idle" }), + pluginConfig: asPluginConfig({ openclaw: { enabled: true, gateways: {}, hooks: {} } }), + firstMessageVariantGate: { + markSessionCreated: () => {}, + clear: () => {}, + }, + managers: createEventHandlerManagers({ + skillMcpManager: { disconnectSession: async () => {} }, + tmuxSessionManager: { + onSessionCreated: async () => {}, + onSessionDeleted: async () => {}, + getTrackedPaneId: (sessionID: string) => (sessionID === "ses_openclaw_idle" ? "%3" : undefined), + }, + }), + hooks: createEventHandlerHooks({}), + }) + + await eventHandler(asEventHandlerInput({ + event: { + type: "session.status", + properties: { sessionID: "ses_openclaw_idle", status: { type: "idle" } }, + }, + })) + + const [call] = openClawSpy.mock.calls[0] ?? [] + expect(call).toMatchObject({ + rawEvent: "session.idle", + context: { + sessionId: "ses_openclaw_idle", + projectPath: "/tmp/project-idle", + tmuxPaneId: "%3", + }, + }) + }) + it("clears stored prompt params on session.deleted", async () => { //#given const eventHandler = createEventHandler({ diff --git a/src/plugin/event.ts b/src/plugin/event.ts index 0124fb8ad..10c1da6df 100644 --- a/src/plugin/event.ts +++ b/src/plugin/event.ts @@ -33,6 +33,7 @@ import { clearSessionModel, getSessionModel, setSessionModel } from "../shared/s import { clearSessionPromptParams } from "../shared/session-prompt-params-state"; import { deleteSessionTools } from "../shared/session-tools-store"; import { lspManager } from "../tools"; +import { dispatchOpenClawEvent } from "../openclaw/runtime-dispatch"; import type { CreatedHooks } from "../create-hooks"; import type { Managers } from "../create-managers"; @@ -341,6 +342,17 @@ export function createEventHandler(args: { } recentSyntheticIdles.set(sessionID, Date.now()); await dispatchToHooks(syntheticIdle as EventInput); + if (pluginConfig.openclaw) { + await dispatchOpenClawEvent({ + config: pluginConfig.openclaw, + rawEvent: "session.idle", + context: { + sessionId: sessionID, + projectPath: pluginContext.directory, + tmuxPaneId: managers.tmuxSessionManager.getTrackedPaneId?.(sessionID) ?? process.env.TMUX_PANE, + }, + }); + } } const { event } = input; @@ -369,6 +381,18 @@ export function createEventHandler(args: { }, ); } + + if (pluginConfig.openclaw && sessionInfo?.id) { + await dispatchOpenClawEvent({ + config: pluginConfig.openclaw, + rawEvent: event.type, + context: { + sessionId: sessionInfo.id, + projectPath: pluginContext.directory, + tmuxPaneId: managers.tmuxSessionManager.getTrackedPaneId?.(sessionInfo.id) ?? process.env.TMUX_PANE, + }, + }); + } } if (event.type === "session.deleted") { @@ -392,6 +416,17 @@ export function createEventHandler(args: { clearSessionModel(sessionInfo.id); clearSessionPromptParams(sessionInfo.id); syncSubagentSessions.delete(sessionInfo.id); + if (pluginConfig.openclaw) { + await dispatchOpenClawEvent({ + config: pluginConfig.openclaw, + rawEvent: event.type, + context: { + sessionId: sessionInfo.id, + projectPath: pluginContext.directory, + tmuxPaneId: managers.tmuxSessionManager.getTrackedPaneId?.(sessionInfo.id) ?? process.env.TMUX_PANE, + }, + }); + } if (wasSyncSubagentSession) { subagentSessions.delete(sessionInfo.id); } @@ -412,6 +447,21 @@ export function createEventHandler(args: { restoreBackgroundOutputConsumption(sessionID, messageID); } + if (event.type === "session.idle" && pluginConfig.openclaw) { + const sessionID = props?.sessionID as string | undefined; + if (sessionID) { + await dispatchOpenClawEvent({ + config: pluginConfig.openclaw, + rawEvent: event.type, + context: { + sessionId: sessionID, + projectPath: pluginContext.directory, + tmuxPaneId: managers.tmuxSessionManager.getTrackedPaneId?.(sessionID) ?? process.env.TMUX_PANE, + }, + }); + } + } + if (event.type === "message.updated") { const info = props?.info as Record | undefined; const sessionID = info?.sessionID as string | undefined; diff --git a/src/plugin/tool-execute-before.test.ts b/src/plugin/tool-execute-before.test.ts index 06303504e..80daa79a8 100644 --- a/src/plugin/tool-execute-before.test.ts +++ b/src/plugin/tool-execute-before.test.ts @@ -1,7 +1,8 @@ -const { describe, expect, test } = require("bun:test") +const { afterEach, describe, expect, test } = require("bun:test") const { createToolExecuteBeforeHandler } = require("./tool-execute-before") const { createToolRegistry } = require("./tool-registry") const { builtinTools } = require("../tools") +const { resetStorageClient } = require("../tools/session-manager/storage") describe("createToolExecuteBeforeHandler", () => { test("does not execute subagent question blocker hook for question tool", async () => { @@ -222,11 +223,19 @@ describe("createToolExecuteBeforeHandler", () => { }) describe("createToolRegistry", () => { + afterEach(() => { + resetStorageClient() + }) + function createRegistryInput(overrides = {}) { return { ctx: { directory: process.cwd(), - client: {}, + client: { + session: { + messages: async () => ({ data: [] }), + }, + }, }, pluginConfig: { ...overrides, diff --git a/src/plugin/tool-registry.test.ts b/src/plugin/tool-registry.test.ts index 07588b0ca..4c52b5b79 100644 --- a/src/plugin/tool-registry.test.ts +++ b/src/plugin/tool-registry.test.ts @@ -1,8 +1,9 @@ -import { describe, expect, test } from "bun:test" +import { describe, expect, mock, spyOn, test } from "bun:test" import { tool } from "@opencode-ai/plugin" +import type { OhMyOpenCodeConfig } from "../config" +import * as openclawRuntimeDispatch from "../openclaw/runtime-dispatch" import type { ToolsRecord } from "./types" -import { createToolRegistry, trimToolsToCap } from "./tool-registry" const fakeTool = tool({ description: "test tool", @@ -12,6 +13,58 @@ const fakeTool = tool({ }, }) +const delegateTaskTool = tool({ + description: "task tool", + args: {}, + async execute(): Promise { + return "ok" + }, +}) + +const syncSessionCreatedCallbacks: Array< + ((event: { sessionID: string; parentID: string; title: string }) => Promise) | undefined +> = [] + +mock.module("../tools", () => ({ + builtinTools: { bash: fakeTool, read: fakeTool }, + createBackgroundTools: mock(() => ({})), + createCallOmoAgent: mock(() => fakeTool), + createLookAt: mock(() => fakeTool), + createSkillMcpTool: mock(() => fakeTool), + createSkillTool: mock(() => fakeTool), + createGrepTools: mock(() => ({})), + createGlobTools: mock(() => ({})), + createAstGrepTools: mock(() => ({})), + createSessionManagerTools: mock(() => ({})), + createDelegateTask: mock((options: { onSyncSessionCreated?: typeof syncSessionCreatedCallbacks[number] }) => { + syncSessionCreatedCallbacks.push(options.onSyncSessionCreated) + return delegateTaskTool + }), + discoverCommandsSync: mock(() => []), + interactive_bash: fakeTool, + createTaskCreateTool: mock(() => fakeTool), + createTaskGetTool: mock(() => fakeTool), + createTaskList: mock(() => fakeTool), + createTaskUpdateTool: mock(() => fakeTool), + createHashlineEditTool: mock(() => fakeTool), +})) + +const trackedPaneBySession = new Map() + +const { createToolRegistry, trimToolsToCap } = await import("./tool-registry") +const dispatchOpenClawEvent = spyOn(openclawRuntimeDispatch, "dispatchOpenClawEvent") + +function createPluginConfig(overrides: Partial = {}): OhMyOpenCodeConfig { + return { + git_master: { + commit_footer: false, + include_co_authored_by: false, + git_env_prefix: "", + }, + ...overrides, + } +} + describe("#given tool trimming prioritization", () => { test("#when max_tools trims a hashline edit registration named edit #then edit is removed before higher-priority tools", () => { const filteredTools = { @@ -30,9 +83,11 @@ describe("#given tool trimming prioritization", () => { describe("#given task_system configuration", () => { test("#when task_system is omitted #then task tools are not registered by default", () => { + syncSessionCreatedCallbacks.length = 0 + const result = createToolRegistry({ ctx: { directory: "/tmp" } as Parameters[0]["ctx"], - pluginConfig: {}, + pluginConfig: createPluginConfig(), managers: { backgroundManager: {}, tmuxSessionManager: {}, @@ -55,11 +110,13 @@ describe("#given task_system configuration", () => { }) test("#when task_system is enabled #then task tools are registered", () => { + syncSessionCreatedCallbacks.length = 0 + const result = createToolRegistry({ ctx: { directory: "/tmp" } as Parameters[0]["ctx"], - pluginConfig: { + pluginConfig: createPluginConfig({ experimental: { task_system: true }, - }, + }), managers: { backgroundManager: {}, tmuxSessionManager: {}, @@ -84,9 +141,11 @@ describe("#given task_system configuration", () => { describe("#given tmux integration is disabled", () => { test("#when system tmux is available #then interactive_bash remains registered", () => { + syncSessionCreatedCallbacks.length = 0 + const result = createToolRegistry({ ctx: { directory: "/tmp" } as Parameters[0]["ctx"], - pluginConfig: { + pluginConfig: createPluginConfig({ tmux: { enabled: false, layout: "main-vertical", @@ -95,7 +154,7 @@ describe("#given tmux integration is disabled", () => { agent_pane_min_width: 40, isolation: "inline", }, - }, + }), managers: { backgroundManager: {}, tmuxSessionManager: {}, @@ -115,9 +174,11 @@ describe("#given tmux integration is disabled", () => { }) test("#when system tmux is unavailable #then interactive_bash is not registered", () => { + syncSessionCreatedCallbacks.length = 0 + const result = createToolRegistry({ ctx: { directory: "/tmp" } as Parameters[0]["ctx"], - pluginConfig: { + pluginConfig: createPluginConfig({ tmux: { enabled: false, layout: "main-vertical", @@ -126,7 +187,7 @@ describe("#given tmux integration is disabled", () => { agent_pane_min_width: 40, isolation: "inline", }, - }, + }), managers: { backgroundManager: {}, tmuxSessionManager: {}, @@ -145,3 +206,64 @@ describe("#given tmux integration is disabled", () => { expect(result.filteredTools).not.toHaveProperty("interactive_bash") }) }) + +describe("#given openclaw is enabled for sync task sessions", () => { + test("#when the sync session-created callback runs #then it dispatches openclaw with the tracked pane id", async () => { + syncSessionCreatedCallbacks.length = 0 + dispatchOpenClawEvent.mockReset() + trackedPaneBySession.clear() + + const tmuxSessionManager = { + async onSessionCreated(event: { properties?: { info?: { id?: string } } }): Promise { + const sessionID = event.properties?.info?.id + if (sessionID) { + trackedPaneBySession.set(sessionID, `%pane-${sessionID}`) + } + }, + getTrackedPaneId(sessionID: string): string | undefined { + return trackedPaneBySession.get(sessionID) + }, + } + + const openclawConfig = { + enabled: true, + gateways: {}, + hooks: {}, + } + + createToolRegistry({ + ctx: { directory: "/tmp/project" } as Parameters[0]["ctx"], + pluginConfig: createPluginConfig({ openclaw: openclawConfig }), + managers: { + backgroundManager: {}, + tmuxSessionManager, + skillMcpManager: {}, + } as Parameters[0]["managers"], + skillContext: { + mergedSkills: [], + availableSkills: [], + browserProvider: "playwright", + disabledSkills: new Set(), + }, + availableCategories: [], + }) + + const onSyncSessionCreated = syncSessionCreatedCallbacks[syncSessionCreatedCallbacks.length - 1] + await onSyncSessionCreated?.({ + sessionID: "ses-sync-1", + parentID: "ses-parent", + title: "sync task", + }) + + expect(dispatchOpenClawEvent).toHaveBeenCalledTimes(1) + expect(dispatchOpenClawEvent).toHaveBeenCalledWith({ + config: openclawConfig, + rawEvent: "session.created", + context: { + sessionId: "ses-sync-1", + projectPath: "/tmp/project", + tmuxPaneId: "%pane-ses-sync-1", + }, + }) + }) +}) diff --git a/src/plugin/tool-registry.ts b/src/plugin/tool-registry.ts index ed99f7c22..98bd4c9a9 100644 --- a/src/plugin/tool-registry.ts +++ b/src/plugin/tool-registry.ts @@ -6,6 +6,7 @@ import type { } from "../agents/dynamic-agent-prompt-builder" import type { OhMyOpenCodeConfig } from "../config" import { isInteractiveBashEnabled } from "../create-runtime-tmux-config" +import * as openclawRuntimeDispatch from "../openclaw/runtime-dispatch" import type { PluginContext, ToolsRecord } from "./types" import { @@ -158,6 +159,18 @@ export function createToolRegistry(args: { }, }, }) + + if (pluginConfig.openclaw) { + await openclawRuntimeDispatch.dispatchOpenClawEvent({ + config: pluginConfig.openclaw, + rawEvent: "session.created", + context: { + sessionId: event.sessionID, + projectPath: ctx.directory, + tmuxPaneId: managers.tmuxSessionManager.getTrackedPaneId?.(event.sessionID) ?? process.env.TMUX_PANE, + }, + }) + } }, }) diff --git a/src/tools/skill/async-description-refresh.test.ts b/src/tools/skill/async-description-refresh.test.ts index 190f58a06..9712c2678 100644 --- a/src/tools/skill/async-description-refresh.test.ts +++ b/src/tools/skill/async-description-refresh.test.ts @@ -19,13 +19,15 @@ function createMockSkill(name: string): LoadedSkill { } async function waitForRefresh(predicate: () => boolean): Promise { - for (let attempt = 0; attempt < 20; attempt += 1) { + for (let attempt = 0; attempt < 200; attempt += 1) { if (predicate()) { return } - await new Promise((resolve) => setTimeout(resolve, 0)) + await new Promise((resolve) => setTimeout(resolve, 10)) } + + throw new Error("Timed out waiting for async skill description refresh") } describe("skill tool - async native skill description refresh", () => {