diff --git a/extensions/discord/src/channel-actions.test.ts b/extensions/discord/src/channel-actions.test.ts index e99a4ef1726..6c4dc07776e 100644 --- a/extensions/discord/src/channel-actions.test.ts +++ b/extensions/discord/src/channel-actions.test.ts @@ -113,6 +113,21 @@ describe("discordMessageActions", () => { expect(discovery?.schema).toBeUndefined(); }); + it.each(["read", "search"])("routes %s actions through gateway execution mode", (action) => { + expect(discordMessageActions.resolveExecutionMode?.({ action: action as never })).toBe( + "gateway", + ); + }); + + it.each(["send", "edit", "delete", "react", "pin", "poll"])( + "routes %s actions through local execution mode", + (action) => { + expect(discordMessageActions.resolveExecutionMode?.({ action: action as never })).toBe( + "local", + ); + }, + ); + it("extracts send targets for message and thread reply actions", () => { expect( discordMessageActions.extractToolSend?.({ diff --git a/extensions/discord/src/channel-actions.ts b/extensions/discord/src/channel-actions.ts index 6dcc0d82e5c..03e1da871a4 100644 --- a/extensions/discord/src/channel-actions.ts +++ b/extensions/discord/src/channel-actions.ts @@ -160,6 +160,8 @@ function describeDiscordMessageTool({ } export const discordMessageActions: ChannelMessageActionAdapter = { + resolveExecutionMode: ({ action }) => + action === "read" || action === "search" ? "gateway" : "local", describeMessageTool: describeDiscordMessageTool, extractToolSend: ({ args }) => { const action = normalizeOptionalString(args.action) ?? ""; diff --git a/extensions/discord/src/proxy-request-client.test.ts b/extensions/discord/src/proxy-request-client.test.ts new file mode 100644 index 00000000000..0f4928ff885 --- /dev/null +++ b/extensions/discord/src/proxy-request-client.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it, vi } from "vitest"; +import { createDiscordRequestClient, DISCORD_REST_TIMEOUT_MS } from "./proxy-request-client.js"; + +describe("createDiscordRequestClient", () => { + it("injects an abort timeout signal into fetch calls", async () => { + const fetchSpy = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => { + expect(init?.signal).toBeDefined(); + expect(init!.signal!.aborted).toBe(false); + return new Response(JSON.stringify([]), { status: 200 }); + }); + + const client = createDiscordRequestClient("Bot test-token", { + fetch: fetchSpy as never, + queueRequests: false, + }); + + await client.get("/channels/123/messages"); + expect(fetchSpy).toHaveBeenCalledTimes(1); + }); + + it( + "aborts hanging requests after the timeout", + async () => { + const fetchSpy = vi.fn( + (_input: string | URL | Request, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => { + reject(new DOMException("The operation was aborted.", "AbortError")); + }); + }), + ); + + const client = createDiscordRequestClient("Bot test-token", { + fetch: fetchSpy as never, + queueRequests: false, + }); + + await expect(client.get("/channels/123/messages")).rejects.toThrow(); + }, + DISCORD_REST_TIMEOUT_MS + 5_000, + ); + + it("always injects a timeout signal even without a caller signal", async () => { + let receivedSignal: AbortSignal | undefined; + + const fetchSpy = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => { + receivedSignal = init?.signal ?? undefined; + return new Response(JSON.stringify({}), { status: 200 }); + }); + + const client = createDiscordRequestClient("Bot test-token", { + fetch: fetchSpy as never, + queueRequests: false, + }); + + await client.get("/channels/123/messages"); + + expect(receivedSignal).toBeDefined(); + expect(receivedSignal!.aborted).toBe(false); + }); + + it("exports a reasonable timeout constant", () => { + expect(DISCORD_REST_TIMEOUT_MS).toBeGreaterThanOrEqual(5_000); + expect(DISCORD_REST_TIMEOUT_MS).toBeLessThanOrEqual(30_000); + }); +}); diff --git a/extensions/discord/src/proxy-request-client.ts b/extensions/discord/src/proxy-request-client.ts index e33dd84b085..782bfe9a3c2 100644 --- a/extensions/discord/src/proxy-request-client.ts +++ b/extensions/discord/src/proxy-request-client.ts @@ -3,6 +3,8 @@ import { FormData as UndiciFormData } from "undici"; export type ProxyRequestClientOptions = RequestClientOptions; +export const DISCORD_REST_TIMEOUT_MS = 15_000; + function toUndiciFormData(body: FormData): UndiciFormData { const converted = new UndiciFormData(); for (const [key, value] of body.entries()) { @@ -22,15 +24,17 @@ function toUndiciFormData(body: FormData): UndiciFormData { function wrapDiscordFetch(fetchImpl: NonNullable) { return (input: string | URL | Request, init?: RequestInit): Promise => { + const signal = AbortSignal.timeout(DISCORD_REST_TIMEOUT_MS); if (init?.body instanceof FormData) { // Carbon builds global FormData; undici-backed proxy fetch needs undici's // FormData class to preserve multipart boundaries. return fetchImpl(input, { ...init, + signal, body: toUndiciFormData(init.body) as unknown as BodyInit, }); } - return fetchImpl(input, init); + return fetchImpl(input, { ...init, signal }); }; } diff --git a/extensions/discord/src/send.messages.test.ts b/extensions/discord/src/send.messages.test.ts new file mode 100644 index 00000000000..ed44723e67a --- /dev/null +++ b/extensions/discord/src/send.messages.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it, vi } from "vitest"; + +const restMock = { + get: vi.fn(), +}; + +vi.mock("./send.shared.js", () => ({ + resolveDiscordRest: () => restMock, +})); + +const { readMessagesDiscord, searchMessagesDiscord } = await import("./send.messages.js"); + +describe("readMessagesDiscord", () => { + it("returns messages from the REST client", async () => { + const messages = [{ id: "1", content: "hello" }]; + restMock.get.mockResolvedValueOnce(messages); + + const result = await readMessagesDiscord("C1", { limit: 5 }, { cfg: {} as never }); + + expect(result).toEqual(messages); + expect(restMock.get).toHaveBeenCalledWith(expect.stringContaining("C1"), { limit: 5 }); + }); + + it("propagates REST errors", async () => { + restMock.get.mockRejectedValueOnce(new Error("Discord API error")); + + await expect(readMessagesDiscord("C1", {}, { cfg: {} as never })).rejects.toThrow( + "Discord API error", + ); + }); +}); + +describe("searchMessagesDiscord", () => { + it("returns search results from the REST client", async () => { + const results = { messages: [[{ id: "1" }]], total_results: 1 }; + restMock.get.mockResolvedValueOnce(results); + + const result = await searchMessagesDiscord( + { guildId: "G1", content: "test", limit: 1 }, + { cfg: {} as never }, + ); + + expect(result).toEqual(results); + }); + + it("propagates REST errors", async () => { + restMock.get.mockRejectedValueOnce(new Error("Discord API error")); + + await expect( + searchMessagesDiscord({ guildId: "G1", content: "test" }, { cfg: {} as never }), + ).rejects.toThrow("Discord API error"); + }); +}); diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index 53b1edc5a53..c95fd632f38 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -2824,6 +2824,34 @@ describe("dispatchReplyFromConfig", () => { expect(internalHookMocks.triggerInternalHook).not.toHaveBeenCalled(); }); + it("falls back to CommandTargetSessionKey for internal hook when SessionKey is empty", async () => { + setNoAbort(); + const cfg = emptyConfig; + const dispatcher = createDispatcher(); + const ctx = buildTestCtx({ + Provider: "discord", + Surface: "discord", + CommandBody: "hello", + MessageSid: "msg-99", + }); + (ctx as MsgContext).SessionKey = undefined; + (ctx as MsgContext).CommandTargetSessionKey = "agent:main:discord:guild:123"; + + const replyResolver = async () => ({ text: "reply" }) satisfies ReplyPayload; + await dispatchReplyFromConfig({ ctx, cfg, dispatcher, replyResolver }); + + expect(internalHookMocks.createInternalHookEvent).toHaveBeenCalledWith( + "message", + "received", + "agent:main:discord:guild:123", + expect.objectContaining({ + content: "hello", + messageId: "msg-99", + }), + ); + expect(internalHookMocks.triggerInternalHook).toHaveBeenCalledTimes(1); + }); + it("emits diagnostics when enabled", async () => { setNoAbort(); const cfg = { diagnostics: { enabled: true } } as OpenClawConfig; diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index c239fcdd316..0b913287788 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -292,7 +292,8 @@ export async function dispatchReplyFromConfig( const channel = normalizeLowercaseStringOrEmpty(ctx.Surface ?? ctx.Provider ?? "unknown"); const chatId = ctx.To ?? ctx.From; const messageId = ctx.MessageSid ?? ctx.MessageSidFirst ?? ctx.MessageSidLast; - const sessionKey = ctx.SessionKey; + const sessionKey = + normalizeOptionalString(ctx.SessionKey) ?? normalizeOptionalString(ctx.CommandTargetSessionKey); const startTime = diagnosticsEnabled ? Date.now() : 0; const canTrackSession = diagnosticsEnabled && Boolean(sessionKey);