Merge pull request #3206 from Momentum96/fix/openclaw-reply-listener
fix(openclaw): stabilize reply listener wiring and runtime dispatch
This commit is contained in:
YeonGyu-Kim
@@ -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<string[]> {
|
||||
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),
|
||||
),
|
||||
)
|
||||
|
||||
+113
-54
@@ -1,23 +1,66 @@
|
||||
/// <reference types="bun-types" />
|
||||
|
||||
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<void>
|
||||
} | null = null
|
||||
const trackedPaneBySession = new Map<string, string>()
|
||||
|
||||
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<void> {}
|
||||
async onSessionCreated(..._args: unknown[]): Promise<void> {}
|
||||
}
|
||||
mock.module("./features/task-toast-manager", () => ({
|
||||
initTaskToastManager: mock(() => {}),
|
||||
}))
|
||||
|
||||
mock.module("./features/tmux-subagent", () => ({
|
||||
TmuxSessionManager: class TmuxSessionManager {
|
||||
constructor(..._args: unknown[]) {}
|
||||
|
||||
async cleanup(): Promise<void> {}
|
||||
async onSessionCreated(event: { properties?: { info?: { id?: string } } }): Promise<void> {
|
||||
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<typeof createManagers>[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<typeof createManagers>[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<typeof createManagers>[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<typeof createManagers>[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",
|
||||
},
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
/// <reference path="../../../bun-test.d.ts" />
|
||||
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
|
||||
|
||||
+179
-1
@@ -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<typeof OhMyOpenCodePlugin>[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<typeof OhMyOpenCodePlugin>[0])
|
||||
|
||||
// then
|
||||
expect(mockInitializeOpenClaw).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
+5
-1
@@ -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()
|
||||
|
||||
@@ -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",
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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")
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,413 @@
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, mock, spyOn, test } from "bun:test"
|
||||
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "fs"
|
||||
import { tmpdir } from "os"
|
||||
import { join } from "path"
|
||||
import type { OpenClawConfig } from "../types"
|
||||
|
||||
interface MockSpawnProcess {
|
||||
pid: number
|
||||
unref(): void
|
||||
}
|
||||
|
||||
type SpawnImplementation = (...args: unknown[]) => MockSpawnProcess
|
||||
|
||||
const originalHome = process.env.HOME
|
||||
const originalUserProfile = process.env.USERPROFILE
|
||||
const originalStartupTimeout = process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS
|
||||
|
||||
const tempHome = mkdtempSync(join(tmpdir(), "openclaw-reply-listener-"))
|
||||
const stateDir = join(tempHome, ".omx", "state")
|
||||
const configFilePath = join(stateDir, "reply-listener-config.json")
|
||||
const stateFilePath = join(stateDir, "reply-listener-state.json")
|
||||
const pidFilePath = join(stateDir, "reply-listener.pid")
|
||||
|
||||
const livePids = new Set<number>()
|
||||
const daemonPids = new Set<number>()
|
||||
|
||||
let spawnImplementation: SpawnImplementation = () => ({
|
||||
pid: 0,
|
||||
unref() {
|
||||
},
|
||||
})
|
||||
|
||||
let replyListenerModule: typeof import("../reply-listener")
|
||||
|
||||
function createConfig(): OpenClawConfig {
|
||||
return {
|
||||
enabled: true,
|
||||
gateways: {
|
||||
gateway: {
|
||||
type: "http",
|
||||
url: "https://example.com",
|
||||
method: "POST",
|
||||
},
|
||||
},
|
||||
hooks: {},
|
||||
replyListener: {
|
||||
discordBotToken: "discord-token",
|
||||
discordChannelId: "channel-1",
|
||||
authorizedDiscordUserIds: ["user-1"],
|
||||
pollIntervalMs: 10,
|
||||
rateLimitPerMinute: 10,
|
||||
maxMessageLength: 500,
|
||||
includePrefix: true,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function getReplyListenerConfigSignature(config: OpenClawConfig): string {
|
||||
return JSON.stringify(config.replyListener ?? null)
|
||||
}
|
||||
|
||||
function resetStateDir(): void {
|
||||
rmSync(stateDir, { recursive: true, force: true })
|
||||
mkdirSync(stateDir, { recursive: true })
|
||||
livePids.clear()
|
||||
daemonPids.clear()
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
process.env.HOME = tempHome
|
||||
process.env.USERPROFILE = tempHome
|
||||
|
||||
mock.module("../reply-listener-spawn", () => ({
|
||||
spawnReplyListenerDaemon: (...args: unknown[]) => spawnImplementation(...args),
|
||||
}))
|
||||
|
||||
mock.module("../reply-listener-process", () => ({
|
||||
isReplyListenerProcessRunning: (pid: number) => livePids.has(pid),
|
||||
isReplyListenerDaemonProcess: async (pid: number) => daemonPids.has(pid),
|
||||
}))
|
||||
|
||||
mock.module("../tmux", () => ({
|
||||
isTmuxAvailable: async () => true,
|
||||
captureTmuxPane: async () => "",
|
||||
analyzePaneContent: () => ({ confidence: 1 }),
|
||||
sendToPane: async () => true,
|
||||
}))
|
||||
|
||||
replyListenerModule = await import("../reply-listener")
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
resetStateDir()
|
||||
process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS = "25"
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
if (originalHome === undefined) delete process.env.HOME
|
||||
else process.env.HOME = originalHome
|
||||
|
||||
if (originalUserProfile === undefined) delete process.env.USERPROFILE
|
||||
else process.env.USERPROFILE = originalUserProfile
|
||||
|
||||
if (originalStartupTimeout === undefined) {
|
||||
delete process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS
|
||||
} else {
|
||||
process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS = originalStartupTimeout
|
||||
}
|
||||
|
||||
rmSync(tempHome, { recursive: true, force: true })
|
||||
mock.restore()
|
||||
})
|
||||
|
||||
describe("startReplyListener", () => {
|
||||
test("returns the child's ready state only after detached startup reaches the poll loop", async () => {
|
||||
const killSpy = spyOn(process, "kill").mockImplementation((pid: number | string) => {
|
||||
if (pid === 4321) {
|
||||
return true
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
spawnImplementation = () => {
|
||||
const markReady = (): void => {
|
||||
if (!existsSync(stateFilePath)) {
|
||||
setTimeout(markReady, 5)
|
||||
return
|
||||
}
|
||||
|
||||
const pendingState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record<string, unknown>
|
||||
writeFileSync(
|
||||
stateFilePath,
|
||||
JSON.stringify(
|
||||
{
|
||||
...pendingState,
|
||||
isRunning: true,
|
||||
pid: 4321,
|
||||
lastPollAt: "2026-04-07T00:00:00.000Z",
|
||||
discordLastMessageId: "discord-99",
|
||||
messagesSeen: 4,
|
||||
},
|
||||
null,
|
||||
2,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
setTimeout(markReady, 5)
|
||||
|
||||
return {
|
||||
pid: 4321,
|
||||
unref() {
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
const result = await replyListenerModule.startReplyListener(createConfig())
|
||||
|
||||
try {
|
||||
expect(result.success).toBe(true)
|
||||
expect(result.state).toMatchObject({
|
||||
isRunning: true,
|
||||
pid: 4321,
|
||||
lastPollAt: "2026-04-07T00:00:00.000Z",
|
||||
discordLastMessageId: "discord-99",
|
||||
lastDiscordMessageId: "discord-99",
|
||||
messagesSeen: 4,
|
||||
})
|
||||
|
||||
const persistedState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record<string, unknown>
|
||||
expect(persistedState.messagesSeen).toBe(4)
|
||||
expect(persistedState.discordLastMessageId).toBe("discord-99")
|
||||
expect(persistedState.lastDiscordMessageId).toBe("discord-99")
|
||||
} finally {
|
||||
killSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
test("does not report success or leave stale running state when detached child never becomes ready", async () => {
|
||||
spawnImplementation = () => ({
|
||||
pid: 9876,
|
||||
unref() {
|
||||
},
|
||||
})
|
||||
|
||||
const result = await replyListenerModule.startReplyListener(createConfig())
|
||||
|
||||
expect(result.success).toBe(false)
|
||||
expect(result.message).toContain("ready")
|
||||
expect(existsSync(pidFilePath)).toBe(false)
|
||||
|
||||
if (existsSync(stateFilePath)) {
|
||||
const persistedState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record<string, unknown>
|
||||
expect(persistedState.isRunning).toBe(false)
|
||||
expect(persistedState.pid).toBeNull()
|
||||
}
|
||||
})
|
||||
|
||||
test("does not restart an already running daemon when persisted config already matches", async () => {
|
||||
const existingPid = 3210
|
||||
livePids.add(existingPid)
|
||||
daemonPids.add(existingPid)
|
||||
writeFileSync(pidFilePath, `${existingPid}`)
|
||||
writeFileSync(
|
||||
stateFilePath,
|
||||
JSON.stringify({ isRunning: true, pid: existingPid, startupToken: "existing", errors: 0 }, null, 2),
|
||||
)
|
||||
writeFileSync(configFilePath, JSON.stringify({ ...createConfig(), replyListener: { ...createConfig().replyListener, pollIntervalMs: 500 } }, null, 2))
|
||||
|
||||
let spawnCalls = 0
|
||||
spawnImplementation = () => {
|
||||
spawnCalls += 1
|
||||
return {
|
||||
pid: 9999,
|
||||
unref() {
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
const killSpy = spyOn(process, "kill").mockImplementation(() => true)
|
||||
|
||||
try {
|
||||
const result = await replyListenerModule.startReplyListener(createConfig())
|
||||
|
||||
expect(result.success).toBe(true)
|
||||
expect(result.message).toContain("already running")
|
||||
expect(spawnCalls).toBe(0)
|
||||
expect(killSpy).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
killSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
test("restarts an already running daemon when persisted reply-listener config is stale", async () => {
|
||||
const existingPid = 3210
|
||||
livePids.add(existingPid)
|
||||
daemonPids.add(existingPid)
|
||||
writeFileSync(pidFilePath, `${existingPid}`)
|
||||
writeFileSync(
|
||||
stateFilePath,
|
||||
JSON.stringify({ isRunning: true, pid: existingPid, startupToken: "existing", errors: 0 }, null, 2),
|
||||
)
|
||||
writeFileSync(
|
||||
configFilePath,
|
||||
JSON.stringify({
|
||||
...createConfig(),
|
||||
replyListener: {
|
||||
...createConfig().replyListener,
|
||||
discordChannelId: "stale-channel",
|
||||
authorizedDiscordUserIds: ["stale-user"],
|
||||
pollIntervalMs: 500,
|
||||
},
|
||||
}, null, 2),
|
||||
)
|
||||
|
||||
const killSpy = spyOn(process, "kill").mockImplementation((pid: number | string) => {
|
||||
if (typeof pid === "number") {
|
||||
livePids.delete(pid)
|
||||
daemonPids.delete(pid)
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
let spawnCalls = 0
|
||||
spawnImplementation = () => {
|
||||
spawnCalls += 1
|
||||
const nextPid = 4321
|
||||
livePids.add(nextPid)
|
||||
daemonPids.add(nextPid)
|
||||
|
||||
const markReady = (): void => {
|
||||
if (!existsSync(stateFilePath)) {
|
||||
setTimeout(markReady, 5)
|
||||
return
|
||||
}
|
||||
|
||||
const pendingState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record<string, unknown>
|
||||
writeFileSync(
|
||||
stateFilePath,
|
||||
JSON.stringify(
|
||||
{
|
||||
...pendingState,
|
||||
isRunning: true,
|
||||
pid: nextPid,
|
||||
lastPollAt: "2026-04-07T00:00:00.000Z",
|
||||
messagesSeen: 2,
|
||||
},
|
||||
null,
|
||||
2,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
setTimeout(markReady, 5)
|
||||
|
||||
return {
|
||||
pid: nextPid,
|
||||
unref() {
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await replyListenerModule.startReplyListener(createConfig())
|
||||
|
||||
expect(result.success).toBe(true)
|
||||
expect(spawnCalls).toBe(1)
|
||||
expect(killSpy).toHaveBeenCalledWith(existingPid, "SIGTERM")
|
||||
|
||||
const persistedConfig = JSON.parse(readFileSync(configFilePath, "utf-8")) as OpenClawConfig
|
||||
expect(persistedConfig.replyListener?.discordChannelId).toBe("channel-1")
|
||||
expect(persistedConfig.replyListener?.authorizedDiscordUserIds).toEqual(["user-1"])
|
||||
expect(persistedConfig.replyListener?.pollIntervalMs).toBe(500)
|
||||
} finally {
|
||||
killSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
test("restarts an already running daemon when runtime state config signature is stale even if persisted config matches", async () => {
|
||||
const existingPid = 3210
|
||||
const matchingConfig: OpenClawConfig = {
|
||||
...createConfig(),
|
||||
replyListener: {
|
||||
...createConfig().replyListener!,
|
||||
pollIntervalMs: 500,
|
||||
},
|
||||
}
|
||||
const baseConfig = matchingConfig
|
||||
const staleConfig: OpenClawConfig = {
|
||||
...baseConfig,
|
||||
replyListener: {
|
||||
...baseConfig.replyListener!,
|
||||
discordBotToken: "stale-token",
|
||||
},
|
||||
}
|
||||
|
||||
livePids.add(existingPid)
|
||||
daemonPids.add(existingPid)
|
||||
writeFileSync(pidFilePath, `${existingPid}`)
|
||||
writeFileSync(
|
||||
stateFilePath,
|
||||
JSON.stringify(
|
||||
{
|
||||
isRunning: true,
|
||||
pid: existingPid,
|
||||
startupToken: "existing",
|
||||
errors: 0,
|
||||
configSignature: getReplyListenerConfigSignature(staleConfig),
|
||||
},
|
||||
null,
|
||||
2,
|
||||
),
|
||||
)
|
||||
writeFileSync(configFilePath, JSON.stringify(matchingConfig, null, 2))
|
||||
|
||||
const killSpy = spyOn(process, "kill").mockImplementation((pid: number | string) => {
|
||||
if (typeof pid === "number") {
|
||||
livePids.delete(pid)
|
||||
daemonPids.delete(pid)
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
let spawnCalls = 0
|
||||
spawnImplementation = () => {
|
||||
spawnCalls += 1
|
||||
const nextPid = 4321
|
||||
livePids.add(nextPid)
|
||||
daemonPids.add(nextPid)
|
||||
|
||||
const markReady = (): void => {
|
||||
if (!existsSync(stateFilePath)) {
|
||||
setTimeout(markReady, 5)
|
||||
return
|
||||
}
|
||||
|
||||
const pendingState = JSON.parse(readFileSync(stateFilePath, "utf-8")) as Record<string, unknown>
|
||||
writeFileSync(
|
||||
stateFilePath,
|
||||
JSON.stringify(
|
||||
{
|
||||
...pendingState,
|
||||
isRunning: true,
|
||||
pid: nextPid,
|
||||
lastPollAt: "2026-04-07T00:00:00.000Z",
|
||||
messagesSeen: 1,
|
||||
},
|
||||
null,
|
||||
2,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
setTimeout(markReady, 5)
|
||||
|
||||
return {
|
||||
pid: nextPid,
|
||||
unref() {
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await replyListenerModule.startReplyListener(createConfig())
|
||||
|
||||
expect(result.success).toBe(true)
|
||||
expect(spawnCalls).toBe(1)
|
||||
expect(killSpy).toHaveBeenCalledWith(existingPid, "SIGTERM")
|
||||
} finally {
|
||||
killSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,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")
|
||||
})
|
||||
})
|
||||
@@ -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<string, unknown> | null {
|
||||
return typeof value === "object" && value !== null ? (value as Record<string, unknown>) : null
|
||||
}
|
||||
|
||||
function firstStringValue(record: Record<string, unknown>, 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<WakeResult, "messageId" | "platform" | "channelId" | "threadId"> {
|
||||
const record = asRecord(payload)
|
||||
if (!record) return {}
|
||||
|
||||
const nestedCandidates = [record, asRecord(record.data), asRecord(record.result), asRecord(record.message)]
|
||||
.filter((candidate): candidate is Record<string, unknown> => candidate !== null)
|
||||
|
||||
let bestMatch: Pick<WakeResult, "messageId" | "platform" | "channelId" | "threadId"> = {}
|
||||
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<WakeResult, "messageId" | "platform" | "channelId" | "threadId"> {
|
||||
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<WakeResult> {
|
||||
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<string, string | undefined>,
|
||||
): Promise<{ gateway: string; success: boolean; error?: string }> {
|
||||
): Promise<WakeResult> {
|
||||
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<typeof setTimeout> | undefined
|
||||
const timeoutPromise = new Promise<never>((_, 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,
|
||||
|
||||
@@ -132,10 +132,16 @@ export async function wakeOpenClaw(
|
||||
}
|
||||
|
||||
export async function initializeOpenClaw(config: OpenClawConfig): Promise<void> {
|
||||
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 }
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
import { lookupByMessageId } from "./session-registry"
|
||||
import { injectReplyIntoPane, ReplyListenerRateLimiter } from "./reply-listener-injection"
|
||||
import { logReplyListenerMessage } from "./reply-listener-log"
|
||||
import {
|
||||
recordSeenDiscordMessage,
|
||||
writeReplyListenerDaemonState,
|
||||
type ReplyListenerDaemonState,
|
||||
} from "./reply-listener-state"
|
||||
import type { OpenClawConfig } from "./types"
|
||||
|
||||
interface DiscordMessage {
|
||||
id: string
|
||||
content: string
|
||||
author: { id: string }
|
||||
message_reference?: { message_id?: string }
|
||||
}
|
||||
|
||||
let discordBackoffUntil = 0
|
||||
|
||||
export async function pollDiscordReplies(
|
||||
config: OpenClawConfig,
|
||||
state: ReplyListenerDaemonState,
|
||||
rateLimiter: ReplyListenerRateLimiter,
|
||||
): Promise<void> {
|
||||
const replyListener = config.replyListener
|
||||
if (!replyListener?.discordBotToken || !replyListener.discordChannelId) return
|
||||
if (!replyListener.authorizedDiscordUserIds || replyListener.authorizedDiscordUserIds.length === 0) {
|
||||
return
|
||||
}
|
||||
if (Date.now() < discordBackoffUntil) return
|
||||
|
||||
try {
|
||||
const after = state.discordLastMessageId
|
||||
? `?after=${state.discordLastMessageId}&limit=10`
|
||||
: "?limit=10"
|
||||
const url = `https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages${after}`
|
||||
|
||||
const controller = new AbortController()
|
||||
const timeout = setTimeout(() => controller.abort(), 10000)
|
||||
const response = await fetch(url, {
|
||||
method: "GET",
|
||||
headers: { Authorization: `Bot ${replyListener.discordBotToken}` },
|
||||
signal: controller.signal,
|
||||
})
|
||||
clearTimeout(timeout)
|
||||
|
||||
const remaining = response.headers.get("x-ratelimit-remaining")
|
||||
const reset = response.headers.get("x-ratelimit-reset")
|
||||
if (remaining !== null && Number.parseInt(remaining, 10) < 2) {
|
||||
const parsedReset = reset ? Number.parseFloat(reset) : Number.NaN
|
||||
const resetTime = Number.isFinite(parsedReset) ? parsedReset * 1000 : Date.now() + 10000
|
||||
discordBackoffUntil = resetTime
|
||||
logReplyListenerMessage(
|
||||
`WARN: Discord rate limit low (remaining: ${remaining}), backing off until ${new Date(resetTime).toISOString()}`,
|
||||
)
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
state.errors += 1
|
||||
state.lastError = `Discord API error: HTTP ${response.status}`
|
||||
logReplyListenerMessage(state.lastError)
|
||||
writeReplyListenerDaemonState(state)
|
||||
return
|
||||
}
|
||||
|
||||
const messages = await response.json()
|
||||
if (!Array.isArray(messages) || messages.length === 0) return
|
||||
|
||||
for (const message of [...messages as DiscordMessage[]].reverse()) {
|
||||
recordSeenDiscordMessage(state, message.id)
|
||||
writeReplyListenerDaemonState(state)
|
||||
|
||||
const replyToMessageId = message.message_reference?.message_id
|
||||
if (!replyToMessageId) continue
|
||||
if (!replyListener.authorizedDiscordUserIds.includes(message.author.id)) continue
|
||||
|
||||
const mapping = lookupByMessageId("discord-bot", replyToMessageId)
|
||||
if (!mapping) continue
|
||||
|
||||
if (!rateLimiter.canProceed()) {
|
||||
logReplyListenerMessage(`WARN: Rate limit exceeded, dropping Discord message ${message.id}`)
|
||||
state.errors += 1
|
||||
continue
|
||||
}
|
||||
|
||||
const success = await injectReplyIntoPane(mapping.tmuxPaneId, message.content, "discord", config)
|
||||
if (success) {
|
||||
state.messagesInjected += 1
|
||||
try {
|
||||
await fetch(
|
||||
`https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages/${message.id}/reactions/%E2%9C%85/@me`,
|
||||
{
|
||||
method: "PUT",
|
||||
headers: { Authorization: `Bot ${replyListener.discordBotToken}` },
|
||||
},
|
||||
)
|
||||
} catch {
|
||||
}
|
||||
} else {
|
||||
state.errors += 1
|
||||
}
|
||||
|
||||
writeReplyListenerDaemonState(state)
|
||||
}
|
||||
} catch (error) {
|
||||
state.errors += 1
|
||||
state.lastError = error instanceof Error ? error.message : String(error)
|
||||
logReplyListenerMessage(`Discord polling error: ${state.lastError}`)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
import { removeMessagesByPane } from "./session-registry"
|
||||
import { analyzePaneContent, captureTmuxPane, sendToPane } from "./tmux"
|
||||
import { logReplyListenerMessage } from "./reply-listener-log"
|
||||
import type { OpenClawConfig } from "./types"
|
||||
|
||||
export function sanitizeReplyInput(text: string): string {
|
||||
return text
|
||||
.replace(/[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]/g, "")
|
||||
.replace(/[\u200e\u200f\u202a-\u202e\u2066-\u2069]/g, "")
|
||||
.replace(/\r?\n/g, " ")
|
||||
.replace(/\\/g, "\\\\")
|
||||
.replace(/`/g, "\\`")
|
||||
.replace(/\$\(/g, "\\$(")
|
||||
.replace(/\$\{/g, "\\${")
|
||||
.trim()
|
||||
}
|
||||
|
||||
export class ReplyListenerRateLimiter {
|
||||
private readonly maxPerMinute: number
|
||||
private readonly timestamps: number[] = []
|
||||
private readonly windowMs = 60 * 1000
|
||||
|
||||
constructor(maxPerMinute: number) {
|
||||
this.maxPerMinute = maxPerMinute
|
||||
}
|
||||
|
||||
canProceed(): boolean {
|
||||
const now = Date.now()
|
||||
const recent = this.timestamps.filter((timestamp) => now - timestamp < this.windowMs)
|
||||
this.timestamps.length = 0
|
||||
this.timestamps.push(...recent)
|
||||
|
||||
if (this.timestamps.length >= this.maxPerMinute) {
|
||||
return false
|
||||
}
|
||||
|
||||
this.timestamps.push(now)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
export async function injectReplyIntoPane(
|
||||
paneId: string,
|
||||
text: string,
|
||||
platform: string,
|
||||
config: OpenClawConfig,
|
||||
): Promise<boolean> {
|
||||
const replyListener = config.replyListener
|
||||
const content = await captureTmuxPane(paneId, 15)
|
||||
const analysis = analyzePaneContent(content)
|
||||
|
||||
if (analysis.confidence < 0.3) {
|
||||
logReplyListenerMessage(
|
||||
`WARN: Pane ${paneId} does not appear to be running OpenCode CLI (confidence: ${analysis.confidence}). Skipping injection, removing stale mapping.`,
|
||||
)
|
||||
removeMessagesByPane(paneId)
|
||||
return false
|
||||
}
|
||||
|
||||
const prefix = replyListener?.includePrefix === false ? "" : `[reply:${platform}] `
|
||||
const sanitized = sanitizeReplyInput(prefix + text)
|
||||
const truncated = sanitized.slice(0, replyListener?.maxMessageLength ?? 500)
|
||||
const success = await sendToPane(paneId, truncated, true)
|
||||
|
||||
if (success) {
|
||||
logReplyListenerMessage(
|
||||
`Injected reply from ${platform} into pane ${paneId}: "${truncated.slice(0, 50)}${truncated.length > 50 ? "..." : ""}"`,
|
||||
)
|
||||
} else {
|
||||
logReplyListenerMessage(`ERROR: Failed to inject reply into pane ${paneId}`)
|
||||
}
|
||||
|
||||
return success
|
||||
}
|
||||
@@ -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 {
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
import { existsSync, mkdirSync } from "fs"
|
||||
import { homedir } from "os"
|
||||
import { join } from "path"
|
||||
|
||||
export const REPLY_LISTENER_SECURE_FILE_MODE = 0o600
|
||||
|
||||
function resolveReplyListenerHomeDir(): string {
|
||||
return process.env.HOME ?? process.env.USERPROFILE ?? homedir()
|
||||
}
|
||||
|
||||
export function getReplyListenerStateDir(): string {
|
||||
return join(resolveReplyListenerHomeDir(), ".omx", "state")
|
||||
}
|
||||
|
||||
export function getReplyListenerPidFilePath(): string {
|
||||
return join(getReplyListenerStateDir(), "reply-listener.pid")
|
||||
}
|
||||
|
||||
export function getReplyListenerStateFilePath(): string {
|
||||
return join(getReplyListenerStateDir(), "reply-listener-state.json")
|
||||
}
|
||||
|
||||
export function getReplyListenerConfigFilePath(): string {
|
||||
return join(getReplyListenerStateDir(), "reply-listener-config.json")
|
||||
}
|
||||
|
||||
export function getReplyListenerLogFilePath(): string {
|
||||
return join(getReplyListenerStateDir(), "reply-listener.log")
|
||||
}
|
||||
|
||||
export function ensureReplyListenerStateDir(): void {
|
||||
const stateDir = getReplyListenerStateDir()
|
||||
if (!existsSync(stateDir)) {
|
||||
mkdirSync(stateDir, { recursive: true, mode: 0o700 })
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,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<string, string>): Record<string, string> {
|
||||
const env: Record<string, string> = {}
|
||||
|
||||
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<boolean> {
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
}),
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import { randomUUID } from "crypto"
|
||||
import type { ReplyListenerDaemonState } from "./reply-listener-state"
|
||||
|
||||
const DEFAULT_REPLY_LISTENER_STARTUP_TIMEOUT_MS = 500
|
||||
const REPLY_LISTENER_READY_POLL_INTERVAL_MS = 10
|
||||
|
||||
interface WaitForReplyListenerReadyOptions {
|
||||
pid: number
|
||||
startupToken: string
|
||||
timeoutMs: number
|
||||
readState: () => ReplyListenerDaemonState | null
|
||||
sleep: (ms: number) => Promise<void>
|
||||
}
|
||||
|
||||
function isPositiveInteger(value: number): boolean {
|
||||
return Number.isInteger(value) && value > 0
|
||||
}
|
||||
|
||||
export function createReplyListenerStartupToken(): string {
|
||||
return randomUUID()
|
||||
}
|
||||
|
||||
export function getReplyListenerStartupTimeoutMs(): number {
|
||||
const raw = process.env.OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TIMEOUT_MS
|
||||
if (!raw) return DEFAULT_REPLY_LISTENER_STARTUP_TIMEOUT_MS
|
||||
|
||||
const parsed = Number.parseInt(raw, 10)
|
||||
return isPositiveInteger(parsed) ? parsed : DEFAULT_REPLY_LISTENER_STARTUP_TIMEOUT_MS
|
||||
}
|
||||
|
||||
function isReadyState(
|
||||
state: ReplyListenerDaemonState | null,
|
||||
pid: number,
|
||||
startupToken: string,
|
||||
): state is ReplyListenerDaemonState {
|
||||
return Boolean(
|
||||
state
|
||||
&& state.isRunning
|
||||
&& state.pid === pid
|
||||
&& state.startupToken === startupToken
|
||||
&& state.lastPollAt !== null,
|
||||
)
|
||||
}
|
||||
|
||||
export async function waitForReplyListenerReady(
|
||||
options: WaitForReplyListenerReadyOptions,
|
||||
): Promise<ReplyListenerDaemonState | null> {
|
||||
const deadline = Date.now() + options.timeoutMs
|
||||
|
||||
while (Date.now() <= deadline) {
|
||||
const state = options.readState()
|
||||
if (isReadyState(state, options.pid, options.startupToken)) {
|
||||
return state
|
||||
}
|
||||
|
||||
await options.sleep(REPLY_LISTENER_READY_POLL_INTERVAL_MS)
|
||||
}
|
||||
|
||||
return null
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
import { existsSync, readFileSync, unlinkSync } from "fs"
|
||||
import type { OpenClawConfig } from "./types"
|
||||
import { writeSecureReplyListenerFile } from "./reply-listener-log"
|
||||
import {
|
||||
getReplyListenerConfigFilePath,
|
||||
getReplyListenerPidFilePath,
|
||||
getReplyListenerStateFilePath,
|
||||
} from "./reply-listener-paths"
|
||||
|
||||
export const REPLY_LISTENER_STARTUP_TOKEN_ENV = "OMO_OPENCLAW_REPLY_LISTENER_STARTUP_TOKEN"
|
||||
|
||||
export interface ReplyListenerDaemonState {
|
||||
isRunning: boolean
|
||||
pid: number | null
|
||||
startedAt: string
|
||||
startupToken: string | null
|
||||
configSignature: string | null
|
||||
lastPollAt: string | null
|
||||
telegramLastUpdateId: number | null
|
||||
discordLastMessageId: string | null
|
||||
lastDiscordMessageId: string | null
|
||||
messagesSeen: number
|
||||
messagesInjected: number
|
||||
errors: number
|
||||
lastError?: string
|
||||
}
|
||||
|
||||
function createDefaultReplyListenerState(): ReplyListenerDaemonState {
|
||||
return {
|
||||
isRunning: false,
|
||||
pid: null,
|
||||
startedAt: new Date().toISOString(),
|
||||
startupToken: null,
|
||||
configSignature: null,
|
||||
lastPollAt: null,
|
||||
telegramLastUpdateId: null,
|
||||
discordLastMessageId: null,
|
||||
lastDiscordMessageId: null,
|
||||
messagesSeen: 0,
|
||||
messagesInjected: 0,
|
||||
errors: 0,
|
||||
}
|
||||
}
|
||||
|
||||
function isNumber(value: unknown): value is number {
|
||||
return typeof value === "number" && Number.isFinite(value)
|
||||
}
|
||||
|
||||
function normalizeReplyListenerState(raw: unknown): ReplyListenerDaemonState {
|
||||
const defaults = createDefaultReplyListenerState()
|
||||
|
||||
if (typeof raw !== "object" || raw === null) {
|
||||
return defaults
|
||||
}
|
||||
|
||||
const state = raw as Partial<ReplyListenerDaemonState>
|
||||
return {
|
||||
isRunning: state.isRunning === true,
|
||||
pid: isNumber(state.pid) ? state.pid : null,
|
||||
startedAt: typeof state.startedAt === "string" ? state.startedAt : defaults.startedAt,
|
||||
startupToken: typeof state.startupToken === "string" ? state.startupToken : null,
|
||||
configSignature: typeof state.configSignature === "string" ? state.configSignature : null,
|
||||
lastPollAt: typeof state.lastPollAt === "string" ? state.lastPollAt : null,
|
||||
telegramLastUpdateId: isNumber(state.telegramLastUpdateId) ? state.telegramLastUpdateId : null,
|
||||
discordLastMessageId: getDiscordMessageId(state),
|
||||
lastDiscordMessageId: getDiscordMessageId(state),
|
||||
messagesSeen: isNumber(state.messagesSeen) ? state.messagesSeen : 0,
|
||||
messagesInjected: isNumber(state.messagesInjected) ? state.messagesInjected : 0,
|
||||
errors: isNumber(state.errors) ? state.errors : 0,
|
||||
...(typeof state.lastError === "string" ? { lastError: state.lastError } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
function getDiscordMessageId(state: Partial<ReplyListenerDaemonState>): string | null {
|
||||
if (typeof state.lastDiscordMessageId === "string") {
|
||||
return state.lastDiscordMessageId
|
||||
}
|
||||
|
||||
if (typeof state.discordLastMessageId === "string") {
|
||||
return state.discordLastMessageId
|
||||
}
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
export function createPendingReplyListenerState(startupToken: string): ReplyListenerDaemonState {
|
||||
return {
|
||||
...createDefaultReplyListenerState(),
|
||||
startedAt: new Date().toISOString(),
|
||||
startupToken,
|
||||
}
|
||||
}
|
||||
|
||||
export function readReplyListenerDaemonState(): ReplyListenerDaemonState | null {
|
||||
try {
|
||||
const stateFilePath = getReplyListenerStateFilePath()
|
||||
if (!existsSync(stateFilePath)) return null
|
||||
return normalizeReplyListenerState(JSON.parse(readFileSync(stateFilePath, "utf-8")))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export function writeReplyListenerDaemonState(state: ReplyListenerDaemonState): void {
|
||||
writeSecureReplyListenerFile(
|
||||
getReplyListenerStateFilePath(),
|
||||
JSON.stringify(
|
||||
{
|
||||
...state,
|
||||
lastDiscordMessageId: state.lastDiscordMessageId ?? state.discordLastMessageId,
|
||||
discordLastMessageId: state.discordLastMessageId ?? state.lastDiscordMessageId,
|
||||
},
|
||||
null,
|
||||
2,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
export function readReplyListenerDaemonConfig(): OpenClawConfig | null {
|
||||
try {
|
||||
const configFilePath = getReplyListenerConfigFilePath()
|
||||
if (!existsSync(configFilePath)) return null
|
||||
return JSON.parse(readFileSync(configFilePath, "utf-8")) as OpenClawConfig
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export function writeReplyListenerDaemonConfig(config: OpenClawConfig): void {
|
||||
writeSecureReplyListenerFile(getReplyListenerConfigFilePath(), JSON.stringify(config, null, 2))
|
||||
}
|
||||
|
||||
export function readReplyListenerPid(): number | null {
|
||||
try {
|
||||
const pidFilePath = getReplyListenerPidFilePath()
|
||||
if (!existsSync(pidFilePath)) return null
|
||||
const pid = Number.parseInt(readFileSync(pidFilePath, "utf-8").trim(), 10)
|
||||
return Number.isNaN(pid) ? null : pid
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export function writeReplyListenerPid(pid: number): void {
|
||||
writeSecureReplyListenerFile(getReplyListenerPidFilePath(), String(pid))
|
||||
}
|
||||
|
||||
export function removeReplyListenerPid(): void {
|
||||
const pidFilePath = getReplyListenerPidFilePath()
|
||||
if (existsSync(pidFilePath)) {
|
||||
unlinkSync(pidFilePath)
|
||||
}
|
||||
}
|
||||
|
||||
export function getReplyListenerStartupTokenFromEnv(): string | null {
|
||||
const token = process.env[REPLY_LISTENER_STARTUP_TOKEN_ENV]
|
||||
return token && token.length > 0 ? token : null
|
||||
}
|
||||
|
||||
export function recordReplyListenerPoll(state: ReplyListenerDaemonState, pid: number): void {
|
||||
state.isRunning = true
|
||||
state.pid = pid
|
||||
state.lastPollAt = new Date().toISOString()
|
||||
}
|
||||
|
||||
export function recordSeenDiscordMessage(
|
||||
state: ReplyListenerDaemonState,
|
||||
messageId: string,
|
||||
): void {
|
||||
state.discordLastMessageId = messageId
|
||||
state.lastDiscordMessageId = messageId
|
||||
state.messagesSeen += 1
|
||||
}
|
||||
|
||||
export function markReplyListenerStopped(
|
||||
state: ReplyListenerDaemonState | null,
|
||||
error?: string,
|
||||
): ReplyListenerDaemonState {
|
||||
const nextState = state ?? createDefaultReplyListenerState()
|
||||
nextState.isRunning = false
|
||||
nextState.pid = null
|
||||
nextState.startupToken = null
|
||||
if (error) {
|
||||
nextState.lastError = error
|
||||
}
|
||||
return nextState
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
import { lookupByMessageId } from "./session-registry"
|
||||
import { injectReplyIntoPane, ReplyListenerRateLimiter } from "./reply-listener-injection"
|
||||
import { logReplyListenerMessage } from "./reply-listener-log"
|
||||
import { writeReplyListenerDaemonState, type ReplyListenerDaemonState } from "./reply-listener-state"
|
||||
import type { OpenClawConfig } from "./types"
|
||||
|
||||
interface TelegramMessage {
|
||||
message_id?: number
|
||||
chat?: { id?: number | string }
|
||||
text?: string
|
||||
reply_to_message?: { message_id?: number }
|
||||
}
|
||||
|
||||
interface TelegramUpdate {
|
||||
update_id?: number
|
||||
message?: TelegramMessage
|
||||
}
|
||||
|
||||
function parseTelegramUpdatesResponse(body: unknown): TelegramUpdate[] {
|
||||
if (typeof body !== "object" || body === null) return []
|
||||
const result = (body as { result?: TelegramUpdate[] }).result
|
||||
return Array.isArray(result) ? result : []
|
||||
}
|
||||
|
||||
export async function pollTelegramReplies(
|
||||
config: OpenClawConfig,
|
||||
state: ReplyListenerDaemonState,
|
||||
rateLimiter: ReplyListenerRateLimiter,
|
||||
): Promise<void> {
|
||||
const replyListener = config.replyListener
|
||||
if (!replyListener?.telegramBotToken || !replyListener.telegramChatId) return
|
||||
|
||||
try {
|
||||
const offset = state.telegramLastUpdateId ? state.telegramLastUpdateId + 1 : 0
|
||||
const url = `https://api.telegram.org/bot${replyListener.telegramBotToken}/getUpdates?offset=${offset}&timeout=0`
|
||||
|
||||
const controller = new AbortController()
|
||||
const timeout = setTimeout(() => controller.abort(), 10000)
|
||||
const response = await fetch(url, { method: "GET", signal: controller.signal })
|
||||
clearTimeout(timeout)
|
||||
|
||||
if (!response.ok) {
|
||||
logReplyListenerMessage(`Telegram API error: HTTP ${response.status}`)
|
||||
return
|
||||
}
|
||||
|
||||
const updates = parseTelegramUpdatesResponse(await response.json())
|
||||
for (const update of updates) {
|
||||
const message = update.message
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeReplyListenerDaemonState(state)
|
||||
|
||||
if (!message?.reply_to_message?.message_id) continue
|
||||
if (String(message.chat?.id) !== replyListener.telegramChatId) continue
|
||||
if (!message.text) continue
|
||||
|
||||
const mapping = lookupByMessageId("telegram", String(message.reply_to_message.message_id))
|
||||
if (!mapping) continue
|
||||
|
||||
if (!rateLimiter.canProceed()) {
|
||||
logReplyListenerMessage(`WARN: Rate limit exceeded, dropping Telegram message ${message.message_id}`)
|
||||
state.errors += 1
|
||||
continue
|
||||
}
|
||||
|
||||
const success = await injectReplyIntoPane(mapping.tmuxPaneId, message.text, "telegram", config)
|
||||
if (success) {
|
||||
state.messagesInjected += 1
|
||||
try {
|
||||
await fetch(`https://api.telegram.org/bot${replyListener.telegramBotToken}/sendMessage`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
chat_id: replyListener.telegramChatId,
|
||||
text: "Injected into Codex CLI session.",
|
||||
reply_to_message_id: message.message_id,
|
||||
}),
|
||||
})
|
||||
} catch {
|
||||
}
|
||||
} else {
|
||||
state.errors += 1
|
||||
}
|
||||
|
||||
writeReplyListenerDaemonState(state)
|
||||
}
|
||||
} catch (error) {
|
||||
state.errors += 1
|
||||
state.lastError = error instanceof Error ? error.message : String(error)
|
||||
logReplyListenerMessage(`Telegram polling error: ${state.lastError}`)
|
||||
}
|
||||
}
|
||||
+232
-620
@@ -1,562 +1,118 @@
|
||||
import {
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
readFileSync,
|
||||
writeFileSync,
|
||||
unlinkSync,
|
||||
chmodSync,
|
||||
statSync,
|
||||
appendFileSync,
|
||||
renameSync,
|
||||
} from "fs"
|
||||
import { join, dirname } from "path"
|
||||
import { homedir } from "os"
|
||||
import { spawn } from "bun" // Use bun spawn
|
||||
import { captureTmuxPane, analyzePaneContent, sendToPane, isTmuxAvailable } from "./tmux"
|
||||
import { lookupByMessageId, removeMessagesByPane, pruneStale } from "./session-registry"
|
||||
import type { OpenClawConfig } from "./types"
|
||||
import { dirname, join } from "path"
|
||||
import { normalizeReplyListenerConfig } from "./config"
|
||||
import { pollDiscordReplies } from "./reply-listener-discord"
|
||||
import { ReplyListenerRateLimiter } from "./reply-listener-injection"
|
||||
import { logReplyListenerMessage } from "./reply-listener-log"
|
||||
import {
|
||||
isReplyListenerDaemonProcess,
|
||||
isReplyListenerProcessRunning,
|
||||
} from "./reply-listener-process"
|
||||
import { spawnReplyListenerDaemon } from "./reply-listener-spawn"
|
||||
import { ensureReplyListenerStateDir } from "./reply-listener-paths"
|
||||
import {
|
||||
createPendingReplyListenerState,
|
||||
getReplyListenerStartupTokenFromEnv,
|
||||
markReplyListenerStopped,
|
||||
readReplyListenerDaemonConfig,
|
||||
readReplyListenerDaemonState,
|
||||
readReplyListenerPid,
|
||||
recordReplyListenerPoll,
|
||||
removeReplyListenerPid,
|
||||
type ReplyListenerDaemonState,
|
||||
writeReplyListenerDaemonConfig,
|
||||
writeReplyListenerDaemonState,
|
||||
writeReplyListenerPid,
|
||||
} from "./reply-listener-state"
|
||||
import {
|
||||
createReplyListenerStartupToken,
|
||||
getReplyListenerStartupTimeoutMs,
|
||||
waitForReplyListenerReady,
|
||||
} from "./reply-listener-startup"
|
||||
import { pollTelegramReplies } from "./reply-listener-telegram"
|
||||
import { pruneStale } from "./session-registry"
|
||||
import { isTmuxAvailable } from "./tmux"
|
||||
import type { OpenClawConfig } from "./types"
|
||||
|
||||
const SECURE_FILE_MODE = 0o600
|
||||
const MAX_LOG_SIZE_BYTES = 1 * 1024 * 1024
|
||||
const DAEMON_ENV_ALLOWLIST = [
|
||||
"PATH",
|
||||
"HOME",
|
||||
"USERPROFILE",
|
||||
"USER",
|
||||
"USERNAME",
|
||||
"LOGNAME",
|
||||
"LANG",
|
||||
"LC_ALL",
|
||||
"LC_CTYPE",
|
||||
"TERM",
|
||||
"TMUX",
|
||||
"TMUX_PANE",
|
||||
"TMPDIR",
|
||||
"TMP",
|
||||
"TEMP",
|
||||
"XDG_RUNTIME_DIR",
|
||||
"XDG_DATA_HOME",
|
||||
"XDG_CONFIG_HOME",
|
||||
"SHELL",
|
||||
"NODE_ENV",
|
||||
"HTTP_PROXY",
|
||||
"HTTPS_PROXY",
|
||||
"http_proxy",
|
||||
"https_proxy",
|
||||
"NO_PROXY",
|
||||
"no_proxy",
|
||||
"SystemRoot",
|
||||
"SYSTEMROOT",
|
||||
"windir",
|
||||
"COMSPEC",
|
||||
]
|
||||
const PRUNE_INTERVAL_MS = 60 * 60 * 1000
|
||||
const REPLY_LISTENER_STOP_TIMEOUT_MS = 1_000
|
||||
|
||||
const DEFAULT_STATE_DIR = join(homedir(), ".omx", "state")
|
||||
const PID_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener.pid")
|
||||
const STATE_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener-state.json")
|
||||
const CONFIG_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener-config.json")
|
||||
const LOG_FILE_PATH = join(DEFAULT_STATE_DIR, "reply-listener.log")
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms))
|
||||
}
|
||||
|
||||
export const DAEMON_IDENTITY_MARKER = "--openclaw-reply-listener-daemon"
|
||||
async function terminateReplyListenerProcess(pid: number): Promise<void> {
|
||||
if (!isReplyListenerProcessRunning(pid)) return
|
||||
if (!(await isReplyListenerDaemonProcess(pid))) return
|
||||
|
||||
function createMinimalDaemonEnv(): Record<string, string> {
|
||||
const env: Record<string, string> = {}
|
||||
for (const key of DAEMON_ENV_ALLOWLIST) {
|
||||
if (process.env[key] !== undefined) {
|
||||
env[key] = process.env[key] as string
|
||||
try {
|
||||
process.kill(pid, "SIGTERM")
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
function hasReplyListenerCredentials(config: OpenClawConfig): boolean {
|
||||
return Boolean(config.replyListener?.discordBotToken || config.replyListener?.telegramBotToken)
|
||||
}
|
||||
|
||||
function getNormalizedReplyListenerConfig(config: OpenClawConfig): OpenClawConfig {
|
||||
return normalizeReplyListenerConfig(config)
|
||||
}
|
||||
|
||||
function getReplyListenerRuntimeSignature(config: Pick<OpenClawConfig, "replyListener"> | null): string {
|
||||
return JSON.stringify(config?.replyListener ?? null)
|
||||
}
|
||||
|
||||
async function waitForDaemonToStop(timeoutMs: number): Promise<boolean> {
|
||||
const deadline = Date.now() + timeoutMs
|
||||
|
||||
while (Date.now() <= deadline) {
|
||||
if (!(await isDaemonRunning())) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return env
|
||||
}
|
||||
|
||||
function ensureStateDir(): void {
|
||||
if (!existsSync(DEFAULT_STATE_DIR)) {
|
||||
mkdirSync(DEFAULT_STATE_DIR, { recursive: true, mode: 0o700 })
|
||||
}
|
||||
}
|
||||
|
||||
function writeSecureFile(filePath: string, content: string): void {
|
||||
ensureStateDir()
|
||||
writeFileSync(filePath, content, { mode: SECURE_FILE_MODE })
|
||||
try {
|
||||
chmodSync(filePath, SECURE_FILE_MODE)
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
function rotateLogIfNeeded(logPath: string): void {
|
||||
try {
|
||||
if (!existsSync(logPath)) return
|
||||
const stats = statSync(logPath)
|
||||
if (stats.size > MAX_LOG_SIZE_BYTES) {
|
||||
const backupPath = `${logPath}.old`
|
||||
if (existsSync(backupPath)) {
|
||||
unlinkSync(backupPath)
|
||||
}
|
||||
renameSync(logPath, backupPath)
|
||||
}
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
function log(message: string): void {
|
||||
try {
|
||||
ensureStateDir()
|
||||
rotateLogIfNeeded(LOG_FILE_PATH)
|
||||
const timestamp = new Date().toISOString()
|
||||
const logLine = `[${timestamp}] ${message}\n`
|
||||
appendFileSync(LOG_FILE_PATH, logLine, { mode: SECURE_FILE_MODE })
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
export function logReplyListenerMessage(message: string): void {
|
||||
log(message)
|
||||
}
|
||||
|
||||
interface DaemonState {
|
||||
isRunning: boolean
|
||||
pid: number | null
|
||||
startedAt: string
|
||||
lastPollAt: string | null
|
||||
telegramLastUpdateId: number | null
|
||||
discordLastMessageId: string | null
|
||||
messagesInjected: number
|
||||
errors: number
|
||||
lastError?: string
|
||||
}
|
||||
|
||||
interface TelegramMessage {
|
||||
message_id?: number
|
||||
chat?: { id?: number | string }
|
||||
text?: string
|
||||
reply_to_message?: { message_id?: number }
|
||||
}
|
||||
|
||||
interface TelegramUpdate {
|
||||
update_id?: number
|
||||
message?: TelegramMessage
|
||||
}
|
||||
|
||||
interface TelegramUpdatesResponse {
|
||||
result?: TelegramUpdate[]
|
||||
}
|
||||
|
||||
function parseTelegramUpdatesResponse(body: unknown): TelegramUpdate[] {
|
||||
if (typeof body !== "object" || body === null) {
|
||||
return []
|
||||
await sleep(10)
|
||||
}
|
||||
|
||||
const result = (body as TelegramUpdatesResponse).result
|
||||
return Array.isArray(result) ? result : []
|
||||
}
|
||||
|
||||
function readDaemonState(): DaemonState | null {
|
||||
try {
|
||||
if (!existsSync(STATE_FILE_PATH)) return null
|
||||
const content = readFileSync(STATE_FILE_PATH, "utf-8")
|
||||
return JSON.parse(content)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function writeDaemonState(state: DaemonState): void {
|
||||
writeSecureFile(STATE_FILE_PATH, JSON.stringify(state, null, 2))
|
||||
}
|
||||
|
||||
function readDaemonConfig(): OpenClawConfig | null {
|
||||
try {
|
||||
if (!existsSync(CONFIG_FILE_PATH)) return null
|
||||
const content = readFileSync(CONFIG_FILE_PATH, "utf-8")
|
||||
return JSON.parse(content)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function writeDaemonConfig(config: OpenClawConfig): void {
|
||||
writeSecureFile(CONFIG_FILE_PATH, JSON.stringify(config, null, 2))
|
||||
}
|
||||
|
||||
function readPidFile(): number | null {
|
||||
try {
|
||||
if (!existsSync(PID_FILE_PATH)) return null
|
||||
const content = readFileSync(PID_FILE_PATH, "utf-8")
|
||||
const pid = parseInt(content.trim(), 10)
|
||||
if (Number.isNaN(pid)) return null
|
||||
return pid
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function writePidFile(pid: number): void {
|
||||
writeSecureFile(PID_FILE_PATH, String(pid))
|
||||
}
|
||||
|
||||
function removePidFile(): void {
|
||||
if (existsSync(PID_FILE_PATH)) {
|
||||
unlinkSync(PID_FILE_PATH)
|
||||
}
|
||||
}
|
||||
|
||||
function isProcessRunning(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export async function isReplyListenerProcess(pid: number): Promise<boolean> {
|
||||
try {
|
||||
if (process.platform === "linux") {
|
||||
const cmdline = readFileSync(`/proc/${pid}/cmdline`, "utf-8")
|
||||
return cmdline.includes(DAEMON_IDENTITY_MARKER)
|
||||
}
|
||||
const proc = spawn(["ps", "-p", String(pid), "-o", "args="], {
|
||||
stdout: "pipe",
|
||||
stderr: "ignore",
|
||||
})
|
||||
const stdout = await new Response(proc.stdout).text()
|
||||
if (proc.exitCode !== 0) return false
|
||||
return stdout.includes(DAEMON_IDENTITY_MARKER)
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
return !(await isDaemonRunning())
|
||||
}
|
||||
|
||||
export async function isDaemonRunning(): Promise<boolean> {
|
||||
const pid = readPidFile()
|
||||
const pid = readReplyListenerPid()
|
||||
if (pid === null) return false
|
||||
if (!isProcessRunning(pid)) {
|
||||
removePidFile()
|
||||
if (!isReplyListenerProcessRunning(pid)) {
|
||||
removeReplyListenerPid()
|
||||
return false
|
||||
}
|
||||
if (!(await isReplyListenerProcess(pid))) {
|
||||
removePidFile()
|
||||
if (!(await isReplyListenerDaemonProcess(pid))) {
|
||||
removeReplyListenerPid()
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
export function sanitizeReplyInput(text: string): string {
|
||||
return text
|
||||
.replace(/[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]/g, "")
|
||||
.replace(/[\u200e\u200f\u202a-\u202e\u2066-\u2069]/g, "")
|
||||
.replace(/\r?\n/g, " ")
|
||||
.replace(/\\/g, "\\\\")
|
||||
.replace(/`/g, "\\`")
|
||||
.replace(/\$\(/g, "\\$(")
|
||||
.replace(/\$\{/g, "\\${")
|
||||
.trim()
|
||||
}
|
||||
|
||||
class RateLimiter {
|
||||
maxPerMinute: number
|
||||
timestamps: number[] = []
|
||||
windowMs = 60 * 1000
|
||||
|
||||
constructor(maxPerMinute: number) {
|
||||
this.maxPerMinute = maxPerMinute
|
||||
}
|
||||
|
||||
canProceed(): boolean {
|
||||
const now = Date.now()
|
||||
this.timestamps = this.timestamps.filter((t) => now - t < this.windowMs)
|
||||
if (this.timestamps.length >= this.maxPerMinute) return false
|
||||
this.timestamps.push(now)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
async function injectReply(
|
||||
paneId: string,
|
||||
text: string,
|
||||
platform: string,
|
||||
config: OpenClawConfig,
|
||||
): Promise<boolean> {
|
||||
const replyListener = config.replyListener
|
||||
const content = await captureTmuxPane(paneId, 15)
|
||||
const analysis = analyzePaneContent(content)
|
||||
|
||||
if (analysis.confidence < 0.3) { // Lower threshold for simple check
|
||||
log(
|
||||
`WARN: Pane ${paneId} does not appear to be running OpenCode CLI (confidence: ${analysis.confidence}). Skipping injection, removing stale mapping.`,
|
||||
)
|
||||
removeMessagesByPane(paneId)
|
||||
return false
|
||||
}
|
||||
|
||||
const prefix = replyListener?.includePrefix === false ? "" : `[reply:${platform}] `
|
||||
const sanitized = sanitizeReplyInput(prefix + text)
|
||||
const truncated = sanitized.slice(0, replyListener?.maxMessageLength ?? 500)
|
||||
const success = await sendToPane(paneId, truncated, true)
|
||||
|
||||
if (success) {
|
||||
log(
|
||||
`Injected reply from ${platform} into pane ${paneId}: "${truncated.slice(0, 50)}${truncated.length > 50 ? "..." : ""}"`,
|
||||
)
|
||||
} else {
|
||||
log(`ERROR: Failed to inject reply into pane ${paneId}`)
|
||||
}
|
||||
return success
|
||||
}
|
||||
|
||||
let discordBackoffUntil = 0
|
||||
|
||||
async function pollDiscord(
|
||||
config: OpenClawConfig,
|
||||
state: DaemonState,
|
||||
rateLimiter: RateLimiter,
|
||||
): Promise<void> {
|
||||
const replyListener = config.replyListener
|
||||
if (!replyListener?.discordBotToken || !replyListener.discordChannelId) return
|
||||
if (
|
||||
!replyListener.authorizedDiscordUserIds
|
||||
|| replyListener.authorizedDiscordUserIds.length === 0
|
||||
) {
|
||||
return
|
||||
}
|
||||
if (Date.now() < discordBackoffUntil) return
|
||||
|
||||
try {
|
||||
const after = state.discordLastMessageId
|
||||
? `?after=${state.discordLastMessageId}&limit=10`
|
||||
: "?limit=10"
|
||||
const url = `https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages${after}`
|
||||
|
||||
const controller = new AbortController()
|
||||
const timeout = setTimeout(() => controller.abort(), 10000)
|
||||
|
||||
const response = await fetch(url, {
|
||||
method: "GET",
|
||||
headers: { Authorization: `Bot ${replyListener.discordBotToken}` },
|
||||
signal: controller.signal,
|
||||
})
|
||||
|
||||
clearTimeout(timeout)
|
||||
|
||||
const remaining = response.headers.get("x-ratelimit-remaining")
|
||||
const reset = response.headers.get("x-ratelimit-reset")
|
||||
|
||||
if (remaining !== null && parseInt(remaining, 10) < 2) {
|
||||
const parsed = reset ? parseFloat(reset) : Number.NaN
|
||||
const resetTime = Number.isFinite(parsed) ? parsed * 1000 : Date.now() + 10000
|
||||
discordBackoffUntil = resetTime
|
||||
log(
|
||||
`WARN: Discord rate limit low (remaining: ${remaining}), backing off until ${new Date(resetTime).toISOString()}`,
|
||||
)
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
log(`Discord API error: HTTP ${response.status}`)
|
||||
return
|
||||
}
|
||||
|
||||
const messages = await response.json()
|
||||
if (!Array.isArray(messages) || messages.length === 0) return
|
||||
|
||||
const sorted = [...messages].reverse()
|
||||
|
||||
for (const msg of sorted) {
|
||||
if (!msg.message_reference?.message_id) {
|
||||
state.discordLastMessageId = msg.id
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
if (!replyListener.authorizedDiscordUserIds.includes(msg.author.id)) {
|
||||
state.discordLastMessageId = msg.id
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
const mapping = lookupByMessageId("discord-bot", msg.message_reference.message_id)
|
||||
if (!mapping) {
|
||||
state.discordLastMessageId = msg.id
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
if (!rateLimiter.canProceed()) {
|
||||
log(`WARN: Rate limit exceeded, dropping Discord message ${msg.id}`)
|
||||
state.discordLastMessageId = msg.id
|
||||
writeDaemonState(state)
|
||||
state.errors++
|
||||
continue
|
||||
}
|
||||
|
||||
state.discordLastMessageId = msg.id
|
||||
writeDaemonState(state)
|
||||
|
||||
const success = await injectReply(mapping.tmuxPaneId, msg.content, "discord", config)
|
||||
|
||||
if (success) {
|
||||
state.messagesInjected++
|
||||
// Add reaction
|
||||
try {
|
||||
await fetch(
|
||||
`https://discord.com/api/v10/channels/${replyListener.discordChannelId}/messages/${msg.id}/reactions/%E2%9C%85/@me`,
|
||||
{
|
||||
method: "PUT",
|
||||
headers: { Authorization: `Bot ${replyListener.discordBotToken}` },
|
||||
},
|
||||
)
|
||||
} catch {
|
||||
}
|
||||
} else {
|
||||
state.errors++
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
state.errors++
|
||||
state.lastError = error instanceof Error ? error.message : String(error)
|
||||
log(`Discord polling error: ${state.lastError}`)
|
||||
}
|
||||
}
|
||||
|
||||
async function pollTelegram(
|
||||
config: OpenClawConfig,
|
||||
state: DaemonState,
|
||||
rateLimiter: RateLimiter,
|
||||
): Promise<void> {
|
||||
const replyListener = config.replyListener
|
||||
if (!replyListener?.telegramBotToken || !replyListener.telegramChatId) return
|
||||
|
||||
try {
|
||||
const offset = state.telegramLastUpdateId ? state.telegramLastUpdateId + 1 : 0
|
||||
const url = `https://api.telegram.org/bot${replyListener.telegramBotToken}/getUpdates?offset=${offset}&timeout=0`
|
||||
|
||||
const controller = new AbortController()
|
||||
const timeout = setTimeout(() => controller.abort(), 10000)
|
||||
|
||||
const response = await fetch(url, {
|
||||
method: "GET",
|
||||
signal: controller.signal,
|
||||
})
|
||||
|
||||
clearTimeout(timeout)
|
||||
|
||||
if (!response.ok) {
|
||||
log(`Telegram API error: HTTP ${response.status}`)
|
||||
return
|
||||
}
|
||||
|
||||
const body = await response.json()
|
||||
const updates = parseTelegramUpdatesResponse(body)
|
||||
|
||||
for (const update of updates) {
|
||||
const msg = update.message
|
||||
if (!msg) {
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
if (msg.reply_to_message?.message_id === undefined) {
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
if (String(msg.chat?.id) !== replyListener.telegramChatId) {
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
const mapping = lookupByMessageId("telegram", String(msg.reply_to_message.message_id))
|
||||
if (!mapping) {
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
const text = msg.text || ""
|
||||
if (!text) {
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeDaemonState(state)
|
||||
continue
|
||||
}
|
||||
|
||||
if (!rateLimiter.canProceed()) {
|
||||
log(`WARN: Rate limit exceeded, dropping Telegram message ${msg.message_id}`)
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeDaemonState(state)
|
||||
state.errors++
|
||||
continue
|
||||
}
|
||||
|
||||
state.telegramLastUpdateId = update.update_id ?? state.telegramLastUpdateId
|
||||
writeDaemonState(state)
|
||||
|
||||
const success = await injectReply(mapping.tmuxPaneId, text, "telegram", config)
|
||||
|
||||
if (success) {
|
||||
state.messagesInjected++
|
||||
try {
|
||||
await fetch(
|
||||
`https://api.telegram.org/bot${replyListener.telegramBotToken}/sendMessage`,
|
||||
{
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
chat_id: replyListener.telegramChatId,
|
||||
text: "Injected into Codex CLI session.",
|
||||
reply_to_message_id: msg.message_id,
|
||||
}),
|
||||
},
|
||||
)
|
||||
} catch {
|
||||
// Ignore
|
||||
}
|
||||
} else {
|
||||
state.errors++
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
state.errors++
|
||||
state.lastError = error instanceof Error ? error.message : String(error)
|
||||
log(`Telegram polling error: ${state.lastError}`)
|
||||
}
|
||||
}
|
||||
|
||||
const PRUNE_INTERVAL_MS = 60 * 60 * 1000
|
||||
|
||||
export async function pollLoop(): Promise<void> {
|
||||
log("Reply listener daemon starting poll loop")
|
||||
const config = readDaemonConfig()
|
||||
logReplyListenerMessage("Reply listener daemon starting poll loop")
|
||||
|
||||
const config = readReplyListenerDaemonConfig()
|
||||
if (!config) {
|
||||
log("ERROR: No daemon config found, exiting")
|
||||
logReplyListenerMessage("ERROR: No daemon config found, exiting")
|
||||
process.exit(1)
|
||||
}
|
||||
|
||||
const state = readDaemonState() || {
|
||||
isRunning: true,
|
||||
pid: process.pid,
|
||||
startedAt: new Date().toISOString(),
|
||||
lastPollAt: null,
|
||||
telegramLastUpdateId: null,
|
||||
discordLastMessageId: null,
|
||||
messagesInjected: 0,
|
||||
errors: 0,
|
||||
const startupToken = getReplyListenerStartupTokenFromEnv()
|
||||
const state = readReplyListenerDaemonState() ?? createPendingReplyListenerState(startupToken ?? "")
|
||||
state.configSignature = getReplyListenerRuntimeSignature(config)
|
||||
if (startupToken) {
|
||||
state.startupToken = startupToken
|
||||
}
|
||||
|
||||
state.isRunning = true
|
||||
state.pid = process.pid
|
||||
|
||||
const rateLimiter = new RateLimiter(config.replyListener?.rateLimitPerMinute || 10)
|
||||
const rateLimiter = new ReplyListenerRateLimiter(config.replyListener?.rateLimitPerMinute || 10)
|
||||
let lastPruneAt = Date.now()
|
||||
|
||||
const shutdown = (): void => {
|
||||
log("Shutdown signal received")
|
||||
state.isRunning = false
|
||||
writeDaemonState(state)
|
||||
removePidFile()
|
||||
logReplyListenerMessage("Shutdown signal received")
|
||||
writeReplyListenerDaemonState(markReplyListenerStopped(state))
|
||||
removeReplyListenerPid()
|
||||
process.exit(0)
|
||||
}
|
||||
|
||||
@@ -565,51 +121,96 @@ export async function pollLoop(): Promise<void> {
|
||||
|
||||
try {
|
||||
pruneStale()
|
||||
log("Pruned stale registry entries")
|
||||
} catch (e) {
|
||||
log(`WARN: Failed to prune stale entries: ${e}`)
|
||||
logReplyListenerMessage("Pruned stale registry entries")
|
||||
} catch (error) {
|
||||
logReplyListenerMessage(
|
||||
`WARN: Failed to prune stale entries: ${error instanceof Error ? error.message : String(error)}`,
|
||||
)
|
||||
}
|
||||
|
||||
while (state.isRunning) {
|
||||
|
||||
while (state.isRunning || state.pid === null) {
|
||||
try {
|
||||
state.lastPollAt = new Date().toISOString()
|
||||
await pollDiscord(config, state, rateLimiter)
|
||||
await pollTelegram(config, state, rateLimiter)
|
||||
|
||||
recordReplyListenerPoll(state, process.pid)
|
||||
writeReplyListenerDaemonState(state)
|
||||
|
||||
await pollDiscordReplies(config, state, rateLimiter)
|
||||
await pollTelegramReplies(config, state, rateLimiter)
|
||||
|
||||
if (Date.now() - lastPruneAt > PRUNE_INTERVAL_MS) {
|
||||
try {
|
||||
pruneStale()
|
||||
lastPruneAt = Date.now()
|
||||
log("Pruned stale registry entries")
|
||||
} catch (e) {
|
||||
log(`WARN: Prune failed: ${e instanceof Error ? e.message : String(e)}`)
|
||||
logReplyListenerMessage("Pruned stale registry entries")
|
||||
} catch (error) {
|
||||
logReplyListenerMessage(
|
||||
`WARN: Prune failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
writeDaemonState(state)
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, config.replyListener?.pollIntervalMs || 3000),
|
||||
)
|
||||
await sleep(config.replyListener?.pollIntervalMs || 3000)
|
||||
} catch (error) {
|
||||
state.errors++
|
||||
state.errors += 1
|
||||
state.lastError = error instanceof Error ? error.message : String(error)
|
||||
log(`Poll error: ${state.lastError}`)
|
||||
writeDaemonState(state)
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, (config.replyListener?.pollIntervalMs || 3000) * 2),
|
||||
)
|
||||
logReplyListenerMessage(`Poll error: ${state.lastError}`)
|
||||
writeReplyListenerDaemonState(state)
|
||||
await sleep((config.replyListener?.pollIntervalMs || 3000) * 2)
|
||||
}
|
||||
}
|
||||
log("Poll loop ended")
|
||||
|
||||
logReplyListenerMessage("Poll loop ended")
|
||||
}
|
||||
|
||||
export async function startReplyListener(config: OpenClawConfig): Promise<{ success: boolean; message: string; state?: DaemonState; error?: string }> {
|
||||
if (await isDaemonRunning()) {
|
||||
const state = readDaemonState()
|
||||
function createStartFailureResult(
|
||||
message: string,
|
||||
state: ReplyListenerDaemonState,
|
||||
): { success: false; message: string; state: ReplyListenerDaemonState } {
|
||||
return {
|
||||
success: false,
|
||||
message,
|
||||
state,
|
||||
}
|
||||
}
|
||||
|
||||
export async function startReplyListener(
|
||||
config: OpenClawConfig,
|
||||
): Promise<{ success: boolean; message: string; state?: ReplyListenerDaemonState; error?: string }> {
|
||||
const normalizedConfig = getNormalizedReplyListenerConfig(config)
|
||||
const replyListener = normalizedConfig.replyListener
|
||||
if (!replyListener?.discordBotToken && !replyListener?.telegramBotToken) {
|
||||
return {
|
||||
success: true,
|
||||
message: "Reply listener daemon is already running",
|
||||
state: state || undefined,
|
||||
success: false,
|
||||
message: "No enabled reply listener platforms configured (missing bot tokens/channels)",
|
||||
}
|
||||
}
|
||||
|
||||
if (await isDaemonRunning()) {
|
||||
const state = readReplyListenerDaemonState()
|
||||
const runtimeSignature = state?.configSignature ?? getReplyListenerRuntimeSignature(readReplyListenerDaemonConfig())
|
||||
if (runtimeSignature === getReplyListenerRuntimeSignature(normalizedConfig)) {
|
||||
return {
|
||||
success: true,
|
||||
message: "Reply listener daemon is already running",
|
||||
state: state || undefined,
|
||||
}
|
||||
}
|
||||
|
||||
const stopResult = await stopReplyListener()
|
||||
if (!stopResult.success) {
|
||||
return {
|
||||
success: false,
|
||||
message: "Failed to restart reply listener daemon",
|
||||
state: stopResult.state,
|
||||
error: stopResult.error ?? stopResult.message,
|
||||
}
|
||||
}
|
||||
|
||||
if (!(await waitForDaemonToStop(REPLY_LISTENER_STOP_TIMEOUT_MS))) {
|
||||
return {
|
||||
success: false,
|
||||
message: "Timed out waiting for reply listener daemon to stop before restart",
|
||||
state: readReplyListenerDaemonState() || undefined,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -620,108 +221,117 @@ export async function startReplyListener(config: OpenClawConfig): Promise<{ succ
|
||||
}
|
||||
}
|
||||
|
||||
const normalizedConfig = normalizeReplyListenerConfig(config)
|
||||
const replyListener = normalizedConfig.replyListener
|
||||
if (!replyListener?.discordBotToken && !replyListener?.telegramBotToken) {
|
||||
return {
|
||||
success: false,
|
||||
message: "No enabled reply listener platforms configured (missing bot tokens/channels)",
|
||||
}
|
||||
}
|
||||
ensureReplyListenerStateDir()
|
||||
writeReplyListenerDaemonConfig(normalizedConfig)
|
||||
|
||||
writeDaemonConfig(normalizedConfig)
|
||||
ensureStateDir()
|
||||
const startupToken = createReplyListenerStartupToken()
|
||||
const pendingState = createPendingReplyListenerState(startupToken)
|
||||
pendingState.configSignature = getReplyListenerRuntimeSignature(normalizedConfig)
|
||||
writeReplyListenerDaemonState(pendingState)
|
||||
|
||||
const currentFile = import.meta.url
|
||||
const isTs = currentFile.endsWith(".ts")
|
||||
const daemonScript = isTs
|
||||
const daemonScript = currentFile.endsWith(".ts")
|
||||
? join(dirname(new URL(currentFile).pathname), "daemon.ts")
|
||||
: join(dirname(new URL(currentFile).pathname), "daemon.js")
|
||||
|
||||
try {
|
||||
const proc = spawn(["bun", "run", daemonScript, DAEMON_IDENTITY_MARKER], {
|
||||
detached: true,
|
||||
stdio: ["ignore", "ignore", "ignore"],
|
||||
cwd: process.cwd(),
|
||||
env: createMinimalDaemonEnv(),
|
||||
})
|
||||
|
||||
proc.unref()
|
||||
const pid = proc.pid
|
||||
|
||||
if (pid) {
|
||||
writePidFile(pid)
|
||||
const state: DaemonState = {
|
||||
isRunning: true,
|
||||
pid,
|
||||
startedAt: new Date().toISOString(),
|
||||
lastPollAt: null,
|
||||
telegramLastUpdateId: null,
|
||||
discordLastMessageId: null,
|
||||
messagesInjected: 0,
|
||||
errors: 0,
|
||||
}
|
||||
writeDaemonState(state)
|
||||
log(`Reply listener daemon started with PID ${pid}`)
|
||||
return {
|
||||
success: true,
|
||||
message: `Reply listener daemon started with PID ${pid}`,
|
||||
state,
|
||||
}
|
||||
const processInfo = spawnReplyListenerDaemon(daemonScript, startupToken)
|
||||
|
||||
processInfo.unref()
|
||||
|
||||
if (!processInfo.pid) {
|
||||
const stoppedState = markReplyListenerStopped(pendingState, "Failed to start daemon process")
|
||||
writeReplyListenerDaemonState(stoppedState)
|
||||
return createStartFailureResult("Failed to start daemon process", stoppedState)
|
||||
}
|
||||
|
||||
|
||||
writeReplyListenerPid(processInfo.pid)
|
||||
|
||||
const readyState = await waitForReplyListenerReady({
|
||||
pid: processInfo.pid,
|
||||
startupToken,
|
||||
timeoutMs: getReplyListenerStartupTimeoutMs(),
|
||||
readState: readReplyListenerDaemonState,
|
||||
sleep,
|
||||
})
|
||||
|
||||
if (!readyState) {
|
||||
await terminateReplyListenerProcess(processInfo.pid)
|
||||
removeReplyListenerPid()
|
||||
const stoppedState = markReplyListenerStopped(
|
||||
readReplyListenerDaemonState() ?? pendingState,
|
||||
`Reply listener daemon did not become ready within ${getReplyListenerStartupTimeoutMs()}ms`,
|
||||
)
|
||||
writeReplyListenerDaemonState(stoppedState)
|
||||
return createStartFailureResult(
|
||||
`Reply listener daemon did not become ready within ${getReplyListenerStartupTimeoutMs()}ms`,
|
||||
stoppedState,
|
||||
)
|
||||
}
|
||||
|
||||
writeReplyListenerDaemonState(readyState)
|
||||
logReplyListenerMessage(`Reply listener daemon started with PID ${processInfo.pid}`)
|
||||
return {
|
||||
success: false,
|
||||
message: "Failed to start daemon process",
|
||||
success: true,
|
||||
message: `Reply listener daemon started with PID ${processInfo.pid}`,
|
||||
state: readyState,
|
||||
}
|
||||
} catch (error) {
|
||||
const stoppedState = markReplyListenerStopped(
|
||||
readReplyListenerDaemonState() ?? pendingState,
|
||||
error instanceof Error ? error.message : String(error),
|
||||
)
|
||||
writeReplyListenerDaemonState(stoppedState)
|
||||
removeReplyListenerPid()
|
||||
return {
|
||||
success: false,
|
||||
message: "Failed to start daemon",
|
||||
state: stoppedState,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function stopReplyListener(): Promise<{ success: boolean; message: string; state?: DaemonState; error?: string }> {
|
||||
const pid = readPidFile()
|
||||
export async function stopReplyListener(): Promise<{
|
||||
success: boolean
|
||||
message: string
|
||||
state?: ReplyListenerDaemonState
|
||||
error?: string
|
||||
}> {
|
||||
const pid = readReplyListenerPid()
|
||||
if (pid === null) {
|
||||
return {
|
||||
success: true,
|
||||
message: "Reply listener daemon is not running",
|
||||
}
|
||||
}
|
||||
|
||||
if (!isProcessRunning(pid)) {
|
||||
removePidFile()
|
||||
|
||||
if (!isReplyListenerProcessRunning(pid)) {
|
||||
removeReplyListenerPid()
|
||||
return {
|
||||
success: true,
|
||||
message: "Reply listener daemon was not running (cleaned up stale PID file)",
|
||||
}
|
||||
}
|
||||
|
||||
if (!(await isReplyListenerProcess(pid))) {
|
||||
removePidFile()
|
||||
|
||||
if (!(await isReplyListenerDaemonProcess(pid))) {
|
||||
removeReplyListenerPid()
|
||||
return {
|
||||
success: false,
|
||||
message: `Refusing to kill PID ${pid}: process identity does not match the reply listener daemon (stale or reused PID - removed PID file)`,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
try {
|
||||
process.kill(pid, "SIGTERM")
|
||||
removePidFile()
|
||||
const state = readDaemonState()
|
||||
if (state) {
|
||||
state.isRunning = false
|
||||
state.pid = null
|
||||
writeDaemonState(state)
|
||||
}
|
||||
log(`Reply listener daemon stopped (PID ${pid})`)
|
||||
removeReplyListenerPid()
|
||||
const state = markReplyListenerStopped(readReplyListenerDaemonState())
|
||||
writeReplyListenerDaemonState(state)
|
||||
logReplyListenerMessage(`Reply listener daemon stopped (PID ${pid})`)
|
||||
return {
|
||||
success: true,
|
||||
message: `Reply listener daemon stopped (PID ${pid})`,
|
||||
state: state || undefined,
|
||||
state,
|
||||
}
|
||||
} catch (error) {
|
||||
return {
|
||||
@@ -731,3 +341,5 @@ export async function stopReplyListener(): Promise<{ success: boolean; message:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export { logReplyListenerMessage }
|
||||
|
||||
@@ -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<string, string> = {
|
||||
"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<WakeResult | null> {
|
||||
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
|
||||
}
|
||||
@@ -49,4 +49,8 @@ export interface WakeResult {
|
||||
success: boolean
|
||||
error?: string
|
||||
statusCode?: number
|
||||
messageId?: string
|
||||
platform?: string
|
||||
channelId?: string
|
||||
threadId?: string
|
||||
}
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<string, unknown> | undefined;
|
||||
const sessionID = info?.sessionID as string | undefined;
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string> {
|
||||
return "ok"
|
||||
},
|
||||
})
|
||||
|
||||
const syncSessionCreatedCallbacks: Array<
|
||||
((event: { sessionID: string; parentID: string; title: string }) => Promise<void>) | 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<string, string>()
|
||||
|
||||
const { createToolRegistry, trimToolsToCap } = await import("./tool-registry")
|
||||
const dispatchOpenClawEvent = spyOn(openclawRuntimeDispatch, "dispatchOpenClawEvent")
|
||||
|
||||
function createPluginConfig(overrides: Partial<OhMyOpenCodeConfig> = {}): 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<typeof createToolRegistry>[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<typeof createToolRegistry>[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<typeof createToolRegistry>[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<typeof createToolRegistry>[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<void> {
|
||||
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<typeof createToolRegistry>[0]["ctx"],
|
||||
pluginConfig: createPluginConfig({ openclaw: openclawConfig }),
|
||||
managers: {
|
||||
backgroundManager: {},
|
||||
tmuxSessionManager,
|
||||
skillMcpManager: {},
|
||||
} as Parameters<typeof createToolRegistry>[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",
|
||||
},
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
})
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
|
||||
@@ -19,13 +19,15 @@ function createMockSkill(name: string): LoadedSkill {
|
||||
}
|
||||
|
||||
async function waitForRefresh(predicate: () => boolean): Promise<void> {
|
||||
for (let attempt = 0; attempt < 20; attempt += 1) {
|
||||
for (let attempt = 0; attempt < 200; attempt += 1) {
|
||||
if (predicate()) {
|
||||
return
|
||||
}
|
||||
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 0))
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 10))
|
||||
}
|
||||
|
||||
throw new Error("Timed out waiting for async skill description refresh")
|
||||
}
|
||||
|
||||
describe("skill tool - async native skill description refresh", () => {
|
||||
|
||||
Reference in New Issue
Block a user